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
6669 changed files with 255575 additions and 200122 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,8 +53,12 @@ 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'
@@ -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
@@ -34,14 +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"
@@ -63,6 +67,10 @@ jobs:
        if: steps.changed-files.outputs.any_changed == 'true'
        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'
        env:
@@ -154,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/**

@@ -162,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,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
@@ -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})")
@@ -26,7 +26,6 @@ from .rbac import migrate_dataset_permissions_to_rbac, migrate_member_roles_to_r
 from .retention import (
    archive_workflow_runs,
    archive_workflow_runs_plan,
-    backfill_workflow_run_archive_bundles,
    clean_expired_messages,
    clean_workflow_runs,
    cleanup_orphaned_draft_variables,
@@ -55,7 +54,6 @@ __all__ = [
    "archive_workflow_runs",
    "archive_workflow_runs_plan",
    "backfill_plugin_auto_upgrade",
-    "backfill_workflow_run_archive_bundles",
    "clean_expired_messages",
    "clean_workflow_runs",
    "cleanup_orphaned_draft_variables",
@@ -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)
@@ -56,16 +65,8 @@ def _parse_tenant_prefixes(prefixes: str | None) -> list[str]:
    return sorted(set(parsed))


-def _parse_comma_separated_ids(raw_ids: str | None, *, param_name: str) -> list[str] | None:
-    if raw_ids is None:
-        return None
-    parsed = sorted({raw_id.strip() for raw_id in raw_ids.split(",") if raw_id.strip()})
-    if not parsed:
-        raise click.BadParameter(f"{param_name} must not be empty")
-    return parsed
-
-
 def _get_archive_candidate_tenant_ids_by_prefix(
+    session: Session,
    prefix: str,
    *,
    start_from: datetime.datetime | None,
@@ -84,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)
@@ -111,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,
@@ -131,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,
@@ -152,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,
@@ -358,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,
@@ -373,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,
@@ -583,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
@@ -625,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}, "
@@ -648,82 +725,6 @@ def archive_workflow_runs(
    )


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

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

@@ -377,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,
@@ -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."
@@ -11,6 +11,7 @@ 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 (
@@ -27,6 +28,7 @@ from controllers.service_api.app.error import (
    ProviderModelCurrentlyNotSupportError,
    ProviderNotInitializeError,
    ProviderQuotaExceededError,
+    WorkflowVersionExecutionNotAllowedError,
 )
 from controllers.service_api.schema import (
    expect_user_json,
@@ -43,6 +45,7 @@ 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
@@ -56,6 +59,7 @@ from libs.helper import dump_response, to_timestamp
 from models.model import App, AppMode, EndUser
 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
@@ -396,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"
@@ -421,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",
@@ -442,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)
@@ -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:
@@ -27,6 +27,8 @@ from clients.agent_backend import (
    AgentBackendInternalEventType,
    AgentBackendRunClient,
    AgentBackendRunEventAdapter,
+    AgentBackendRunFailedError,
+    AgentBackendRunFailedInternalEvent,
    AgentBackendRunSucceededInternalEvent,
    AgentBackendStreamInternalEvent,
    extract_runtime_layer_specs,
@@ -57,6 +59,14 @@ from core.workflow.nodes.agent_v2.ask_human_resume import build_deferred_tool_re
 from extensions.ext_database import db
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
 from graphon.model_runtime.entities.message_entities import AssistantPromptMessage, PromptMessage, UserPromptMessage
+from graphon.model_runtime.errors.invoke import (
+    InvokeAuthorizationError,
+    InvokeBadRequestError,
+    InvokeConnectionError,
+    InvokeError,
+    InvokeRateLimitError,
+    InvokeServerUnavailableError,
+)
 from models.agent_config_entities import AgentSoulConfig
 from models.enums import CreatorUserRole
 from models.model import MessageAgentThought
@@ -71,6 +81,33 @@ class _DefaultSessionScopeSnapshotId:

 _DEFAULT_SESSION_SCOPE_SNAPSHOT_ID = _DefaultSessionScopeSnapshotId()

+_AGENT_BACKEND_INVOKE_ERROR_BY_REASON: Mapping[str, type[InvokeError]] = {
+    "InvokeAuthorizationError": InvokeAuthorizationError,
+    "InvokeBadRequestError": InvokeBadRequestError,
+    "CredentialsValidateFailedError": InvokeBadRequestError,
+    "InvokeConnectionError": InvokeConnectionError,
+    "InvokeRateLimitError": InvokeRateLimitError,
+    "InvokeServerUnavailableError": InvokeServerUnavailableError,
+}
+
+
+def _agent_backend_failure_to_exception(event: AgentBackendRunFailedInternalEvent) -> Exception:
+    err_cls = _AGENT_BACKEND_INVOKE_ERROR_BY_REASON.get(event.reason or "")
+    if err_cls is not None:
+        return err_cls(event.error)
+    message = event.error or "Agent backend run did not complete successfully."
+    return AgentBackendRunFailedError(
+        event.run_id,
+        {
+            "error": event.error,
+            "reason": event.reason,
+            "source_event_id": event.source_event_id,
+        },
+        message=message,
+        reason=event.reason,
+        source_event_id=event.source_event_id,
+    )
+

 def _prompt_messages_from_query(user_query: str | None) -> list[PromptMessage]:
    if not user_query:
@@ -412,12 +449,15 @@ class _AgentProcessRecorder:
    def _lookup_tool_thought(self, *, index: int, tool_call_id: str | None) -> str | None:
        if tool_call_id and tool_call_id in self._tool_by_call_id:
            return self._tool_by_call_id[tool_call_id]
+        if index < 0:
+            return None
        return self._tool_by_index.get(index)

    def _remember_tool_thought(
        self, *, index: int, tool_call_id: str | None, tool_name: str | None, thought_id: str
    ) -> None:
-        self._tool_by_index[index] = thought_id
+        if index >= 0:
+            self._tool_by_index[index] = thought_id
        if tool_call_id:
            self._tool_by_call_id[tool_call_id] = thought_id
        if tool_name:
@@ -433,6 +473,10 @@ class _AgentProcessRecorder:
        return None

    def _mark_tool_observed(self, thought_id: str) -> None:
+        self._tool_by_index = {index: value for index, value in self._tool_by_index.items() if value != thought_id}
+        self._tool_by_call_id = {
+            tool_call_id: value for tool_call_id, value in self._tool_by_call_id.items() if value != thought_id
+        }
        for open_thought_ids in self._open_tool_by_name.values():
            open_thought_ids.discard(thought_id)

@@ -530,7 +574,12 @@ def _event_index(data: dict[str, Any]) -> int:


 def _string_or_none(value: Any) -> str | None:
-    return value if isinstance(value, str) and value else None
+    if not isinstance(value, str):
+        return None
+    normalized = value.strip()
+    if not normalized or normalized.lower() in {"none", "null"}:
+        return None
+    return normalized


 def _json_or_text(value: Any) -> str | None:
@@ -652,8 +701,12 @@ class AgentAppRunner:
            return

        if not isinstance(terminal, AgentBackendRunSucceededInternalEvent):
-            error = getattr(terminal, "error", None) or "Agent backend run did not complete successfully."
-            raise AgentBackendError(str(error))
+            if isinstance(terminal, AgentBackendRunFailedInternalEvent):
+                reason = terminal.reason
+                if reason == "sandbox_expired":
+                    raise AgentBackendError("The agent session sandbox has expired. Please start a new conversation.")
+                raise _agent_backend_failure_to_exception(terminal)
+            raise AgentBackendError("Agent backend run did not complete successfully.")

        answer = self._terminal_output_to_answer(terminal.output)
        try:
@@ -205,6 +205,7 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
                target=self._generate_worker,
                kwargs={
                    "flask_app": current_app._get_current_object(),  # type: ignore
+                    "session": db.session(),
                    "context": context,
                    "application_generate_entity": application_generate_entity,
                    "queue_manager": queue_manager,
@@ -5,10 +5,11 @@ from typing import Any, Union, cast

 from pydantic import JsonValue

+from clients.agent_backend.errors import AgentBackendError
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.task_entities import AppBlockingResponse, AppStreamResponse
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from graphon.model_runtime.errors.invoke import InvokeError
+from graphon.model_runtime.errors.invoke import InvokeError, InvokeRateLimitError

 logger = logging.getLogger(__name__)

@@ -127,6 +128,8 @@ class AppGenerateResponseConverter[TBlockingResponse: AppBlockingResponse](ABC):
            },
            ModelCurrentlyNotSupportError: {"code": "model_currently_not_support", "status": 400},
            InvokeError: {"code": "completion_request_error", "status": 400},
+            AgentBackendError: {"code": "completion_request_error", "status": 400},
+            InvokeRateLimitError: {"code": "rate_limit_error", "status": 429},
        }

        # Determine the response based on the type of exception
@@ -4,6 +4,7 @@ import time
 from sqlalchemy import select
 from sqlalchemy.orm import Session

+from clients.agent_backend.errors import AgentBackendError
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import (
    AppGenerateEntity,
@@ -54,7 +55,7 @@ class BasedGenerateTaskPipeline[AppGenerateEntityT: AppGenerateEntity]:
        match e:
            case InvokeAuthorizationError():
                err = InvokeAuthorizationError("Incorrect API key provided")
-            case InvokeError() | ValueError():
+            case InvokeError() | ValueError() | AgentBackendError():
                err = e
            case _:
                description = getattr(e, "description", None)
@@ -420,6 +420,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline[EasyUIAppGenerat
        message.total_price = usage.total_price
        message.currency = usage.currency
        self._task_state.llm_result.usage.latency = message.provider_response_latency
+        self._task_state.metadata.usage = self._task_state.llm_result.usage
        message.message_metadata = self._task_state.metadata.model_dump_json()

        if trace_manager:
@@ -27,7 +27,7 @@ class ProviderCredentialsCache:
            try:
                cached_provider_credentials = cached_provider_credentials.decode("utf-8")
                cached_provider_credentials = json.loads(cached_provider_credentials)
-            except JSONDecodeError:
+            except (JSONDecodeError, UnicodeDecodeError):
                return None

            return dict(cached_provider_credentials)
@@ -24,7 +24,7 @@ class ProviderCredentialsCache(ABC):
            try:
                cached_credentials = cached_credentials.decode("utf-8")
                return dict(json.loads(cached_credentials))
-            except JSONDecodeError:
+            except (JSONDecodeError, UnicodeDecodeError):
                return None
        return None

@@ -30,7 +30,7 @@ class ToolParameterCache:
            try:
                cached_tool_parameter = cached_tool_parameter.decode("utf-8")
                cached_tool_parameter = json.loads(cached_tool_parameter)
-            except JSONDecodeError:
+            except (JSONDecodeError, UnicodeDecodeError):
                return None

            return dict(cached_tool_parameter)
@@ -276,6 +276,7 @@ class RequestRequestDownloadFile(BaseModel):
        "validation",
    ]
    file: RequestDownloadFileMapping
+    for_external: bool = True

    model_config = ConfigDict(extra="forbid")

@@ -103,16 +103,17 @@ class ApiTool(Tool):
            elif not isinstance(credentials["api_key_value"], str):
                raise ToolProviderCredentialValidationError("api_key_value must be a string")

+            api_key_value = credentials["api_key_value"]
            if "api_key_header_prefix" in credentials:
                api_key_header_prefix = credentials["api_key_header_prefix"]
-                if api_key_header_prefix == "basic" and credentials["api_key_value"]:
-                    credentials["api_key_value"] = f"Basic {credentials['api_key_value']}"
-                elif api_key_header_prefix == "bearer" and credentials["api_key_value"]:
-                    credentials["api_key_value"] = f"Bearer {credentials['api_key_value']}"
+                if api_key_header_prefix == "basic" and api_key_value:
+                    api_key_value = f"Basic {api_key_value}"
+                elif api_key_header_prefix == "bearer" and api_key_value:
+                    api_key_value = f"Bearer {api_key_value}"
                elif api_key_header_prefix == "custom":
                    pass

-            headers[api_key_header] = credentials["api_key_value"]
+            headers[api_key_header] = api_key_value

        elif credentials["auth_type"] == "api_key_query":
            # For query parameter authentication, we don't add anything to headers
@@ -4,6 +4,7 @@ import hmac
 import logging
 import os
 import time
+import urllib.parse
 from collections.abc import Generator
 from mimetypes import guess_extension, guess_type
 from uuid import uuid4
@@ -26,7 +27,7 @@ logger = logging.getLogger(__name__)
 class ToolFileManager:
    @staticmethod
    def _build_graph_file_reference(tool_file: ToolFile) -> File:
-        extension = guess_extension(tool_file.mimetype) or ".bin"
+        extension = resolve_extension(filename=tool_file.name, mimetype=tool_file.mimetype)
        return File(
            file_type=get_file_type_by_mime_type(tool_file.mimetype),
            transfer_method=FileTransferMethod.TOOL_FILE,
@@ -70,7 +71,7 @@ class ToolFileManager:
        mimetype: str,
        filename: str | None = None,
    ) -> ToolFile:
-        extension = guess_extension(mimetype) or ".bin"
+        extension = resolve_extension(filename=filename, mimetype=mimetype)
        unique_name = uuid4().hex
        unique_filename = f"{unique_name}{extension}"
        # default just as before
@@ -120,7 +121,8 @@ class ToolFileManager:
            or response.headers.get("Content-Type", "").split(";")[0].strip()
            or "application/octet-stream"
        )
-        extension = guess_extension(mimetype) or ".bin"
+        url_filename = os.path.basename(urllib.parse.urlparse(file_url).path)
+        extension = resolve_extension(filename=url_filename, mimetype=mimetype)
        unique_name = uuid4().hex
        filename = f"{unique_name}{extension}"
        filepath = f"tools/{tenant_id}/{filename}"
@@ -220,4 +222,11 @@ def _factory() -> ToolFileManager:
    return ToolFileManager()


+def resolve_extension(*, filename: str | None, mimetype: str) -> str:
+    filename_extension = os.path.splitext(filename or "")[1].lower()
+    if filename_extension:
+        return filename_extension
+    return guess_extension(mimetype) or ".bin"
+
+
 set_tool_file_manager_factory(_factory)
@@ -3,7 +3,6 @@ import re
 from collections.abc import Generator
 from datetime import date, datetime
 from decimal import Decimal
-from mimetypes import guess_extension
 from typing import Any
 from uuid import UUID

@@ -11,7 +10,7 @@ import numpy as np
 import pytz

 from core.tools.entities.tool_entities import ToolInvokeMessage
-from core.tools.tool_file_manager import ToolFileManager
+from core.tools.tool_file_manager import ToolFileManager, resolve_extension
 from core.workflow.file_reference import parse_file_reference
 from graphon.file import File, FileTransferMethod, FileType
 from libs.login import current_user
@@ -91,7 +90,8 @@ class ToolFileMessageTransformer:
                        conversation_id=conversation_id,
                    )

-                    url = f"/files/tools/{tool_file.id}{guess_extension(tool_file.mimetype) or '.png'}"
+                    extension = resolve_extension(filename=tool_file.name, mimetype=tool_file.mimetype)
+                    url = cls.get_tool_file_url(tool_file_id=tool_file.id, extension=extension)
                    meta = cls._with_tool_file_meta(
                        message.meta,
                        tool_file_id=str(tool_file.id),
@@ -136,7 +136,8 @@ class ToolFileMessageTransformer:
                    filename=filename,
                )

-                url = cls.get_tool_file_url(tool_file_id=tool_file.id, extension=guess_extension(tool_file.mimetype))
+                extension = resolve_extension(filename=tool_file.name, mimetype=tool_file.mimetype)
+                url = cls.get_tool_file_url(tool_file_id=tool_file.id, extension=extension)
                meta = cls._with_tool_file_meta(meta, tool_file_id=str(tool_file.id))

                # check if file is image
@@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Any, override
 from agenton.compositor import CompositorSessionSnapshot

 from clients.agent_backend import (
+    AgentBackendAgentMessageDeltaInternalEvent,
    AgentBackendDeferredToolCallInternalEvent,
    AgentBackendError,
    AgentBackendHTTPError,
@@ -481,6 +482,10 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                        if isinstance(internal_event, AgentBackendStreamInternalEvent):
                            self._record_stream_metadata(metadata, internal_event)
                        continue
+                    if internal_event.type == AgentBackendInternalEventType.AGENT_MESSAGE_DELTA:
+                        if isinstance(internal_event, AgentBackendAgentMessageDeltaInternalEvent):
+                            self._record_agent_message_delta_metadata(metadata, internal_event)
+                        continue
                    metadata["agent_backend"] = {
                        **dict(metadata.get("agent_backend") or {}),
                        "stream_event_count": stream_event_count,
@@ -734,6 +739,17 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                agent_backend["usage"] = dict(usage)
        metadata["agent_backend"] = agent_backend

+    @staticmethod
+    def _record_agent_message_delta_metadata(
+        metadata: dict[str, Any], event: AgentBackendAgentMessageDeltaInternalEvent
+    ) -> None:
+        agent_backend = dict(metadata.get("agent_backend") or {})
+        agent_backend["agent_message_delta_count"] = int(agent_backend.get("agent_message_delta_count") or 0) + 1
+        agent_backend["agent_message_delta_length"] = int(agent_backend.get("agent_message_delta_length") or 0) + len(
+            event.delta
+        )
+        metadata["agent_backend"] = agent_backend
+
    @classmethod
    @override
    def _extract_variable_selector_to_variable_mapping(
@@ -10,13 +10,13 @@ trustworthy metadata.

 from __future__ import annotations

-from mimetypes import guess_extension
 from uuid import UUID

 from sqlalchemy import select
 from sqlalchemy.exc import DataError, SQLAlchemyError

 from core.db.session_factory import session_factory
+from core.tools.tool_file_manager import resolve_extension
 from core.workflow.file_reference import build_file_reference
 from graphon.file import File, FileTransferMethod, get_file_type_by_mime_type
 from models.tools import ToolFile
@@ -46,7 +46,7 @@ def reback_tool_file_output(*, tenant_id: str, tool_file_id: str) -> File | None
        return None

    mime_type = tool_file.mimetype or ""
-    extension = guess_extension(mime_type) or ".bin"
+    extension = resolve_extension(filename=tool_file.name, mimetype=mime_type)
    return File(
        type=get_file_type_by_mime_type(mime_type),
        transfer_method=FileTransferMethod.TOOL_FILE,
@@ -35,10 +35,10 @@ if [[ "${MODE}" == "worker" ]]; then
  if [[ -z "${CELERY_QUEUES}" ]]; then
    if [[ "${EDITION}" == "CLOUD" ]]; then
      # Cloud edition: separate queues for dataset and trigger tasks
-      DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow_professional,workflow_team,workflow_sandbox,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention,workflow_based_app_execution"
+      DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,app_rbac,plugin,workflow_storage,conversation,workflow_professional,workflow_team,workflow_sandbox,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention,workflow_based_app_execution"
    else
      # Community edition (SELF_HOSTED): dataset, pipeline and workflow have separate queues
-      DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention,workflow_based_app_execution"
+      DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,app_rbac,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention,workflow_based_app_execution"
    fi
  else
    DEFAULT_QUEUES="${CELERY_QUEUES}"
@@ -155,8 +155,8 @@ def init_app(app: DifyApp) -> Celery:
        "tasks.trigger_processing_tasks",  # async trigger processing
        "tasks.generate_summary_index_task",  # summary index generation
        "tasks.regenerate_summary_index_task",  # summary index regeneration
+        "tasks.initialize_created_app_rbac_access_task",  # app access initialization
        "tasks.app_generate.resume_agent_app_task",  # ENG-635: Agent v2 chat ask_human resume
-        "tasks.workflow_run_archive_download_tasks",  # workflow-run archive download preparation
    ]
    day = dify_config.CELERY_BEAT_SCHEDULER_TIME

@@ -7,7 +7,6 @@ def init_app(app: DifyApp):
        archive_workflow_runs,
        archive_workflow_runs_plan,
        backfill_plugin_auto_upgrade,
-        backfill_workflow_run_archive_bundles,
        clean_expired_messages,
        clean_workflow_runs,
        cleanup_orphaned_draft_variables,
@@ -78,7 +77,6 @@ def init_app(app: DifyApp):
        install_rag_pipeline_plugins,
        archive_workflow_runs_plan,
        archive_workflow_runs,
-        backfill_workflow_run_archive_bundles,
        delete_archived_workflow_runs,
        restore_workflow_runs,
        clean_workflow_runs,
@@ -3,6 +3,7 @@
 from __future__ import annotations

 import mimetypes
+import os
 import uuid
 from collections.abc import Mapping, Sequence
 from typing import Any, Literal, NotRequired, TypedDict, assert_never, cast
@@ -285,7 +286,7 @@ def _build_from_remote_url(
        raise ValueError("Invalid file url")

    mime_type, filename, file_size = get_remote_file_info(url)
-    extension = mimetypes.guess_extension(mime_type) or ("." + filename.split(".")[-1] if "." in filename else ".bin")
+    extension = os.path.splitext(filename)[1].lower() or mimetypes.guess_extension(mime_type) or ".bin"
    detected_file_type = standardize_file_type(extension=extension, mime_type=mime_type)
    file_type = _resolve_file_type(
        detected_file_type=detected_file_type,
@@ -326,7 +327,12 @@ def _build_from_tool_file(
        if tool_file is None:
            raise ValueError(f"ToolFile {tool_file_id} not found")

-        extension = "." + tool_file.file_key.split(".")[-1] if "." in tool_file.file_key else ".bin"
+        extension = (
+            os.path.splitext(tool_file.name)[1].lower()
+            or mimetypes.guess_extension(tool_file.mimetype)
+            or os.path.splitext(tool_file.file_key)[1].lower()
+            or ".bin"
+        )
        detected_file_type = standardize_file_type(extension=extension, mime_type=tool_file.mimetype)
        file_type = _resolve_file_type(
            detected_file_type=detected_file_type,
@@ -1,9 +1,10 @@
 from __future__ import annotations

 from datetime import datetime
+from decimal import Decimal
 from uuid import uuid4

-from pydantic import Field, field_validator
+from pydantic import Field, computed_field, field_validator

 from core.entities.execution_extra_content import ExecutionExtraContentDomainModel
 from fields.base import ResponseModel
@@ -55,10 +56,19 @@ class MessageListItem(ResponseModel):
    created_at: int | None = None
    agent_thoughts: list[AgentThought]
    message_files: list[MessageFile]
+    message_tokens: int = 0
+    answer_tokens: int = 0
+    provider_response_latency: float = 0
+    total_price: Decimal | None = None
+    currency: str | None = None
    status: str
    error: str | None = None
    extra_contents: list[ExecutionExtraContentDomainModel]

+    @computed_field
+    def total_tokens(self) -> int:
+        return self.message_tokens + self.answer_tokens
+
    @field_validator("inputs", mode="before")
    @classmethod
    def _normalize_inputs(cls, value: JSONValueType) -> JSONValueType:
@@ -11,7 +11,6 @@ import hashlib
 import logging
 from collections.abc import Generator
 from typing import Any, cast
-from urllib.parse import quote

 import boto3
 import orjson
@@ -198,22 +197,13 @@ class ArchiveStorage:
        except ClientError as e:
            raise ArchiveStorageError(f"Failed to delete object '{key}': {e}")

-    def generate_presigned_url(
-        self,
-        key: str,
-        expires_in: int = 3600,
-        *,
-        filename: str | None = None,
-        content_type: str | None = None,
-    ) -> str:
+    def generate_presigned_url(self, key: str, expires_in: int = 3600) -> str:
        """
        Generate a pre-signed URL for downloading an object.

        Args:
            key: Object key (path) within the bucket
            expires_in: URL validity duration in seconds (default: 1 hour)
-            filename: Optional browser download filename
-            content_type: Optional response content type

        Returns:
            Pre-signed URL string.
@@ -221,15 +211,10 @@ class ArchiveStorage:
        Raises:
            ArchiveStorageError: If generation fails
        """
-        params = {"Bucket": self.bucket, "Key": key}
-        if filename:
-            params["ResponseContentDisposition"] = f"attachment; filename*=UTF-8''{quote(filename)}"
-        if content_type:
-            params["ResponseContentType"] = content_type
        try:
            return self.client.generate_presigned_url(
                ClientMethod="get_object",
-                Params=params,
+                Params={"Bucket": self.bucket, "Key": key},
                ExpiresIn=expires_in,
            )
        except ClientError as e:
@@ -22,7 +22,6 @@ logger = logging.getLogger(__name__)

 CSRF_WHITE_LIST = [
    re.compile(r"/console/api/apps/[a-f0-9-]+/workflows/draft"),
-    re.compile(r"/console/api/workflow-run-archives/downloads/[a-f0-9]+/file"),
 ]


@@ -1,61 +0,0 @@
-"""add workflow run archive bundle index table
-
-Revision ID: 7a1c2d9e4b60
-Revises: c3d4e5f6a7b8
-Create Date: 2026-06-25 15:00:00.000000
-
-"""
-
-import sqlalchemy as sa
-from alembic import op
-
-import models
-
-# revision identifiers, used by Alembic.
-revision = "7a1c2d9e4b60"
-down_revision = "c3d4e5f6a7b8"
-branch_labels = None
-depends_on = None
-
-
-def _uuid_column(name: str, **kwargs):
-    if op.get_bind().dialect.name == "postgresql":
-        kwargs.setdefault("server_default", sa.text("uuidv7()"))
-    return sa.Column(name, models.types.StringUUID(), **kwargs)
-
-
-def upgrade() -> None:
-    op.create_table(
-        "workflow_run_archive_bundles",
-        _uuid_column("id", nullable=False),
-        sa.Column("tenant_id", models.types.StringUUID(), nullable=False),
-        sa.Column("year", sa.Integer(), nullable=False),
-        sa.Column("month", sa.Integer(), nullable=False),
-        sa.Column("shard", sa.String(length=32), nullable=False),
-        sa.Column("bundle_id", sa.String(length=64), nullable=False),
-        sa.Column("workflow_run_count", sa.Integer(), nullable=False),
-        sa.Column("row_count", sa.BigInteger(), nullable=False),
-        sa.Column("archive_bytes", sa.BigInteger(), nullable=False),
-        sa.Column("archived_at", sa.DateTime(), nullable=False),
-        sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
-        sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
-        sa.PrimaryKeyConstraint("id", name="workflow_run_archive_bundle_pkey"),
-        sa.UniqueConstraint(
-            "tenant_id",
-            "year",
-            "month",
-            "shard",
-            "bundle_id",
-            name="workflow_run_archive_bundle_identity_uq",
-        ),
-    )
-    op.create_index(
-        "workflow_run_archive_bundle_tenant_month_idx",
-        "workflow_run_archive_bundles",
-        ["tenant_id", "year", "month"],
-    )
-
-
-def downgrade() -> None:
-    op.drop_index("workflow_run_archive_bundle_tenant_month_idx", table_name="workflow_run_archive_bundles")
-    op.drop_table("workflow_run_archive_bundles")
@@ -144,7 +144,6 @@ from .workflow import (
    WorkflowNodeExecutionTriggeredFrom,
    WorkflowPause,
    WorkflowRun,
-    WorkflowRunArchiveBundle,
    WorkflowType,
    resolve_workflow_kind,
 )
@@ -283,7 +282,6 @@ __all__ = [
    "WorkflowNodeExecutionTriggeredFrom",
    "WorkflowPause",
    "WorkflowRun",
-    "WorkflowRunArchiveBundle",
    "WorkflowRunTriggeredFrom",
    "WorkflowSchedulePlan",
    "WorkflowToolProvider",
@@ -209,7 +209,7 @@ class Dataset(Base):
    pipeline_id = mapped_column(StringUUID, nullable=True)
    chunk_structure = mapped_column(sa.String(255), nullable=True)
    enable_api = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("true"))
-    is_multimodal = mapped_column(sa.Boolean, default=False, nullable=False, server_default=db.text("false"))
+    is_multimodal = mapped_column(sa.Boolean, default=False, nullable=False, server_default=sa.text("false"))

    @property
    def total_documents(self):
@@ -1446,40 +1446,6 @@ class WorkflowArchiveLog(TypeBase):
        }


-class WorkflowRunArchiveBundle(DefaultFieldsDCMixin, TypeBase):
-    """
-    Query index for one immutable V2 workflow-run archive bundle.
-
-    R2 manifest objects remain the recoverable archive source of truth. This table stores the small subset needed to
-    list tenant/month archives and locate bundles without listing object storage online. Missing rows can be rebuilt
-    from existing manifests by a backfill/reconciliation command.
-    """
-
-    __tablename__ = "workflow_run_archive_bundles"
-    __table_args__ = (
-        sa.PrimaryKeyConstraint("id", name="workflow_run_archive_bundle_pkey"),
-        sa.UniqueConstraint(
-            "tenant_id",
-            "year",
-            "month",
-            "shard",
-            "bundle_id",
-            name="workflow_run_archive_bundle_identity_uq",
-        ),
-        sa.Index("workflow_run_archive_bundle_tenant_month_idx", "tenant_id", "year", "month"),
-    )
-
-    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    year: Mapped[int] = mapped_column(sa.Integer, nullable=False)
-    month: Mapped[int] = mapped_column(sa.Integer, nullable=False)
-    shard: Mapped[str] = mapped_column(String(32), nullable=False)
-    bundle_id: Mapped[str] = mapped_column(String(64), nullable=False)
-    workflow_run_count: Mapped[int] = mapped_column(sa.Integer, nullable=False)
-    row_count: Mapped[int] = mapped_column(sa.BigInteger, nullable=False)
-    archive_bytes: Mapped[int] = mapped_column(sa.BigInteger, nullable=False)
-    archived_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
-
-
 class ConversationVariable(TypeBase):
    __tablename__ = "workflow_conversation_variables"

@@ -9690,61 +9690,6 @@ Suggest example workflow-generator instructions for the tenant
 | 200 | Suggestions generated successfully | **application/json**: [GeneratorResponse](#generatorresponse)<br> |
 | 400 | Invalid request parameters |  |

-### [GET] /workflow-run-archives
-List monthly workflow-run archive metadata for the current workspace
-
-#### Responses
-
-| Code | Description | Schema |
-| ---- | ----------- | ------ |
-| 200 | Success | **application/json**: [WorkflowRunArchiveListResponse](#workflowrunarchivelistresponse)<br> |
-
-### [POST] /workflow-run-archives/downloads
-Create or return a temporary workflow-run archive download task
-
-#### Request Body
-
-| Required | Schema |
-| -------- | ------ |
-|  Yes | **application/json**: [WorkflowRunArchiveDownloadPayload](#workflowrunarchivedownloadpayload)<br> |
-
-#### Responses
-
-| Code | Description | Schema |
-| ---- | ----------- | ------ |
-| 202 | Download task accepted | **application/json**: [WorkflowRunArchiveDownloadTaskResponse](#workflowrunarchivedownloadtaskresponse)<br> |
-
-### [GET] /workflow-run-archives/downloads/{download_id}
-Get a temporary workflow-run archive download task
-
-#### Parameters
-
-| Name | Located in | Description | Required | Schema |
-| ---- | ---------- | ----------- | -------- | ------ |
-| download_id | path |  | Yes | string |
-
-#### Responses
-
-| Code | Description | Schema |
-| ---- | ----------- | ------ |
-| 200 | Success | **application/json**: [WorkflowRunArchiveDownloadTaskResponse](#workflowrunarchivedownloadtaskresponse)<br> |
-
-### [GET] /workflow-run-archives/downloads/{download_id}/file
-Redirect to a prepared workflow-run archive ZIP file
-
-#### Parameters
-
-| Name | Located in | Description | Required | Schema |
-| ---- | ---------- | ----------- | -------- | ------ |
-| download_id | path |  | Yes | string |
-
-#### Responses
-
-| Code | Description | Schema |
-| ---- | ----------- | ------ |
-| 302 | Redirect to pre-signed archive storage URL | **application/json**: [RedirectResponse](#redirectresponse)<br> |
-| 409 | Download task is not ready |  |
-
 ### [GET] /workflow/{workflow_run_id}/events
 **Get workflow execution events stream after resume**

@@ -17669,19 +17614,25 @@ Built-in tool icons are URL strings; API-based tool icons are provider-defined p
 | ---- | ---- | ----------- | -------- |
 | agent_thoughts | [ [AgentThought](#agentthought) ] |  | Yes |
 | answer | string |  | Yes |
+| answer_tokens | integer |  | No |
 | conversation_id | string |  | Yes |
 | created_at | integer |  | No |
+| currency | string |  | No |
 | error | string |  | No |
 | extra_contents | [ [HumanInputContent](#humaninputcontent) ] |  | Yes |
 | feedback | [SimpleFeedback](#simplefeedback) |  | No |
 | id | string |  | Yes |
 | inputs | object |  | Yes |
 | message_files | [ [MessageFile](#messagefile) ] |  | Yes |
+| message_tokens | integer |  | No |
 | metadata | [JSONValueType](#jsonvaluetype) |  | No |
 | parent_message_id | string |  | No |
+| provider_response_latency | number |  | No |
 | query | string |  | Yes |
 | retriever_resources | [ [RetrieverResource](#retrieverresource) ] |  | Yes |
 | status | string |  | Yes |
+| total_price | string |  | No |
+| total_tokens | integer |  | Yes |

 #### ExternalApiTemplateListQuery

@@ -20771,6 +20722,7 @@ Whitelist scopes accepted by RBAC app and dataset access config APIs.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | access_policy_ids | [ string ] |  | No |
+| account_ids | [ string ] |  | No |

 #### ReplaceUserAccessPoliciesResponse

@@ -23349,71 +23301,6 @@ tenant's default model. The underlying generator never raises — an empty
 | result | string |  | Yes |
 | updated_at | integer |  | Yes |

-#### WorkflowRunArchiveDownloadPayload
-
-Request body for preparing one monthly workflow-run archive download.
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| month | integer |  | Yes |
-| year | integer |  | Yes |
-
-#### WorkflowRunArchiveDownloadStatus
-
-Lifecycle state for an asynchronous archive download request.
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| WorkflowRunArchiveDownloadStatus | string | Lifecycle state for an asynchronous archive download request. |  |
-
-#### WorkflowRunArchiveDownloadTaskResponse
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| archive_bytes | integer |  | Yes |
-| bundle_count | integer |  | Yes |
-| created_at | dateTime |  | Yes |
-| download_id | string |  | Yes |
-| error | string |  | No |
-| expires_at | dateTime |  | Yes |
-| file_name | string |  | No |
-| file_size_bytes | integer |  | No |
-| finished_at | string |  | No |
-| month | integer |  | Yes |
-| started_at | string |  | No |
-| status | [WorkflowRunArchiveDownloadStatus](#workflowrunarchivedownloadstatus) |  | Yes |
-| updated_at | dateTime |  | Yes |
-| year | integer |  | Yes |
-
-#### WorkflowRunArchiveListResponse
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| months | [ [WorkflowRunArchiveMonthResponse](#workflowrunarchivemonthresponse) ] |  | Yes |
-| summary | [WorkflowRunArchiveSummaryResponse](#workflowrunarchivesummaryresponse) |  | Yes |
-
-#### WorkflowRunArchiveMonthResponse
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| archive_bytes | integer |  | Yes |
-| bundle_count | integer |  | Yes |
-| download_task | [WorkflowRunArchiveDownloadTaskResponse](#workflowrunarchivedownloadtaskresponse) |  | No |
-| latest_archived_at | dateTime |  | Yes |
-| month | integer |  | Yes |
-| row_count | integer |  | Yes |
-| workflow_run_count | integer |  | Yes |
-| year | integer |  | Yes |
-
-#### WorkflowRunArchiveSummaryResponse
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| archive_bytes | integer |  | Yes |
-| archived_month_count | integer |  | Yes |
-| latest_archived_at | string |  | No |
-| workflow_run_count | integer |  | Yes |
-
 #### WorkflowRunCountQuery

 | Name | Type | Description | Required |
@@ -2235,7 +2235,7 @@ Execute a specific workflow version identified by its ID. Useful for running a p
 | 200 | Successful response. The content type and structure depend on the `response_mode` parameter in the request.  - If `response_mode` is `blocking`, returns `application/json` with a `WorkflowBlockingResponse` object. - If `response_mode` is `streaming`, returns `text/event-stream` with a stream of `ChunkWorkflowEvent` objects. | **application/json**: [GeneratedAppResponse](#generatedappresponse)<br>**text/event-stream**: [GeneratedAppResponse](#generatedappresponse)<br> |
 | 400 | - `not_workflow_app` : App mode does not match the API route. - `bad_request` : Workflow is a draft or has an invalid ID format. - `provider_not_initialize` : No valid model provider credentials found. - `provider_quota_exceeded` : Model provider quota exhausted. - `model_currently_not_support` : Current model unavailable. - `completion_request_error` : Workflow execution request failed. - `invalid_param` : Required parameter missing or invalid. |  |
 | 401 | Unauthorized - invalid API token |  |
-| 403 | Forbidden - token scope, app, dataset, or workspace access denied |  |
+| 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. - `rate_limit_error` : The upstream model provider rate limit was exceeded. |  |
 | 500 | `internal_server_error` : Internal server error. |  |
@@ -3467,18 +3467,24 @@ Model class for i18n object.
 | ---- | ---- | ----------- | -------- |
 | agent_thoughts | [ [AgentThought](#agentthought) ] |  | Yes |
 | answer | string |  | Yes |
+| answer_tokens | integer |  | No |
 | conversation_id | string |  | Yes |
 | created_at | integer |  | No |
+| currency | string |  | No |
 | error | string |  | No |
 | extra_contents | [ [HumanInputContent](#humaninputcontent) ] |  | Yes |
 | feedback | [SimpleFeedback](#simplefeedback) |  | No |
 | id | string |  | Yes |
 | inputs | object |  | Yes |
 | message_files | [ [MessageFile](#messagefile) ] |  | Yes |
+| message_tokens | integer |  | No |
 | parent_message_id | string |  | No |
+| provider_response_latency | number |  | No |
 | query | string |  | Yes |
 | retriever_resources | [ [RetrieverResource](#retrieverresource) ] |  | Yes |
 | status | string |  | Yes |
+| total_price | string |  | No |
+| total_tokens | integer |  | Yes |

 #### MessageListQuery

@@ -1685,19 +1685,25 @@ in form definiton, or a variable while the workflow is running.
 | ---- | ---- | ----------- | -------- |
 | agent_thoughts | [ [AgentThought](#agentthought) ] |  | Yes |
 | answer | string |  | Yes |
+| answer_tokens | integer |  | No |
 | conversation_id | string |  | Yes |
 | created_at | integer |  | No |
+| currency | string |  | No |
 | error | string |  | No |
 | extra_contents | [ [HumanInputContent](#humaninputcontent) ] |  | Yes |
 | feedback | [SimpleFeedback](#simplefeedback) |  | No |
 | id | string |  | Yes |
 | inputs | object |  | Yes |
 | message_files | [ [MessageFile](#messagefile) ] |  | Yes |
+| message_tokens | integer |  | No |
 | metadata | [JSONValueType](#jsonvaluetype) |  | No |
 | parent_message_id | string |  | No |
+| provider_response_latency | number |  | No |
 | query | string |  | Yes |
 | retriever_resources | [ [RetrieverResource](#retrieverresource) ] |  | Yes |
 | status | string |  | Yes |
+| total_price | string |  | No |
+| total_tokens | integer |  | Yes |

 #### WebModelConfigResponse

@@ -1,5 +1,6 @@
 from __future__ import annotations

+import logging
 from datetime import UTC, datetime
 from types import SimpleNamespace
 from typing import cast
@@ -259,14 +260,16 @@ def test_get_project_url_success(trace_instance: AliyunDataTrace):
    assert trace_instance.get_project_url() == "project-url"


-def test_get_project_url_error(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
+def test_get_project_url_error(
+    trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+):
    monkeypatch.setattr(trace_instance.trace_client, "get_project_url", MagicMock(side_effect=Exception("boom")))
-    logger_mock = MagicMock()
-    monkeypatch.setattr(aliyun_trace_module, "logger", logger_mock)

+    caplog.set_level(logging.INFO, logger=aliyun_trace_module.logger.name)
    with pytest.raises(ValueError, match=r"Aliyun get project url failed: boom"):
        trace_instance.get_project_url()
-    logger_mock.info.assert_called()
+
+    assert "Aliyun get project url failed: boom" in caplog.text


 def test_workflow_trace_adds_workflow_and_node_spans(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
@@ -2,6 +2,7 @@

 from __future__ import annotations

+import logging
 import sys
 import types
 from types import SimpleNamespace
@@ -87,7 +88,6 @@ class PatchedCoreComponents(TypedDict):
    tracer: MagicMock
    span: MagicMock
    tracer_provider: MagicMock
-    logger: MagicMock
    trace_api: Any


@@ -148,9 +148,6 @@ def patch_core_components(monkeypatch: pytest.MonkeyPatch) -> PatchedCoreCompone
    resource = MagicMock(name="resource")
    monkeypatch.setattr(client_module, "Resource", MagicMock(return_value=resource))

-    logger_mock = MagicMock(name="tencent_logger")
-    monkeypatch.setattr(client_module, "logger", logger_mock)
-
    trace_api_stub = SimpleNamespace(
        set_span_in_context=MagicMock(name="set_span_in_context", return_value="trace-context"),
        NonRecordingSpan=MagicMock(name="non_recording_span", side_effect=lambda ctx: f"non-{ctx}"),
@@ -174,7 +171,6 @@ def patch_core_components(monkeypatch: pytest.MonkeyPatch) -> PatchedCoreCompone
        "tracer": tracer,
        "span": span,
        "tracer_provider": tracer_provider,
-        "logger": logger_mock,
        "trace_api": trace_api_stub,
    }

@@ -268,14 +264,15 @@ def test_record_methods_skip_when_histogram_missing() -> None:
    client.record_trace_duration(0.5)


-def test_record_llm_duration_handles_exceptions(patch_core_components: PatchedCoreComponents) -> None:
+def test_record_llm_duration_handles_exceptions(caplog: pytest.LogCaptureFixture) -> None:
    client = _build_client()
    client.hist_llm_duration = MagicMock(name="hist_llm_duration")
    client.hist_llm_duration.record.side_effect = RuntimeError("boom")

+    caplog.set_level(logging.DEBUG, logger=client_module.logger.name)
    client.record_llm_duration(0.2)
-    logger = patch_core_components["logger"]
-    logger.debug.assert_called()
+
+    assert "[Tencent APM] Failed to record LLM duration" in caplog.text


 def test_create_and_export_span_sets_attributes(patch_core_components: PatchedCoreComponents) -> None:
@@ -328,12 +325,15 @@ def test_create_and_export_span_uses_parent_context(patch_core_components: Patch
    trace_api.set_span_in_context.assert_called_once()


-def test_create_and_export_span_exception_logs_error(patch_core_components: PatchedCoreComponents) -> None:
+def test_create_and_export_span_exception_logs_error(
+    patch_core_components: PatchedCoreComponents, caplog: pytest.LogCaptureFixture
+) -> None:
    client = _build_client()
    span = patch_core_components["span"]
    span.get_span_context.return_value = _make_span_context(span_id=2)
    client.tracer.start_span.side_effect = RuntimeError("boom")

+    caplog.set_level(logging.DEBUG, logger=client_module.logger.name)
    client._create_and_export_span(
        SpanData(
            trace_id=1,
@@ -346,8 +346,10 @@ def test_create_and_export_span_exception_logs_error(patch_core_components: Patc
            end_time=1,
        )
    )
-    logger = patch_core_components["logger"]
-    logger.exception.assert_called_once()
+
+    error_records = [record for record in caplog.records if record.levelno == logging.ERROR]
+    assert len(error_records) == 1
+    assert error_records[0].getMessage() == "[Tencent APM] Error creating span: span"


 def test_api_check_connects_successfully(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -423,23 +425,18 @@ def test_shutdown_flushes_all_components(patch_core_components: PatchedCoreCompo
    metric_reader.shutdown.assert_called_once()


-def test_shutdown_logs_when_meter_provider_fails(patch_core_components: PatchedCoreComponents) -> None:
+def test_shutdown_logs_when_meter_provider_fails(caplog: pytest.LogCaptureFixture) -> None:
    client = _build_client()
    meter_provider = meter_provider_instances[-1]
    meter_provider.shutdown.side_effect = RuntimeError("boom")
    assert client.metric_reader is not None
    client.metric_reader.shutdown.side_effect = RuntimeError("boom")

+    caplog.set_level(logging.DEBUG, logger=client_module.logger.name)
    client.shutdown()
-    logger = patch_core_components["logger"]
-    logger.debug.assert_any_call(
-        "[Tencent APM] Error shutting down meter provider",
-        exc_info=True,
-    )
-    logger.debug.assert_any_call(
-        "[Tencent APM] Error shutting down metric reader",
-        exc_info=True,
-    )
+
+    assert "[Tencent APM] Error shutting down meter provider" in caplog.text
+    assert "[Tencent APM] Error shutting down metric reader" in caplog.text


 def test_metrics_initialization_failure_sets_histogram_attributes(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -456,10 +453,11 @@ def test_metrics_initialization_failure_sets_histogram_attributes(monkeypatch: p
    assert client.metric_reader is None


-def test_add_span_logs_exception(monkeypatch: pytest.MonkeyPatch, patch_core_components: PatchedCoreComponents) -> None:
+def test_add_span_logs_exception(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
    client = _build_client()
    monkeypatch.setattr(client, "_create_and_export_span", MagicMock(side_effect=RuntimeError("boom")))

+    caplog.set_level(logging.DEBUG, logger=client_module.logger.name)
    client.add_span(
        SpanData(
            trace_id=1,
@@ -473,8 +471,9 @@ def test_add_span_logs_exception(monkeypatch: pytest.MonkeyPatch, patch_core_com
        )
    )

-    logger = patch_core_components["logger"]
-    logger.exception.assert_called_once()
+    error_records = [record for record in caplog.records if record.levelno == logging.ERROR]
+    assert len(error_records) == 1
+    assert error_records[0].getMessage() == "[Tencent APM] Failed to create span: span"


 def test_create_and_export_span_converts_attribute_types(patch_core_components: PatchedCoreComponents) -> None:
@@ -535,16 +534,20 @@ def test_record_trace_duration_converts_attributes() -> None:
    ],
 )
 def test_record_methods_handle_exceptions(
-    method: str, attr_name: str, args: tuple[object, ...], patch_core_components: PatchedCoreComponents
+    method: str, attr_name: str, args: tuple[object, ...], caplog: pytest.LogCaptureFixture
 ) -> None:
    client = _build_client()
    hist_mock = MagicMock(name=attr_name)
    hist_mock.record.side_effect = RuntimeError("boom")
    setattr(client, attr_name, hist_mock)

+    caplog.set_level(logging.DEBUG, logger=client_module.logger.name)
    getattr(client, method)(*args)
-    logger = patch_core_components["logger"]
-    logger.debug.assert_called()
+
+    assert any(
+        record.levelno == logging.DEBUG and record.getMessage().startswith("[Tencent APM] Failed to record")
+        for record in caplog.records
+    )


 def test_metrics_initializes_grpc_metric_exporter() -> None:
@@ -1,21 +1,21 @@
 [project]
 name = "dify-api"
-version = "1.15.0"
+version = "1.16.0-rc1"
 requires-python = "~=3.12.0"

 dependencies = [
    # Legacy: mature and widely deployed
    "bleach>=6.4.0,<7.0.0",
-    "boto3>=1.43.24,<2.0.0",
+    "boto3>=1.43.46,<2.0.0",
    "celery>=5.6.3,<6.0.0",
    "croniter>=6.2.2,<7.0.0",
    "dify-agent",
    "flask>=3.1.3,<4.0.0",
-    "flask-cors>=6.0.2,<7.0.0",
+    "flask-cors>=6.0.5,<7.0.0",
    "gevent>=26.4.0,<26.5.0",
    "gevent-websocket==0.10.1",
    "gmpy2>=2.3.0,<3.0.0",
-    "google-api-python-client>=2.196.0,<3.0.0",
+    "google-api-python-client>=2.198.0,<3.0.0",
    "gunicorn>=26.0.0,<27.0.0",
    "psycogreen>=1.0.2,<2.0.0",
    "psycopg2-binary>=2.9.12,<3.0.0",
@@ -31,7 +31,7 @@ dependencies = [
    "flask-migrate>=4.1.0,<5.0.0",
    "flask-orjson>=2.0.0,<3.0.0",
    "flask-restx>=1.3.2,<2.0.0",
-    "google-cloud-aiplatform>=1.151.0,<2.0.0",
+    "google-cloud-aiplatform>=1.160.0,<2.0.0",
    "httpx[socks]==0.28.1",
    "opentelemetry-distro==0.62b1",
    "opentelemetry-instrumentation-celery==0.62b1",
@@ -191,11 +191,11 @@ dev = [
 # Required for storage clients
 ############################################################
 storage = [
-    "azure-storage-blob>=12.29.0,<13.0.0",
-    "bce-python-sdk==0.9.71",
+    "azure-storage-blob>=12.30.0,<13.0.0",
+    "bce-python-sdk==0.9.72",
    "cos-python-sdk-v5>=1.9.44,<2.0.0",
-    "esdk-obs-python>=3.22.2,<4.0.0",
-    "google-cloud-storage>=3.11.0,<4.0.0",
+    "esdk-obs-python>=3.26.6,<4.0.0",
+    "google-cloud-storage>=3.12.1,<4.0.0",
    "opendal==0.46.0",
    "oss2>=2.19.1,<3.0.0",
    "supabase>=2.31.0,<3.0.0",
@@ -10,6 +10,7 @@ import json
 import logging
 import secrets
 import uuid
+from collections.abc import Iterator
 from datetime import UTC, datetime, timedelta
 from hashlib import sha256
 from typing import Any, NotRequired, TypedDict, cast
@@ -1563,6 +1564,25 @@ class TenantService:

        return updated_accounts

+    @staticmethod
+    def iter_member_account_id_batches(tenant_id: str, batch_size: int, *, session: Session) -> Iterator[list[str]]:
+        """Yield workspace member account ids in bounded, ordered batches."""
+        offset = 0
+        while True:
+            stmt = (
+                select(TenantAccountJoin.account_id)
+                .where(TenantAccountJoin.tenant_id == tenant_id)
+                .order_by(TenantAccountJoin.id)
+                .offset(offset)
+                .limit(batch_size)
+            )
+            account_ids = list(session.scalars(stmt).all())
+            if not account_ids:
+                return
+
+            yield account_ids
+            offset += batch_size
+
    @staticmethod
    def get_dataset_operator_members(tenant: Tenant, *, session: Session) -> list[Account]:
        """Get dataset admin members"""
@@ -6,6 +6,7 @@ with support for different subscription tiers, rate limiting, and execution trac
 """

 import json
+import logging
 from datetime import UTC, datetime
 from typing import Any

@@ -21,7 +22,7 @@ from models.model import App, EndUser
 from models.trigger import WorkflowTriggerLog, WorkflowTriggerLogDict
 from models.workflow import Workflow
 from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerLogRepository
-from services.errors.app import QuotaExceededError, WorkflowNotFoundError, WorkflowQuotaLimitError
+from services.errors.app import QuotaExceededError, WorkflowNotFoundError
 from services.quota_service import QuotaService, unlimited
 from services.workflow.entities import AsyncTriggerResponse, TriggerData, WorkflowTaskData
 from services.workflow.queue_dispatcher import QueueDispatcherManager, QueuePriority
@@ -32,6 +33,8 @@ from tasks.async_workflow_tasks import (
    execute_workflow_team,
 )

+logger = logging.getLogger(__name__)
+

 class AsyncWorkflowService:
    """
@@ -71,7 +74,7 @@ class AsyncWorkflowService:

        Raises:
            WorkflowNotFoundError: If app or workflow not found
-            InvokeDailyRateLimitError: If daily rate limit exceeded
+            QuotaExceededError: If workflow execution quota is exhausted

        Behavior:
            - Non-blocking: Returns immediately after queuing
@@ -145,10 +148,15 @@ class AsyncWorkflowService:
            trigger_log.error = f"Quota limit reached: {e}"
            trigger_log_repo.update(trigger_log)
            session.commit()
+            logger.info(
+                "Workflow quota exceeded for tenant %s, app %s, workflow %s, trigger log %s",
+                trigger_data.tenant_id,
+                trigger_data.app_id,
+                workflow.id,
+                trigger_log.id,
+            )

-            raise WorkflowQuotaLimitError(
-                f"Workflow execution quota limit reached for tenant {trigger_data.tenant_id}"
-            ) from e
+            raise

        # 8. Create task data
        queue_name = dispatcher.get_queue_name()
@@ -206,6 +214,7 @@ class AsyncWorkflowService:

        Raises:
            ValueError: If trigger log not found
+            QuotaExceededError: If workflow execution quota is exhausted

        Behavior:
            - Non-blocking: Returns immediately after queuing retry
@@ -255,6 +255,7 @@ class ResourceUserAccessPoliciesResponse(_RBACModel):

 class ReplaceUserAccessPolicies(_RBACModel):
    access_policy_ids: list[str] = Field(default_factory=list)
+    account_ids: list[str] = Field(default_factory=list)

    @field_validator("access_policy_ids", mode="before")
    @classmethod
@@ -1126,16 +1127,17 @@ class RBACService:
            tenant_id: str,
            account_id: str | None,
            app_id: str,
-            target_account_id: str,
+            target_account_id: str | None,
            payload: ReplaceUserAccessPolicies,
        ) -> ReplaceUserAccessPoliciesResponse:
+            request_data = payload.model_dump(mode="json")
            data = _inner_call(
                "PUT",
                f"{_INNER_PREFIX}/apps/user-access-policies",
                tenant_id=tenant_id,
                account_id=account_id,
                params={"app_id": app_id, "account_id": target_account_id},
-                json=payload.model_dump(mode="json"),
+                json=request_data,
            )
            return ReplaceUserAccessPoliciesResponse.model_validate(data or {})

@@ -1304,7 +1306,7 @@ class RBACService:
                tenant_id=tenant_id,
                account_id=account_id,
                params={"dataset_id": dataset_id, "account_id": target_account_id},
-                json=payload.model_dump(mode="json"),
+                json=payload.model_dump(mode="json", exclude_unset=True),
            )
            return ReplaceUserAccessPoliciesResponse.model_validate(data or {})

@@ -18,12 +18,6 @@ class WorkflowIdFormatError(Exception):
    pass


-class WorkflowQuotaLimitError(Exception):
-    """Raised when workflow execution quota is exceeded (for async/background workflows)."""
-
-    pass
-
-
 class QuotaExceededError(ValueError):
    """Raised when billing quota is exceeded for a feature."""

@@ -45,6 +45,7 @@ class FileRequestService:
        user_from: UserFrom | str,
        invoke_from: InvokeFrom | str,
        file_mapping: Mapping[str, Any],
+        for_external: bool = True,
    ) -> DownloadFileRequestResult:
        """Resolve one file mapping into signed download metadata.

@@ -61,7 +62,7 @@ class FileRequestService:
        )
        with bind_file_access_scope(scope):
            file = self._build_file(mapping=file_mapping, tenant_id=tenant_id)
-            download_url = file_helpers.resolve_file_url(file, for_external=True)
+            download_url = file_helpers.resolve_file_url(file, for_external=for_external)

        if not download_url:
            raise ValueError("file does not support signed download")
@@ -1,347 +0,0 @@
-"""
-Workflow-run archive bundle index helpers.
-
-Archive manifests in object storage remain the recoverable source of truth. This module mirrors their small query
-surface into `workflow_run_archive_bundles` so console listing and download jobs can avoid listing R2 on request.
-The backfill path is intentionally idempotent: every manifest is decoded, checked against the V2 schema markers, and
-upserted by immutable bundle identity.
-"""
-
-import datetime
-import json
-import logging
-import time
-from collections.abc import Sequence
-from dataclasses import dataclass, field
-from typing import TypedDict, cast
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
-
-from extensions.ext_database import db
-from libs.archive_storage import ArchiveStorage, get_archive_storage
-from models.workflow import WorkflowRunArchiveBundle
-from services.retention.workflow_run.constants import (
-    ARCHIVE_BUNDLE_FORMAT,
-    ARCHIVE_BUNDLE_MANIFEST_NAME,
-    ARCHIVE_BUNDLE_SCHEMA_VERSION,
-)
-
-logger = logging.getLogger(__name__)
-
-ARCHIVE_BUNDLE_ROOT_PREFIX = "workflow-runs/v2/"
-
-
-class ArchiveBundleTableManifestEntry(TypedDict):
-    row_count: int
-    checksum: str
-    size_bytes: int
-    object_key: str
-
-
-class ArchiveBundleManifest(TypedDict):
-    schema_version: str
-    archive_format: str
-    tenant_id: str
-    tenant_prefix: str
-    year: int
-    month: int
-    shard: str
-    bundle_id: str
-    object_prefix: str
-    workflow_run_count: int
-    workflow_node_execution_count: int
-    min_created_at: str
-    max_created_at: str
-    min_run_id: str
-    max_run_id: str
-    archived_at: str
-    tables: dict[str, ArchiveBundleTableManifestEntry]
-    run_ids: list[str]
-
-
-@dataclass(frozen=True)
-class ArchiveBundleIndexValues:
-    """Computed DB-index values derived from one manifest."""
-
-    row_count: int
-    archive_bytes: int
-    archived_at: datetime.datetime
-
-
-@dataclass
-class ArchiveBundleIndexBackfillSummary:
-    """Aggregate result for a manifest-to-DB-index reconciliation run."""
-
-    manifests_found: int = 0
-    bundles_processed: int = 0
-    bundles_upserted: int = 0
-    bundles_failed: int = 0
-    workflow_run_count: int = 0
-    row_count: int = 0
-    archive_bytes: int = 0
-    elapsed_time: float = 0.0
-    errors: list[str] = field(default_factory=list)
-
-
-def decode_archive_bundle_manifest(manifest_data: bytes) -> ArchiveBundleManifest:
-    """Decode raw manifest bytes into the V2 archive manifest shape."""
-    return cast(ArchiveBundleManifest, json.loads(manifest_data.decode("utf-8")))
-
-
-def parse_archive_manifest_datetime(value: str) -> datetime.datetime:
-    """Parse manifest datetimes and normalize timezone-aware values to naive UTC for DB storage."""
-    parsed = datetime.datetime.fromisoformat(value)
-    if parsed.tzinfo is None:
-        return parsed
-    return parsed.astimezone(datetime.UTC).replace(tzinfo=None)
-
-
-def calculate_archive_bundle_index_values(
-    manifest: ArchiveBundleManifest,
-    manifest_size_bytes: int,
-) -> ArchiveBundleIndexValues:
-    """Calculate row count, stored bytes, and archived timestamp for the DB index."""
-    _validate_archive_bundle_manifest(manifest)
-    row_count = sum(entry["row_count"] for entry in manifest["tables"].values())
-    archive_bytes = manifest_size_bytes + sum(entry["size_bytes"] for entry in manifest["tables"].values())
-    return ArchiveBundleIndexValues(
-        row_count=row_count,
-        archive_bytes=archive_bytes,
-        archived_at=parse_archive_manifest_datetime(manifest["archived_at"]),
-    )
-
-
-def upsert_archive_bundle_index_from_manifest(
-    session: Session,
-    manifest: ArchiveBundleManifest,
-    manifest_size_bytes: int,
-) -> WorkflowRunArchiveBundle:
-    """
-    Persist one archive manifest into `workflow_run_archive_bundles`.
-
-    The caller owns transaction boundaries. Re-running this function for the same manifest is safe and refreshes the
-    mutable metrics derived from object sizes and row counts.
-    """
-    values = calculate_archive_bundle_index_values(manifest, manifest_size_bytes)
-    existing = session.scalar(
-        select(WorkflowRunArchiveBundle).where(
-            WorkflowRunArchiveBundle.tenant_id == manifest["tenant_id"],
-            WorkflowRunArchiveBundle.year == manifest["year"],
-            WorkflowRunArchiveBundle.month == manifest["month"],
-            WorkflowRunArchiveBundle.shard == manifest["shard"],
-            WorkflowRunArchiveBundle.bundle_id == manifest["bundle_id"],
-        )
-    )
-    if existing is None:
-        bundle = WorkflowRunArchiveBundle(
-            tenant_id=manifest["tenant_id"],
-            year=manifest["year"],
-            month=manifest["month"],
-            shard=manifest["shard"],
-            bundle_id=manifest["bundle_id"],
-            workflow_run_count=manifest["workflow_run_count"],
-            row_count=values.row_count,
-            archive_bytes=values.archive_bytes,
-            archived_at=values.archived_at,
-        )
-        session.add(bundle)
-        return bundle
-
-    existing.workflow_run_count = manifest["workflow_run_count"]
-    existing.row_count = values.row_count
-    existing.archive_bytes = values.archive_bytes
-    existing.archived_at = values.archived_at
-    return existing
-
-
-class WorkflowRunArchiveBundleIndexBackfill:
-    """
-    Rebuild the DB bundle index by scanning object-store manifests.
-
-    Tenant IDs are the cheapest scope because they map directly to the object prefix. Tenant prefixes are supported for
-    rollout reconciliation, but they still require listing all tenants under that prefix and filtering keys locally.
-    """
-
-    storage: ArchiveStorage | None
-    session_factory: sessionmaker[Session]
-
-    def __init__(
-        self,
-        *,
-        storage: ArchiveStorage | None = None,
-        session_factory: sessionmaker[Session] | None = None,
-    ) -> None:
-        self.storage = storage
-        self.session_factory = session_factory or sessionmaker(bind=db.engine, expire_on_commit=False)
-
-    def run(
-        self,
-        *,
-        tenant_ids: Sequence[str] | None = None,
-        tenant_prefixes: Sequence[str] | None = None,
-        year: int | None = None,
-        month: int | None = None,
-        limit: int | None = None,
-        dry_run: bool = False,
-    ) -> ArchiveBundleIndexBackfillSummary:
-        """Scan matching manifest objects and idempotently upsert their DB index rows."""
-        start_time = time.time()
-        summary = ArchiveBundleIndexBackfillSummary()
-        storage = self.storage or get_archive_storage()
-        manifest_keys = self._list_manifest_keys(
-            storage,
-            tenant_ids=tenant_ids,
-            tenant_prefixes=tenant_prefixes,
-            year=year,
-            month=month,
-        )
-        summary.manifests_found = len(manifest_keys)
-
-        if limit is not None:
-            manifest_keys = manifest_keys[:limit]
-
-        for manifest_key in manifest_keys:
-            try:
-                manifest_data = storage.get_object(manifest_key)
-                manifest = decode_archive_bundle_manifest(manifest_data)
-                self._validate_manifest_scope(
-                    manifest,
-                    manifest_key=manifest_key,
-                    tenant_ids=tenant_ids,
-                    tenant_prefixes=tenant_prefixes,
-                    year=year,
-                    month=month,
-                )
-                values = calculate_archive_bundle_index_values(manifest, len(manifest_data))
-                summary.bundles_processed += 1
-                summary.workflow_run_count += manifest["workflow_run_count"]
-                summary.row_count += values.row_count
-                summary.archive_bytes += values.archive_bytes
-                if dry_run:
-                    continue
-
-                with self.session_factory() as session:
-                    upsert_archive_bundle_index_from_manifest(session, manifest, len(manifest_data))
-                    session.commit()
-                summary.bundles_upserted += 1
-            except Exception as exc:
-                logger.warning("Failed to backfill workflow archive bundle index from %s", manifest_key, exc_info=True)
-                summary.bundles_failed += 1
-                summary.errors.append(f"{manifest_key}: {exc}")
-
-        summary.elapsed_time = time.time() - start_time
-        return summary
-
-    @classmethod
-    def _list_manifest_keys(
-        cls,
-        storage: ArchiveStorage,
-        *,
-        tenant_ids: Sequence[str] | None,
-        tenant_prefixes: Sequence[str] | None,
-        year: int | None,
-        month: int | None,
-    ) -> list[str]:
-        prefixes = cls._list_prefixes(tenant_ids=tenant_ids, tenant_prefixes=tenant_prefixes, year=year, month=month)
-        keys: list[str] = []
-        for prefix in prefixes:
-            keys.extend(storage.list_objects(prefix))
-        return sorted(
-            key
-            for key in keys
-            if key.endswith(f"/{ARCHIVE_BUNDLE_MANIFEST_NAME}")
-            and cls._manifest_key_matches_scope(
-                key,
-                tenant_ids=tenant_ids,
-                tenant_prefixes=tenant_prefixes,
-                year=year,
-                month=month,
-            )
-        )
-
-    @staticmethod
-    def _list_prefixes(
-        *,
-        tenant_ids: Sequence[str] | None,
-        tenant_prefixes: Sequence[str] | None,
-        year: int | None,
-        month: int | None,
-    ) -> list[str]:
-        if tenant_ids:
-            prefixes = []
-            for tenant_id in sorted(set(tenant_ids)):
-                prefix = f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix={tenant_id[0].lower()}/tenant_id={tenant_id}/"
-                if year is not None:
-                    prefix += f"year={year:04d}/"
-                    if month is not None:
-                        prefix += f"month={month:02d}/"
-                prefixes.append(prefix)
-            return prefixes
-
-        if tenant_prefixes:
-            return [
-                f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix={tenant_prefix}/"
-                for tenant_prefix in sorted(set(tenant_prefixes))
-            ]
-
-        return [ARCHIVE_BUNDLE_ROOT_PREFIX]
-
-    @staticmethod
-    def _manifest_key_matches_scope(
-        key: str,
-        *,
-        tenant_ids: Sequence[str] | None,
-        tenant_prefixes: Sequence[str] | None,
-        year: int | None,
-        month: int | None,
-    ) -> bool:
-        if tenant_ids and _extract_key_part(key, "tenant_id") not in set(tenant_ids):
-            return False
-        if tenant_prefixes and _extract_key_part(key, "tenant_prefix") not in set(tenant_prefixes):
-            return False
-        if year is not None and _extract_key_part(key, "year") != f"{year:04d}":
-            return False
-        if month is not None and _extract_key_part(key, "month") != f"{month:02d}":
-            return False
-        return True
-
-    @staticmethod
-    def _validate_manifest_scope(
-        manifest: ArchiveBundleManifest,
-        *,
-        manifest_key: str,
-        tenant_ids: Sequence[str] | None,
-        tenant_prefixes: Sequence[str] | None,
-        year: int | None,
-        month: int | None,
-    ) -> None:
-        expected_object_prefix = manifest_key.removesuffix(f"/{ARCHIVE_BUNDLE_MANIFEST_NAME}")
-        if manifest["object_prefix"] != expected_object_prefix:
-            raise ValueError(
-                f"manifest object_prefix mismatch: expected={expected_object_prefix}, "
-                f"actual={manifest['object_prefix']}"
-            )
-        if tenant_ids and manifest["tenant_id"] not in tenant_ids:
-            raise ValueError(f"manifest tenant_id is outside requested scope: {manifest['tenant_id']}")
-        if tenant_prefixes and manifest["tenant_prefix"] not in tenant_prefixes:
-            raise ValueError(f"manifest tenant_prefix is outside requested scope: {manifest['tenant_prefix']}")
-        if year is not None and manifest["year"] != year:
-            raise ValueError(f"manifest year is outside requested scope: {manifest['year']}")
-        if month is not None and manifest["month"] != month:
-            raise ValueError(f"manifest month is outside requested scope: {manifest['month']}")
-
-
-def _validate_archive_bundle_manifest(manifest: ArchiveBundleManifest) -> None:
-    if manifest["schema_version"] != ARCHIVE_BUNDLE_SCHEMA_VERSION:
-        raise ValueError(f"unsupported archive bundle schema version: {manifest['schema_version']}")
-    if manifest["archive_format"] != ARCHIVE_BUNDLE_FORMAT:
-        raise ValueError(f"unsupported archive bundle format: {manifest['archive_format']}")
-
-
-def _extract_key_part(key: str, name: str) -> str | None:
-    prefix = f"{name}="
-    for part in key.split("/"):
-        if part.startswith(prefix):
-            return part[len(prefix) :]
-    return None
@@ -1,325 +0,0 @@
-"""
-Prepare monthly workflow-run archive downloads.
-
-Console requests create a short-lived Redis task and Celery runs this module in the background. The DB bundle index is
-the online lookup source: this preparer never lists archive storage, and it validates the indexed bundle set against the
-stable download id before packaging archive Parquet objects into one user-facing CSV ZIP file.
-"""
-
-import datetime
-import hashlib
-import io
-import logging
-import zipfile
-from collections.abc import Sequence
-from typing import cast
-
-import pyarrow.csv as pa_csv
-import pyarrow.parquet as pq
-from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
-
-from extensions.ext_database import db
-from libs.archive_storage import ArchiveStorage, get_archive_storage, get_export_storage
-from models.workflow import WorkflowRunArchiveBundle
-from services.retention.workflow_run.archive_bundle_index import (
-    ARCHIVE_BUNDLE_ROOT_PREFIX,
-    ArchiveBundleManifest,
-    ArchiveBundleTableManifestEntry,
-    decode_archive_bundle_manifest,
-)
-from services.retention.workflow_run.archive_download_task_cache import (
-    WorkflowRunArchiveDownloadStatus,
-    WorkflowRunArchiveDownloadTask,
-    WorkflowRunArchiveDownloadTaskCache,
-    build_archive_download_id,
-)
-from services.retention.workflow_run.constants import (
-    ARCHIVE_BUNDLE_FORMAT,
-    ARCHIVE_BUNDLE_MANIFEST_NAME,
-    ARCHIVE_BUNDLE_SCHEMA_VERSION,
-)
-
-logger = logging.getLogger(__name__)
-
-ARCHIVE_DOWNLOAD_ROOT_PREFIX = "workflow-runs/downloads/v1/"
-ARCHIVE_DOWNLOAD_MIME_TYPE = "application/zip"
-
-
-class WorkflowRunArchiveDownloadPreparer:
-    """
-    Build one ready-to-download CSV ZIP for a Redis archive download task.
-
-    The output object is deterministic for a given `download_id`, so retrying a failed task overwrites the same
-    temporary object instead of creating unbounded duplicate files. Source archive bundles are read from the archive
-    bucket, while the prepared ZIP is written to the export bucket so object lifecycle policies can expire downloads
-    without touching long-lived archives.
-    """
-
-    archive_storage: ArchiveStorage | None
-    download_storage: ArchiveStorage | None
-    cache: WorkflowRunArchiveDownloadTaskCache
-    session_factory: sessionmaker[Session]
-
-    def __init__(
-        self,
-        *,
-        storage: ArchiveStorage | None = None,
-        archive_storage: ArchiveStorage | None = None,
-        download_storage: ArchiveStorage | None = None,
-        cache: WorkflowRunArchiveDownloadTaskCache | None = None,
-        session_factory: sessionmaker[Session] | None = None,
-    ) -> None:
-        self.archive_storage = archive_storage or storage
-        self.download_storage = download_storage or storage
-        self.cache = cache or WorkflowRunArchiveDownloadTaskCache()
-        self.session_factory = session_factory or sessionmaker(bind=db.engine, expire_on_commit=False)
-
-    def prepare(self, *, tenant_id: str, download_id: str) -> WorkflowRunArchiveDownloadTask | None:
-        """Prepare a ZIP for an existing Redis task and persist terminal task state."""
-        task = self.cache.get(tenant_id=tenant_id, download_id=download_id)
-        if task is None:
-            logger.info("Workflow run archive download task expired before preparation: %s", download_id)
-            return None
-        if task.status == WorkflowRunArchiveDownloadStatus.READY:
-            return task
-        if task.status == WorkflowRunArchiveDownloadStatus.FAILED:
-            logger.info("Skipping failed workflow run archive download task: %s", download_id)
-            return task
-
-        processing_task = self._mark_processing(task)
-        try:
-            archive_storage = self.archive_storage or get_archive_storage()
-            download_storage = self.download_storage or get_export_storage()
-            bundles = self._get_task_bundles(processing_task)
-            payload = self._build_zip_payload(archive_storage, processing_task, bundles)
-            storage_key = build_archive_download_storage_key(processing_task)
-            download_storage.put_object(storage_key, payload)
-            return self._mark_ready(processing_task, storage_key=storage_key, file_size_bytes=len(payload))
-        except Exception as exc:
-            logger.exception("Failed to prepare workflow run archive download %s", download_id)
-            return self._mark_failed(processing_task, error=str(exc))
-
-    def _get_task_bundles(self, task: WorkflowRunArchiveDownloadTask) -> list[WorkflowRunArchiveBundle]:
-        with self.session_factory() as session:
-            return _list_task_bundles(session, task)
-
-    def _build_zip_payload(
-        self,
-        storage: ArchiveStorage,
-        task: WorkflowRunArchiveDownloadTask,
-        bundles: Sequence[WorkflowRunArchiveBundle],
-    ) -> bytes:
-        zip_root = f"workflow-run-logs-{task.year:04d}-{task.month:02d}"
-        csv_buffers: dict[str, io.BytesIO] = {}
-        csv_headers_written: set[str] = set()
-
-        for bundle in bundles:
-            object_prefix = _build_archive_bundle_object_prefix(task, bundle)
-            _, manifest = _load_and_validate_manifest(storage, task, bundle, object_prefix)
-            for table_name in sorted(manifest["tables"]):
-                entry = manifest["tables"][table_name]
-                object_key = entry["object_key"]
-                table_payload = storage.get_object(object_key)
-                _validate_table_payload(object_key=object_key, entry=entry, payload=table_payload)
-                csv_payload = _parquet_payload_to_csv(
-                    table_payload,
-                    include_header=table_name not in csv_headers_written,
-                )
-                if not csv_payload:
-                    continue
-                csv_buffers.setdefault(table_name, io.BytesIO()).write(csv_payload)
-                csv_headers_written.add(table_name)
-
-        buffer = io.BytesIO()
-        with zipfile.ZipFile(buffer, mode="w", compression=zipfile.ZIP_DEFLATED) as archive:
-            for table_name, csv_buffer in sorted(csv_buffers.items()):
-                archive.writestr(f"{zip_root}/{table_name}.csv", csv_buffer.getvalue())
-        return buffer.getvalue()
-
-    def _mark_processing(self, task: WorkflowRunArchiveDownloadTask) -> WorkflowRunArchiveDownloadTask:
-        now = datetime.datetime.now(datetime.UTC)
-        processing_task = task.model_copy(
-            update={
-                "status": WorkflowRunArchiveDownloadStatus.PROCESSING,
-                "error": None,
-                "updated_at": now,
-                "started_at": task.started_at or now,
-            }
-        )
-        self.cache.save(processing_task)
-        return processing_task
-
-    def _mark_ready(
-        self,
-        task: WorkflowRunArchiveDownloadTask,
-        *,
-        storage_key: str,
-        file_size_bytes: int,
-    ) -> WorkflowRunArchiveDownloadTask:
-        now = datetime.datetime.now(datetime.UTC)
-        ready_task = task.model_copy(
-            update={
-                "status": WorkflowRunArchiveDownloadStatus.READY,
-                "file_name": build_archive_download_file_name(task),
-                "storage_key": storage_key,
-                "file_size_bytes": file_size_bytes,
-                "error": None,
-                "updated_at": now,
-                "finished_at": now,
-            }
-        )
-        self.cache.save(ready_task)
-        return ready_task
-
-    def _mark_failed(self, task: WorkflowRunArchiveDownloadTask, *, error: str) -> WorkflowRunArchiveDownloadTask:
-        now = datetime.datetime.now(datetime.UTC)
-        failed_task = task.model_copy(
-            update={
-                "status": WorkflowRunArchiveDownloadStatus.FAILED,
-                "error": error,
-                "updated_at": now,
-                "finished_at": now,
-            }
-        )
-        self.cache.save(failed_task)
-        return failed_task
-
-
-def build_archive_download_file_name(task: WorkflowRunArchiveDownloadTask) -> str:
-    """Return the browser download filename for one monthly archive."""
-    return f"workflow-run-logs-{task.year:04d}-{task.month:02d}.zip"
-
-
-def build_archive_download_storage_key(task: WorkflowRunArchiveDownloadTask) -> str:
-    """Return the deterministic object-store key for a prepared download ZIP."""
-    return (
-        f"{ARCHIVE_DOWNLOAD_ROOT_PREFIX}tenant_prefix={task.tenant_id[0].lower()}/tenant_id={task.tenant_id}/"
-        f"year={task.year:04d}/month={task.month:02d}/{task.download_id}.zip"
-    )
-
-
-def _list_task_bundles(session: Session, task: WorkflowRunArchiveDownloadTask) -> list[WorkflowRunArchiveBundle]:
-    stmt = (
-        select(WorkflowRunArchiveBundle)
-        .where(
-            WorkflowRunArchiveBundle.tenant_id == task.tenant_id,
-            WorkflowRunArchiveBundle.year == task.year,
-            WorkflowRunArchiveBundle.month == task.month,
-        )
-        .order_by(WorkflowRunArchiveBundle.shard, WorkflowRunArchiveBundle.bundle_id)
-    )
-    indexed_bundles = list(session.scalars(stmt))
-    if task.bundle_refs:
-        requested_refs = [(ref.shard, ref.bundle_id) for ref in task.bundle_refs]
-    else:
-        requested_bundle_ids = set(task.bundle_ids)
-        requested_refs = [
-            (bundle.shard, bundle.bundle_id) for bundle in indexed_bundles if bundle.bundle_id in requested_bundle_ids
-        ]
-
-    bundle_by_ref = {(bundle.shard, bundle.bundle_id): bundle for bundle in indexed_bundles}
-    missing_refs = [ref for ref in requested_refs if ref not in bundle_by_ref]
-    if missing_refs:
-        raise ValueError(f"archive bundle index is missing requested bundles: {missing_refs}")
-
-    bundles = [bundle_by_ref[ref] for ref in requested_refs]
-    if len(bundles) != task.bundle_count:
-        raise ValueError(f"archive bundle count changed: expected={task.bundle_count}, actual={len(bundles)}")
-
-    download_id = build_archive_download_id(
-        tenant_id=task.tenant_id,
-        year=task.year,
-        month=task.month,
-        bundle_refs=requested_refs,
-    )
-    if download_id != task.download_id:
-        raise ValueError("archive download id no longer matches indexed bundle set")
-
-    return bundles
-
-
-def _build_archive_bundle_object_prefix(
-    task: WorkflowRunArchiveDownloadTask,
-    bundle: WorkflowRunArchiveBundle,
-) -> str:
-    return (
-        f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix={task.tenant_id[0].lower()}/tenant_id={task.tenant_id}/"
-        f"year={task.year:04d}/month={task.month:02d}/shard={bundle.shard}/bundle={bundle.bundle_id}"
-    )
-
-
-def _load_and_validate_manifest(
-    storage: ArchiveStorage,
-    task: WorkflowRunArchiveDownloadTask,
-    bundle: WorkflowRunArchiveBundle,
-    object_prefix: str,
-) -> tuple[bytes, ArchiveBundleManifest]:
-    manifest_key = f"{object_prefix}/{ARCHIVE_BUNDLE_MANIFEST_NAME}"
-    manifest_data = storage.get_object(manifest_key)
-    manifest = decode_archive_bundle_manifest(manifest_data)
-    _validate_manifest(task=task, bundle=bundle, manifest=manifest, object_prefix=object_prefix)
-    return manifest_data, manifest
-
-
-def _validate_manifest(
-    *,
-    task: WorkflowRunArchiveDownloadTask,
-    bundle: WorkflowRunArchiveBundle,
-    manifest: ArchiveBundleManifest,
-    object_prefix: str,
-) -> None:
-    if manifest["schema_version"] != ARCHIVE_BUNDLE_SCHEMA_VERSION:
-        raise ValueError(f"unsupported archive bundle schema version: {manifest['schema_version']}")
-    if manifest["archive_format"] != ARCHIVE_BUNDLE_FORMAT:
-        raise ValueError(f"unsupported archive bundle format: {manifest['archive_format']}")
-    if manifest["tenant_id"] != task.tenant_id:
-        raise ValueError(f"manifest tenant_id mismatch: expected={task.tenant_id}, actual={manifest['tenant_id']}")
-    if manifest["year"] != task.year:
-        raise ValueError(f"manifest year mismatch: expected={task.year}, actual={manifest['year']}")
-    if manifest["month"] != task.month:
-        raise ValueError(f"manifest month mismatch: expected={task.month}, actual={manifest['month']}")
-    if manifest["shard"] != bundle.shard:
-        raise ValueError(f"manifest shard mismatch: expected={bundle.shard}, actual={manifest['shard']}")
-    if manifest["bundle_id"] != bundle.bundle_id:
-        raise ValueError(f"manifest bundle_id mismatch: expected={bundle.bundle_id}, actual={manifest['bundle_id']}")
-    if manifest["object_prefix"] != object_prefix:
-        raise ValueError(
-            f"manifest object_prefix mismatch: expected={object_prefix}, actual={manifest['object_prefix']}"
-        )
-    if not manifest["tables"]:
-        raise ValueError("manifest tables must not be empty")
-    for table_name, raw_entry in manifest["tables"].items():
-        entry = cast(ArchiveBundleTableManifestEntry, raw_entry)
-        expected_object_key = f"{object_prefix}/{table_name}.parquet"
-        if entry["object_key"] != expected_object_key:
-            raise ValueError(
-                f"manifest object_key mismatch for {table_name}: "
-                f"expected={expected_object_key}, actual={entry['object_key']}"
-            )
-
-
-def _validate_table_payload(
-    *,
-    object_key: str,
-    entry: ArchiveBundleTableManifestEntry,
-    payload: bytes,
-) -> None:
-    if len(payload) != entry["size_bytes"]:
-        raise ValueError(f"archive object size mismatch for {object_key}")
-    checksum = hashlib.md5(payload).hexdigest()
-    if checksum != entry["checksum"]:
-        raise ValueError(f"archive object checksum mismatch for {object_key}")
-
-
-def _parquet_payload_to_csv(payload: bytes, *, include_header: bool) -> bytes:
-    table = pq.read_table(io.BytesIO(payload))
-    if table.num_columns == 0:
-        return b""
-    buffer = io.BytesIO()
-    pa_csv.write_csv(
-        table,
-        buffer,
-        write_options=pa_csv.WriteOptions(include_header=include_header),
-    )
-    return buffer.getvalue()
@@ -1,178 +0,0 @@
-"""Redis-backed temporary state for workflow-run archive downloads."""
-
-import datetime
-import hashlib
-import json
-import logging
-from collections.abc import Sequence
-from enum import StrEnum
-
-from pydantic import BaseModel, ConfigDict, Field
-
-from extensions.ext_redis import RedisClientWrapper, redis_client
-
-logger = logging.getLogger(__name__)
-
-ARCHIVE_DOWNLOAD_FORMAT_VERSION = "v1"
-DEFAULT_ARCHIVE_DOWNLOAD_TASK_TTL_SECONDS = 24 * 60 * 60
-_CACHE_KEY_PREFIX = "workflow_run_archive_download"
-
-
-class WorkflowRunArchiveDownloadStatus(StrEnum):
-    """Lifecycle state for an asynchronous archive download request."""
-
-    PENDING = "pending"
-    PROCESSING = "processing"
-    READY = "ready"
-    FAILED = "failed"
-
-
-class WorkflowRunArchiveBundleRef(BaseModel):
-    """Immutable object-store identity for one bundle included in a download task."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    shard: str
-    bundle_id: str
-
-
-class WorkflowRunArchiveDownloadTask(BaseModel):
-    """Temporary Redis payload for a monthly archive download request."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    download_id: str
-    tenant_id: str
-    requested_by: str
-    year: int = Field(ge=1)
-    month: int = Field(ge=1, le=12)
-    bundle_ids: list[str]
-    bundle_refs: list[WorkflowRunArchiveBundleRef] = Field(default_factory=list)
-    bundle_count: int = Field(ge=0)
-    archive_bytes: int = Field(ge=0)
-    status: WorkflowRunArchiveDownloadStatus
-    file_name: str | None = None
-    storage_key: str | None = None
-    file_size_bytes: int | None = Field(default=None, ge=0)
-    celery_task_id: str | None = None
-    error: str | None = None
-    created_at: datetime.datetime
-    updated_at: datetime.datetime
-    expires_at: datetime.datetime
-    started_at: datetime.datetime | None = None
-    finished_at: datetime.datetime | None = None
-
-
-class WorkflowRunArchiveDownloadTaskCache:
-    """Store ephemeral archive download task state in Redis with a TTL."""
-
-    _redis: RedisClientWrapper
-
-    def __init__(self, redis: RedisClientWrapper = redis_client) -> None:
-        self._redis = redis
-
-    def get(self, *, tenant_id: str, download_id: str) -> WorkflowRunArchiveDownloadTask | None:
-        raw = self._redis.get(self._cache_key(tenant_id=tenant_id, download_id=download_id))
-        if raw is None:
-            return None
-        data = raw.decode("utf-8") if isinstance(raw, bytes | bytearray) else raw
-        try:
-            return WorkflowRunArchiveDownloadTask.model_validate_json(data)
-        except ValueError:
-            logger.warning("Malformed workflow run archive download task cache entry: %s", download_id)
-            return None
-
-    def save(self, task: WorkflowRunArchiveDownloadTask) -> None:
-        ttl_seconds = self._ttl_seconds(task.expires_at)
-        self._redis.setex(
-            self._cache_key(tenant_id=task.tenant_id, download_id=task.download_id),
-            ttl_seconds,
-            task.model_dump_json(),
-        )
-
-    def create_if_absent(self, task: WorkflowRunArchiveDownloadTask) -> bool:
-        ttl_seconds = self._ttl_seconds(task.expires_at)
-        result = self._redis.set(
-            self._cache_key(tenant_id=task.tenant_id, download_id=task.download_id),
-            task.model_dump_json(),
-            ex=ttl_seconds,
-            nx=True,
-        )
-        return bool(result)
-
-    def delete(self, *, tenant_id: str, download_id: str) -> None:
-        self._redis.delete(self._cache_key(tenant_id=tenant_id, download_id=download_id))
-
-    @staticmethod
-    def _cache_key(*, tenant_id: str, download_id: str) -> str:
-        return f"{_CACHE_KEY_PREFIX}:{tenant_id}:{download_id}"
-
-    @staticmethod
-    def _ttl_seconds(expires_at: datetime.datetime) -> int:
-        expires_at_utc = expires_at if expires_at.tzinfo else expires_at.replace(tzinfo=datetime.UTC)
-        remaining = expires_at_utc - datetime.datetime.now(datetime.UTC)
-        return max(int(remaining.total_seconds()), 1)
-
-
-def build_pending_archive_download_task(
-    *,
-    tenant_id: str,
-    requested_by: str,
-    year: int,
-    month: int,
-    bundle_ids: Sequence[str],
-    bundle_refs: Sequence[tuple[str, str]] = (),
-    archive_bytes: int,
-    download_id: str,
-    ttl_seconds: int = DEFAULT_ARCHIVE_DOWNLOAD_TASK_TTL_SECONDS,
-    now: datetime.datetime | None = None,
-) -> WorkflowRunArchiveDownloadTask:
-    """Create the Redis payload stored when the console starts an archive download."""
-    created_at = now or datetime.datetime.now(datetime.UTC)
-    if created_at.tzinfo is None:
-        created_at = created_at.replace(tzinfo=datetime.UTC)
-    normalized_bundle_ids = list(bundle_ids)
-    normalized_bundle_refs = [
-        WorkflowRunArchiveBundleRef(shard=shard, bundle_id=bundle_id) for shard, bundle_id in bundle_refs
-    ]
-    return WorkflowRunArchiveDownloadTask(
-        download_id=download_id,
-        tenant_id=tenant_id,
-        requested_by=requested_by,
-        year=year,
-        month=month,
-        bundle_ids=normalized_bundle_ids,
-        bundle_refs=normalized_bundle_refs,
-        bundle_count=len(normalized_bundle_ids),
-        archive_bytes=archive_bytes,
-        status=WorkflowRunArchiveDownloadStatus.PENDING,
-        created_at=created_at,
-        updated_at=created_at,
-        expires_at=created_at + datetime.timedelta(seconds=ttl_seconds),
-    )
-
-
-def build_archive_download_id(
-    *,
-    tenant_id: str,
-    year: int,
-    month: int,
-    bundle_refs: Sequence[tuple[str, str]],
-    download_format_version: str = ARCHIVE_DOWNLOAD_FORMAT_VERSION,
-) -> str:
-    """Build a stable id for the exact archive download content."""
-    if not bundle_refs:
-        raise ValueError("bundle_refs must not be empty")
-    normalized_refs = sorted(f"{shard}:{bundle_id}" for shard, bundle_id in bundle_refs)
-    payload = json.dumps(
-        {
-            "tenant_id": tenant_id,
-            "year": year,
-            "month": month,
-            "bundle_refs": normalized_refs,
-            "download_format_version": download_format_version,
-        },
-        sort_keys=True,
-        separators=(",", ":"),
-    )
-    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:32]
@@ -1,293 +0,0 @@
-"""
-Console-facing workflow-run archive queries.
-
-The object store remains the recoverable archive source of truth. This module only reads the DB bundle index and writes
-temporary Redis download-task state, so console requests never list R2 online.
-"""
-
-import datetime
-import logging
-import uuid
-from collections.abc import Callable
-from dataclasses import dataclass
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from models.workflow import WorkflowRunArchiveBundle
-from services.retention.workflow_run.archive_download_task_cache import (
-    WorkflowRunArchiveDownloadStatus,
-    WorkflowRunArchiveDownloadTask,
-    WorkflowRunArchiveDownloadTaskCache,
-    build_archive_download_id,
-    build_pending_archive_download_task,
-)
-
-logger = logging.getLogger(__name__)
-
-ArchiveDownloadTaskDispatcher = Callable[
-    [WorkflowRunArchiveDownloadTask, WorkflowRunArchiveDownloadTaskCache],
-    WorkflowRunArchiveDownloadTask,
-]
-
-
-@dataclass(frozen=True)
-class WorkflowRunArchiveMonth:
-    """Aggregated archive metadata for one tenant/month."""
-
-    year: int
-    month: int
-    bundle_count: int
-    workflow_run_count: int
-    row_count: int
-    archive_bytes: int
-    latest_archived_at: datetime.datetime
-    download_task: WorkflowRunArchiveDownloadTask | None
-
-
-@dataclass(frozen=True)
-class WorkflowRunArchiveSummary:
-    """Top-level archive totals shown on the console page."""
-
-    archived_month_count: int
-    workflow_run_count: int
-    archive_bytes: int
-    latest_archived_at: datetime.datetime | None
-
-
-@dataclass(frozen=True)
-class WorkflowRunArchiveList:
-    """Console response model before controller serialization."""
-
-    summary: WorkflowRunArchiveSummary
-    months: list[WorkflowRunArchiveMonth]
-
-
-class WorkflowRunArchiveNotFoundError(Exception):
-    """Raised when no archive bundles exist for a requested tenant/month."""
-
-
-class WorkflowRunArchiveDownloadTaskNotFoundError(Exception):
-    """Raised when the temporary Redis task has expired or never existed."""
-
-
-class WorkflowRunArchiveDownloadNotReadyError(Exception):
-    """Raised when a cached download task has not produced a file yet."""
-
-
-def list_workflow_run_archives(
-    session: Session,
-    tenant_id: str,
-    *,
-    cache: WorkflowRunArchiveDownloadTaskCache | None = None,
-) -> WorkflowRunArchiveList:
-    """Return monthly archive metadata for one tenant from the DB bundle index."""
-    stmt = (
-        select(WorkflowRunArchiveBundle)
-        .where(WorkflowRunArchiveBundle.tenant_id == tenant_id)
-        .order_by(
-            WorkflowRunArchiveBundle.year.desc(),
-            WorkflowRunArchiveBundle.month.desc(),
-            WorkflowRunArchiveBundle.shard,
-            WorkflowRunArchiveBundle.bundle_id,
-        )
-    )
-    month_bundles: dict[tuple[int, int], list[WorkflowRunArchiveBundle]] = {}
-    for bundle in session.scalars(stmt):
-        month_bundles.setdefault((bundle.year, bundle.month), []).append(bundle)
-
-    task_cache = cache or WorkflowRunArchiveDownloadTaskCache()
-    months: list[WorkflowRunArchiveMonth] = []
-    for (year, month), bundles in month_bundles.items():
-        bundle_refs = [(bundle.shard, bundle.bundle_id) for bundle in bundles]
-        months.append(
-            WorkflowRunArchiveMonth(
-                year=year,
-                month=month,
-                bundle_count=len(bundles),
-                workflow_run_count=sum(bundle.workflow_run_count for bundle in bundles),
-                row_count=sum(bundle.row_count for bundle in bundles),
-                archive_bytes=sum(bundle.archive_bytes for bundle in bundles),
-                latest_archived_at=max(bundle.archived_at for bundle in bundles),
-                download_task=_get_cached_month_download_task(
-                    task_cache,
-                    tenant_id=tenant_id,
-                    year=year,
-                    month=month,
-                    bundle_refs=bundle_refs,
-                ),
-            )
-        )
-    latest_archived_at = max((month.latest_archived_at for month in months), default=None)
-    return WorkflowRunArchiveList(
-        summary=WorkflowRunArchiveSummary(
-            archived_month_count=len(months),
-            workflow_run_count=sum(month.workflow_run_count for month in months),
-            archive_bytes=sum(month.archive_bytes for month in months),
-            latest_archived_at=latest_archived_at,
-        ),
-        months=months,
-    )
-
-
-def _get_cached_month_download_task(
-    cache: WorkflowRunArchiveDownloadTaskCache,
-    *,
-    tenant_id: str,
-    year: int,
-    month: int,
-    bundle_refs: list[tuple[str, str]],
-) -> WorkflowRunArchiveDownloadTask | None:
-    if not bundle_refs:
-        return None
-    download_id = build_archive_download_id(
-        tenant_id=tenant_id,
-        year=year,
-        month=month,
-        bundle_refs=bundle_refs,
-    )
-    try:
-        return cache.get(tenant_id=tenant_id, download_id=download_id)
-    except Exception:
-        logger.warning("Failed to read cached workflow run archive download task: %s", download_id, exc_info=True)
-        return None
-
-
-def create_workflow_run_archive_download_task(
-    session: Session,
-    *,
-    tenant_id: str,
-    requested_by: str,
-    year: int,
-    month: int,
-    cache: WorkflowRunArchiveDownloadTaskCache | None = None,
-    dispatcher: ArchiveDownloadTaskDispatcher | None = None,
-) -> WorkflowRunArchiveDownloadTask:
-    """
-    Create or return the idempotent Redis task for downloading one tenant/month archive.
-
-    The task identity is based on the exact ordered bundle set currently indexed for the month. If the month receives a
-    new bundle later, the next request gets a different download id and prepares a fresh file.
-    """
-    bundles = _list_archive_bundles(session, tenant_id=tenant_id, year=year, month=month)
-    if not bundles:
-        raise WorkflowRunArchiveNotFoundError(f"Workflow run archive not found: {year:04d}-{month:02d}")
-
-    bundle_refs = [(bundle.shard, bundle.bundle_id) for bundle in bundles]
-    download_id = build_archive_download_id(
-        tenant_id=tenant_id,
-        year=year,
-        month=month,
-        bundle_refs=bundle_refs,
-    )
-    task = build_pending_archive_download_task(
-        tenant_id=tenant_id,
-        requested_by=requested_by,
-        year=year,
-        month=month,
-        bundle_ids=[bundle.bundle_id for bundle in bundles],
-        bundle_refs=bundle_refs,
-        archive_bytes=sum(bundle.archive_bytes for bundle in bundles),
-        download_id=download_id,
-    )
-    task_cache = cache or WorkflowRunArchiveDownloadTaskCache()
-    dispatch = dispatcher or _dispatch_workflow_run_archive_download_task
-    if task_cache.create_if_absent(task):
-        return dispatch(task, task_cache)
-
-    existing = task_cache.get(tenant_id=tenant_id, download_id=download_id)
-    if existing is not None:
-        if existing.status == WorkflowRunArchiveDownloadStatus.FAILED:
-            task_cache.save(task)
-            return dispatch(task, task_cache)
-        if existing.status == WorkflowRunArchiveDownloadStatus.PENDING and not existing.celery_task_id:
-            return dispatch(existing, task_cache)
-        return existing
-
-    task_cache.save(task)
-    return dispatch(task, task_cache)
-
-
-def get_workflow_run_archive_download_task(
-    *,
-    tenant_id: str,
-    download_id: str,
-    cache: WorkflowRunArchiveDownloadTaskCache | None = None,
-) -> WorkflowRunArchiveDownloadTask:
-    """Return a cached archive download task or raise when the TTL has expired."""
-    task_cache = cache or WorkflowRunArchiveDownloadTaskCache()
-    task = task_cache.get(tenant_id=tenant_id, download_id=download_id)
-    if task is None:
-        raise WorkflowRunArchiveDownloadTaskNotFoundError(f"Workflow run archive download not found: {download_id}")
-    return task
-
-
-def get_ready_workflow_run_archive_download_task(
-    *,
-    tenant_id: str,
-    download_id: str,
-    cache: WorkflowRunArchiveDownloadTaskCache | None = None,
-) -> WorkflowRunArchiveDownloadTask:
-    """Return a ready cached archive download task or raise when the file is not available."""
-    task = get_workflow_run_archive_download_task(tenant_id=tenant_id, download_id=download_id, cache=cache)
-    if task.status != WorkflowRunArchiveDownloadStatus.READY or not task.storage_key or not task.file_name:
-        raise WorkflowRunArchiveDownloadNotReadyError(f"Workflow run archive download is not ready: {download_id}")
-    return task
-
-
-def _list_archive_bundles(
-    session: Session,
-    *,
-    tenant_id: str,
-    year: int,
-    month: int,
-) -> list[WorkflowRunArchiveBundle]:
-    stmt = (
-        select(WorkflowRunArchiveBundle)
-        .where(
-            WorkflowRunArchiveBundle.tenant_id == tenant_id,
-            WorkflowRunArchiveBundle.year == year,
-            WorkflowRunArchiveBundle.month == month,
-        )
-        .order_by(WorkflowRunArchiveBundle.shard, WorkflowRunArchiveBundle.bundle_id)
-    )
-    return list(session.scalars(stmt))
-
-
-def _dispatch_workflow_run_archive_download_task(
-    task: WorkflowRunArchiveDownloadTask,
-    cache: WorkflowRunArchiveDownloadTaskCache,
-) -> WorkflowRunArchiveDownloadTask:
-    """
-    Enqueue background ZIP preparation and persist the Celery id before the worker can start.
-
-    The Redis task key is the idempotency boundary. We generate the Celery id in the API process, save it on the task,
-    then submit with that exact id so duplicate console requests keep seeing one logical download request.
-    """
-    from tasks.workflow_run_archive_download_tasks import prepare_workflow_run_archive_download_task
-
-    now = datetime.datetime.now(datetime.UTC)
-    celery_task_id = uuid.uuid4().hex
-    queued_task = task.model_copy(update={"celery_task_id": celery_task_id, "updated_at": now})
-    cache.save(queued_task)
-
-    try:
-        prepare_workflow_run_archive_download_task.apply_async(
-            args=(queued_task.tenant_id, queued_task.download_id),
-            task_id=celery_task_id,
-        )
-    except Exception:
-        failure_time = datetime.datetime.now(datetime.UTC)
-        failed_task = queued_task.model_copy(
-            update={
-                "status": WorkflowRunArchiveDownloadStatus.FAILED,
-                "error": "Failed to enqueue archive download task.",
-                "updated_at": failure_time,
-                "finished_at": failure_time,
-            }
-        )
-        cache.save(failed_task)
-        logger.exception("Failed to enqueue workflow run archive download task %s", queued_task.download_id)
-        return failed_task
-
-    return queued_task
@@ -5,8 +5,8 @@ This service archives workflow run logs for paid plan users older than the confi
 90 days) to S3-compatible storage.

 Archive V2 writes bundle-level Parquet objects. A bundle contains many workflow runs and their related table rows.
-Bundle metadata lives in the object-store manifest as the recoverable source of truth. Completed bundles are also
-mirrored into a small database index so console listing and download jobs do not list object storage online.
+Bundle metadata lives in the object-store manifest instead of a database table, so archive/delete/restore does not move
+the large-table retention problem into another OLTP table.

 Archive campaigns should use fixed absolute UTC windows for every tenant-prefix/shard execution. Relative windows are
 evaluated at process start and are not safe for multi-day rollout because each command would scan a different window.
@@ -29,12 +29,12 @@ import hashlib
 import json
 import logging
 import time
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass, field
 from enum import Enum
 from threading import Lock
-from typing import Any, NotRequired, TypedDict, cast
+from typing import Any, NotRequired, TypedDict, TypeVar, cast

 import click
 import pyarrow as pa
@@ -64,20 +64,16 @@ from repositories.api_workflow_node_execution_repository import DifyAPIWorkflowN
 from repositories.api_workflow_run_repository import APIWorkflowRunRepository
 from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerLogRepository
 from services.billing_service import BillingService
-from services.retention.workflow_run.archive_bundle_index import (
-    ArchiveBundleManifest,
-    ArchiveBundleTableManifestEntry,
-    decode_archive_bundle_manifest,
-    upsert_archive_bundle_index_from_manifest,
-)
 from services.retention.workflow_run.constants import (
    ARCHIVE_BUNDLE_FORMAT,
    ARCHIVE_BUNDLE_INDEX_NAME,
    ARCHIVE_BUNDLE_MANIFEST_NAME,
    ARCHIVE_BUNDLE_SCHEMA_VERSION,
 )
+from services.retention.workflow_run.db_retry import is_retryable_db_disconnect, run_with_db_retry

 logger = logging.getLogger(__name__)
+T = TypeVar("T")


 class TableStatsManifestEntry(TypedDict):
@@ -214,6 +210,8 @@ class WorkflowRunArchiver:
        "workflow_pause_reasons",
        "workflow_trigger_logs",
    ]
+    DB_RETRY_ATTEMPTS = 3
+    DB_RETRY_DELAYS_SECONDS = (1.0, 2.0)

    start_from: datetime.datetime | None
    end_before: datetime.datetime
@@ -461,16 +459,40 @@ class WorkflowRunArchiver:
        """Fetch a batch of workflow runs to archive."""
        repo = self._get_workflow_run_repo()
        tenant_ids = list(tenant_scope) if tenant_scope is not None else self.tenant_ids or None
-        return repo.get_runs_batch_by_time_range(
-            start_from=self.start_from,
-            end_before=self.end_before,
-            last_seen=last_seen,
-            batch_size=self.batch_size,
-            run_types=self.ARCHIVED_TYPE,
-            tenant_ids=tenant_ids,
-            tenant_prefixes=None if tenant_ids else self.tenant_prefixes or None,
-            run_shard_index=self.run_shard_index,
-            run_shard_total=self.run_shard_total,
+
+        return self._run_with_db_retry(
+            "workflow run batch fetch",
+            lambda: repo.get_runs_batch_by_time_range(
+                start_from=self.start_from,
+                end_before=self.end_before,
+                last_seen=last_seen,
+                batch_size=self.batch_size,
+                run_types=self.ARCHIVED_TYPE,
+                tenant_ids=tenant_ids,
+                tenant_prefixes=None if tenant_ids else self.tenant_prefixes or None,
+                run_shard_index=self.run_shard_index,
+                run_shard_total=self.run_shard_total,
+            ),
+        )
+
+    @staticmethod
+    def _is_retryable_db_disconnect(exc: BaseException) -> bool:
+        return is_retryable_db_disconnect(exc)
+
+    @staticmethod
+    def _safe_rollback(session: Session, bundle_id: str) -> None:
+        try:
+            session.rollback()
+        except Exception:
+            logger.warning("Failed to rollback archive session for bundle %s", bundle_id, exc_info=True)
+
+    def _run_with_db_retry(self, operation_name: str, operation: Callable[[], T]) -> T:
+        return run_with_db_retry(
+            operation_name,
+            operation,
+            logger=logger,
+            attempts=self.DB_RETRY_ATTEMPTS,
+            delays_seconds=self.DB_RETRY_DELAYS_SECONDS,
        )

    def _tenant_scan_scopes(self) -> list[list[str] | None]:
@@ -538,16 +560,14 @@ class WorkflowRunArchiver:
        if self.workers == 1 or len(bundle_groups) == 1:
            results: list[ArchiveResult] = []
            for bundle_runs in bundle_groups:
-                with session_maker() as session:
-                    results.append(self._archive_bundle(session, storage, bundle_runs))
+                results.append(self._archive_bundle_with_retry(session_maker, storage, bundle_runs))
            return results

        results = []
        max_workers = min(self.workers, len(bundle_groups))

        def archive_in_worker(bundle_runs: Sequence[WorkflowRun]) -> ArchiveResult:
-            with session_maker() as session:
-                return self._archive_bundle(session, storage, bundle_runs)
+            return self._archive_bundle_with_retry(session_maker, storage, bundle_runs)

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = [executor.submit(archive_in_worker, bundle_runs) for bundle_runs in bundle_groups]
@@ -555,6 +575,39 @@ class WorkflowRunArchiver:
                results.append(future.result())
        return results

+    def _archive_bundle_with_retry(
+        self,
+        session_maker: sessionmaker[Session],
+        storage: ArchiveStorage | None,
+        runs: Sequence[WorkflowRun],
+    ) -> ArchiveResult:
+        identity = self._build_bundle_identity(runs)
+
+        try:
+            return self._run_with_db_retry(
+                f"archive workflow run bundle {identity.bundle_id}",
+                lambda: self._archive_bundle_once(session_maker, storage, runs),
+            )
+        except Exception as exc:
+            logger.exception("Failed to archive workflow run bundle %s after retries", identity.bundle_id)
+            return ArchiveResult(
+                bundle_id=identity.bundle_id,
+                tenant_id=identity.tenant_id,
+                object_prefix=identity.object_prefix,
+                run_count=len(runs),
+                success=False,
+                error=str(exc),
+            )
+
+    def _archive_bundle_once(
+        self,
+        session_maker: sessionmaker[Session],
+        storage: ArchiveStorage | None,
+        runs: Sequence[WorkflowRun],
+    ) -> ArchiveResult:
+        with session_maker() as session:
+            return self._archive_bundle(session, storage, runs)
+
    def _archive_bundle(
        self,
        session: Session,
@@ -581,7 +634,6 @@ class WorkflowRunArchiver:
                    raise ArchiveStorageNotConfiguredError("Archive storage not configured")
                if storage.object_exists(self._get_manifest_object_key(identity)):
                    self._write_bundle_index(storage, identity)
-                    self._sync_existing_bundle_index(session, storage, identity)
                    result.success = True
                    result.skipped = True
                    result.error = "bundle already archived"
@@ -605,7 +657,6 @@ class WorkflowRunArchiver:
                result.run_count = len(runs)
                if storage.object_exists(self._get_manifest_object_key(identity)):
                    self._write_bundle_index(storage, identity)
-                    self._sync_existing_bundle_index(session, storage, identity)
                    result.success = True
                    result.skipped = True
                    result.error = "filtered bundle already archived"
@@ -637,8 +688,6 @@ class WorkflowRunArchiver:
                    storage.put_object(self._get_table_object_key(identity, table_name), payload)
                storage.put_object(self._get_manifest_object_key(identity), manifest_data)
                self._merge_bundle_manifest_into_index(storage, identity, [run.id for run in runs])
-                manifest = decode_archive_bundle_manifest(manifest_data)
-                upsert_archive_bundle_index_from_manifest(session, manifest, len(manifest_data))
                session.commit()

                logger.info(
@@ -655,30 +704,16 @@ class WorkflowRunArchiver:
                result.success = True

        except Exception as e:
+            if self._is_retryable_db_disconnect(e):
+                self._safe_rollback(session, identity.bundle_id)
+                raise
            logger.exception("Failed to archive workflow run bundle %s", identity.bundle_id)
            result.error = str(e)
-            session.rollback()
+            self._safe_rollback(session, identity.bundle_id)

        result.elapsed_time = time.time() - start_time
        return result

-    def _sync_existing_bundle_index(
-        self,
-        session: Session,
-        storage: ArchiveStorage,
-        identity: ArchiveBundleIdentity,
-    ) -> None:
-        """Best-effort DB index sync for a bundle whose manifest already exists in archive storage."""
-        manifest_key = self._get_manifest_object_key(identity)
-        try:
-            manifest_data = storage.get_object(manifest_key)
-            manifest = decode_archive_bundle_manifest(manifest_data)
-            upsert_archive_bundle_index_from_manifest(session, manifest, len(manifest_data))
-            session.commit()
-        except Exception:
-            session.rollback()
-            logger.warning("Failed to sync workflow archive bundle index for %s", manifest_key, exc_info=True)
-
    def _lock_runs_for_archive(
        self,
        session: Session,
@@ -806,9 +841,9 @@ class WorkflowRunArchiver:
        identity: ArchiveBundleIdentity,
        runs: Sequence[WorkflowRun],
        table_stats: list[TableStats],
-    ) -> ArchiveBundleManifest:
+    ) -> ArchiveManifestDict:
        """Generate a manifest for the archived workflow run bundle."""
-        tables: dict[str, ArchiveBundleTableManifestEntry] = {
+        tables: dict[str, TableStatsManifestEntry] = {
            stat.table_name: {
                "row_count": stat.row_count,
                "checksum": stat.checksum,
@@ -821,8 +856,8 @@ class WorkflowRunArchiver:
        end_before = self.end_before
        if end_before is None:
            raise ValueError("archive window end must be set")
-        archive_window_end = self._format_window_datetime(end_before)
-        if archive_window_end is None:
+        formatted_end_before = self._format_window_datetime(end_before)
+        if formatted_end_before is None:
            raise ValueError("archive window end must be set")
        return ArchiveManifestDict(
            schema_version=ARCHIVE_BUNDLE_SCHEMA_VERSION,
@@ -843,7 +878,7 @@ class WorkflowRunArchiver:
            archived_at=datetime.datetime.now(datetime.UTC).isoformat(),
            campaign_id=self.campaign_id,
            archive_window_start=self._format_window_datetime(self.start_from),
-            archive_window_end=archive_window_end,
+            archive_window_end=formatted_end_before,
            run_shard=identity.shard,
            tables=tables,
            run_ids=[run.id for run in sorted_runs],
@@ -1,10 +1,9 @@
 """
 Maintain V2 workflow-run archive bundles.

-Archive V2 keeps object-store manifests as the recoverable bundle source of truth. This maintenance module still
-discovers delete/restore targets by listing `manifest.json` objects and uses object-store marker files for
-delete/restore state. The separate database bundle index is intended for console listing and download jobs, not as the
-source of truth for destructive maintenance.
+Archive V2 keeps bundle metadata in object-store manifests, not in a database table. This module discovers bundles by
+listing `manifest.json` objects, uses object-store marker files for delete/restore state, and only touches the database
+for source-table validation, deletion, and restoration.

 Each bundle is processed in its own database transaction. A failed bundle leaves source rows unchanged unless the
 transaction has already committed; marker handling makes the next run able to reconcile the common committed-but-marker
@@ -0,0 +1,66 @@
+import logging
+import time
+from collections.abc import Callable
+
+from sqlalchemy.exc import DBAPIError
+from sqlalchemy.exc import OperationalError as SQLAlchemyOperationalError
+
+DEFAULT_DB_RETRY_ATTEMPTS = 3
+DEFAULT_DB_RETRY_DELAYS_SECONDS = (1.0, 2.0)
+
+_DB_DISCONNECT_PATTERNS = (
+    "server closed the connection unexpectedly",
+    "connection already closed",
+    "closed the connection",
+    "connection not open",
+    "terminating connection",
+    "connection reset",
+    "broken pipe",
+    "connection invalidated",
+)
+
+
+def is_retryable_db_disconnect(exc: BaseException) -> bool:
+    if isinstance(exc, DBAPIError) and exc.connection_invalidated:
+        return True
+
+    if not _is_db_operational_error(exc):
+        return False
+
+    original_exception = exc.orig if isinstance(exc, DBAPIError) else None
+    message = f"{exc} {original_exception or ''}".lower()
+    return any(pattern in message for pattern in _DB_DISCONNECT_PATTERNS)
+
+
+def run_with_db_retry[T](
+    operation_name: str,
+    operation: Callable[[], T],
+    *,
+    logger: logging.Logger,
+    attempts: int = DEFAULT_DB_RETRY_ATTEMPTS,
+    delays_seconds: tuple[float, ...] = DEFAULT_DB_RETRY_DELAYS_SECONDS,
+) -> T:
+    for attempt in range(1, attempts + 1):
+        try:
+            return operation()
+        except Exception as exc:
+            if not is_retryable_db_disconnect(exc) or attempt == attempts:
+                raise
+            delay = delays_seconds[min(attempt - 1, len(delays_seconds) - 1)]
+            logger.warning(
+                "Retrying %s after retryable DB disconnect (attempt %s/%s, sleep %.1fs)",
+                operation_name,
+                attempt,
+                attempts,
+                delay,
+                exc_info=True,
+            )
+            time.sleep(delay)
+    raise RuntimeError(f"{operation_name} did not complete")
+
+
+def _is_db_operational_error(exc: BaseException) -> bool:
+    if isinstance(exc, SQLAlchemyOperationalError):
+        return True
+
+    return exc.__class__.__name__ == "OperationalError" and exc.__class__.__module__.startswith(("psycopg", "psycopg2"))
@@ -6,7 +6,7 @@ import uuid
 from datetime import UTC, datetime
 from typing import TypedDict, cast

-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.orm import Session

 from core.db.session_factory import session_factory
@@ -912,12 +912,11 @@ class SummaryIndexService:

            # Disable summary records (don't delete)
            now = naive_utc_now()
-            for summary in summaries:
-                summary.enabled = False
-                summary.disabled_at = now
-                summary.disabled_by = disabled_by
-                session.add(summary)
-
+            session.execute(
+                update(DocumentSegmentSummary)
+                .where(DocumentSegmentSummary.id.in_(s.id for s in summaries))
+                .values(enabled=False, disabled_at=now, disabled_by=disabled_by)
+            )
            session.commit()
            logger.info("Disabled %s summary records for dataset %s", len(summaries), dataset.id)

@@ -101,6 +101,7 @@ class WebhookService:
                - Mapping[str, Any]: The node configuration data

        Raises:
+            QuotaExceededError: If the app trigger is rate limited
            ValueError: If webhook not found, app trigger not found, trigger disabled, or workflow not found
        """
        with Session(db.engine) as session:
@@ -141,8 +142,10 @@ class WebhookService:
                # Only check enabled status if not in debug mode

                if app_trigger.status == AppTriggerStatus.RATE_LIMITED:
-                    raise ValueError(
-                        f"Webhook trigger is rate limited for webhook {webhook_id}, please upgrade your plan."
+                    raise QuotaExceededError(
+                        feature=QuotaType.TRIGGER.value,
+                        tenant_id=webhook_trigger.tenant_id,
+                        required=1,
                    )

                if app_trigger.status != AppTriggerStatus.ENABLED:
@@ -799,6 +802,7 @@ class WebhookService:
            workflow: The workflow to execute

        Raises:
+            QuotaExceededError: If the tenant has exhausted its trigger or workflow execution quota
            ValueError: If tenant owner is not found
            Exception: If workflow execution fails
        """
@@ -824,11 +828,6 @@ class WebhookService:
                quota_charge = QuotaService.reserve(QuotaType.TRIGGER, webhook_trigger.tenant_id)
            except QuotaExceededError:
                AppTriggerService.mark_tenant_triggers_rate_limited(webhook_trigger.tenant_id)
-                logger.info(
-                    "Tenant %s rate limited, skipping webhook trigger %s",
-                    webhook_trigger.tenant_id,
-                    webhook_trigger.webhook_id,
-                )
                raise

            try:
@@ -841,6 +840,14 @@ class WebhookService:
                quota_charge.refund()
                raise

+        except QuotaExceededError as e:
+            logger.info(
+                "Tenant %s quota exceeded for feature %s, skipping webhook trigger %s",
+                webhook_trigger.tenant_id,
+                e.feature,
+                webhook_trigger.webhook_id,
+            )
+            raise
        except Exception:
            logger.exception("Failed to trigger workflow for webhook %s", webhook_trigger.webhook_id)
            raise
@@ -0,0 +1,53 @@
+"""Initialize default RBAC access for existing workspace members after app creation."""
+
+import logging
+
+from celery import shared_task
+
+from configs import dify_config
+from extensions.ext_database import db
+from services.account_service import TenantService
+from services.enterprise import rbac_service as enterprise_rbac_service
+
+logger = logging.getLogger(__name__)
+
+APP_RBAC_ACCOUNT_POLICY_BATCH_SIZE = 500
+APP_RBAC_DEFAULT_ACCESS_POLICY_ID = "default"
+APP_RBAC_QUEUE = "app_rbac"
+
+
+@shared_task(queue=APP_RBAC_QUEUE, bind=True, max_retries=3, default_retry_delay=60)
+def initialize_created_app_rbac_access_task(self, tenant_id: str, account_id: str, app_id: str) -> None:
+    """Grant the default app policy to current workspace members.
+
+    App scope is persisted synchronously before this task is queued. Replacing
+    member policies is idempotent, so retrying the whole synchronization is safe
+    when the enterprise RBAC service is temporarily unavailable.
+    """
+    if not dify_config.RBAC_ENABLED:
+        return
+
+    try:
+        for account_ids in TenantService.iter_member_account_id_batches(
+            tenant_id,
+            APP_RBAC_ACCOUNT_POLICY_BATCH_SIZE,
+            session=db.session(),
+        ):
+            enterprise_rbac_service.RBACService.AppAccess.replace_user_access_policies(
+                tenant_id=tenant_id,
+                account_id=account_id,
+                app_id=app_id,
+                target_account_id=None,
+                payload=enterprise_rbac_service.ReplaceUserAccessPolicies(
+                    access_policy_ids=[APP_RBAC_DEFAULT_ACCESS_POLICY_ID],
+                    account_ids=account_ids,
+                ),
+            )
+    except Exception as exc:
+        logger.exception(
+            "Failed to initialize app RBAC access; retrying: tenant_id=%s app_id=%s attempt=%s",
+            tenant_id,
+            app_id,
+            self.request.retries + 1,
+        )
+        raise self.retry(exc=exc)
@@ -1,18 +0,0 @@
-"""Celery tasks for preparing workflow-run archive downloads."""
-
-import logging
-
-from celery import shared_task
-
-from services.retention.workflow_run.archive_download_preparation import WorkflowRunArchiveDownloadPreparer
-
-logger = logging.getLogger(__name__)
-
-WORKFLOW_RUN_ARCHIVE_DOWNLOAD_QUEUE = "workflow_archive"
-
-
-@shared_task(queue=WORKFLOW_RUN_ARCHIVE_DOWNLOAD_QUEUE)
-def prepare_workflow_run_archive_download_task(tenant_id: str, download_id: str) -> None:
-    """Prepare a cached workflow-run archive download in the background."""
-    logger.info("Preparing workflow run archive download: tenant=%s download_id=%s", tenant_id, download_id)
-    WorkflowRunArchiveDownloadPreparer().prepare(tenant_id=tenant_id, download_id=download_id)
@@ -6,9 +6,10 @@ from unittest.mock import ANY, MagicMock, patch
 import pyarrow as pa
 import pyarrow.parquet as pq
 import pytest
+from sqlalchemy.exc import OperationalError

-from models.workflow import WorkflowRunArchiveBundle
 from services.retention.workflow_run.archive_paid_plan_workflow_run import (
+    ArchiveResult,
    ArchiveSummary,
    WorkflowRunArchiver,
 )
@@ -33,6 +34,30 @@ class FakeArchiveStorage:
        return sorted(key for key in self.objects if key.startswith(prefix))


+def _db_disconnect_error() -> OperationalError:
+    return OperationalError(
+        "select 1",
+        {},
+        RuntimeError("server closed the connection unexpectedly"),
+        connection_invalidated=True,
+    )
+
+
+def _run(run_id: str = "run-1"):
+    run = MagicMock()
+    run.id = run_id
+    run.tenant_id = "tenant-1"
+    run.created_at = datetime.datetime(2025, 3, 15, 10, 0, 0)
+    return run
+
+
+def _session_context(session):
+    context = MagicMock()
+    context.__enter__.return_value = session
+    context.__exit__.return_value = False
+    return context
+
+
 class TestWorkflowRunArchiverInit:
    def test_start_from_without_end_before_raises(self):
        with pytest.raises(ValueError, match="start_from and end_before must be provided together"):
@@ -140,6 +165,32 @@ class TestWorkflowRunArchiverInit:
        repo.get_runs_batch_by_time_range.assert_called_once()
        assert repo.get_runs_batch_by_time_range.call_args.kwargs["tenant_ids"] == ["tenant-b"]

+    def test_get_runs_batch_retries_retryable_db_disconnect(self):
+        repo = MagicMock()
+        repo.get_runs_batch_by_time_range.side_effect = [_db_disconnect_error(), []]
+        archiver = WorkflowRunArchiver(workflow_run_repo=repo)
+
+        with patch("services.retention.workflow_run.db_retry.time.sleep") as sleep:
+            runs = archiver._get_runs_batch(None)
+
+        assert runs == []
+        assert repo.get_runs_batch_by_time_range.call_count == 2
+        sleep.assert_called_once_with(1.0)
+
+    def test_get_runs_batch_does_not_retry_non_db_broken_pipe_error(self):
+        repo = MagicMock()
+        repo.get_runs_batch_by_time_range.side_effect = RuntimeError("broken pipe")
+        archiver = WorkflowRunArchiver(workflow_run_repo=repo)
+
+        with (
+            patch("services.retention.workflow_run.db_retry.time.sleep") as sleep,
+            pytest.raises(RuntimeError, match="broken pipe"),
+        ):
+            archiver._get_runs_batch(None)
+
+        repo.get_runs_batch_by_time_range.assert_called_once()
+        sleep.assert_not_called()
+
    def test_start_message_includes_shard(self):
        archiver = WorkflowRunArchiver(tenant_prefixes=["0"], run_shard_index=1, run_shard_total=4)

@@ -352,6 +403,72 @@ class TestDryRunArchive:
        assert summary.table_stats["workflow_app_logs"].size_bytes == 32


+class TestArchiveDbRetry:
+    def test_archive_bundle_groups_retries_with_fresh_session(self):
+        archiver = WorkflowRunArchiver(days=90)
+        run = _run()
+        session_maker = MagicMock(
+            side_effect=[
+                _session_context(MagicMock(name="session-1")),
+                _session_context(MagicMock(name="session-2")),
+            ]
+        )
+        success = ArchiveResult(
+            bundle_id=archiver._build_bundle_identity([run]).bundle_id,
+            tenant_id=run.tenant_id,
+            object_prefix=archiver._build_bundle_identity([run]).object_prefix,
+            run_count=1,
+            success=True,
+        )
+
+        with (
+            patch.object(archiver, "_archive_bundle", side_effect=[_db_disconnect_error(), success]) as archive_bundle,
+            patch("services.retention.workflow_run.db_retry.time.sleep") as sleep,
+        ):
+            results = archiver._archive_bundle_groups(session_maker, MagicMock(), [[run]])
+
+        assert results == [success]
+        assert archive_bundle.call_count == 2
+        assert session_maker.call_count == 2
+        sleep.assert_called_once_with(1.0)
+
+    def test_archive_bundle_groups_returns_failed_result_after_retry_exhaustion(self):
+        archiver = WorkflowRunArchiver(days=90)
+        run = _run()
+        session_maker = MagicMock(
+            side_effect=[
+                _session_context(MagicMock(name="session-1")),
+                _session_context(MagicMock(name="session-2")),
+                _session_context(MagicMock(name="session-3")),
+            ]
+        )
+
+        with (
+            patch.object(archiver, "_archive_bundle", side_effect=[_db_disconnect_error()] * 3) as archive_bundle,
+            patch("services.retention.workflow_run.db_retry.time.sleep") as sleep,
+        ):
+            results = archiver._archive_bundle_groups(session_maker, MagicMock(), [[run]])
+
+        assert len(results) == 1
+        assert results[0].success is False
+        assert "server closed the connection unexpectedly" in (results[0].error or "")
+        assert archive_bundle.call_count == archiver.DB_RETRY_ATTEMPTS
+        assert session_maker.call_count == archiver.DB_RETRY_ATTEMPTS
+        assert sleep.call_count == archiver.DB_RETRY_ATTEMPTS - 1
+
+    def test_archive_bundle_uses_safe_rollback_when_failure_rolls_back_badly(self):
+        archiver = WorkflowRunArchiver(days=90, dry_run=True)
+        session = MagicMock()
+        session.rollback.side_effect = RuntimeError("rollback failed")
+
+        with patch.object(archiver, "_extract_bundle_data", side_effect=RuntimeError("extract failed")):
+            result = archiver._archive_bundle(session, None, [_run()])
+
+        assert result.success is False
+        assert result.error == "extract failed"
+        session.rollback.assert_called_once()
+
+
 class TestArchiveRunIdempotency:
    def _index_payload(self, archiver: WorkflowRunArchiver, run_ids: list[str], run) -> tuple[str, bytes]:
        identity = archiver._build_bundle_identity([run])
@@ -401,37 +518,6 @@ class TestArchiveRunIdempotency:
        assert result.skipped is True
        assert result.error == "bundle already archived"

-    def test_successful_bundle_persists_archive_index(self):
-        archiver = WorkflowRunArchiver(days=90)
-        run = MagicMock()
-        run.id = str(uuid.uuid4())
-        run.tenant_id = str(uuid.uuid4())
-        run.created_at = datetime.datetime(2025, 3, 15, 10, 0, 0)
-        session = MagicMock()
-        session.scalar.return_value = None
-        storage = MagicMock()
-        storage.object_exists.return_value = False
-        table_data = {
-            "workflow_runs": [{"id": run.id, "tenant_id": run.tenant_id}],
-            "workflow_node_executions": [{"id": str(uuid.uuid4()), "workflow_run_id": run.id}],
-        }
-
-        with (
-            patch.object(archiver, "_lock_runs_for_archive", return_value=[run]),
-            patch.object(archiver, "_extract_bundle_data", return_value=table_data),
-        ):
-            result = archiver._archive_bundle(session, storage, [run])
-
-        archived_bundle = session.add.call_args.args[0]
-        assert result.success is True
-        assert isinstance(archived_bundle, WorkflowRunArchiveBundle)
-        assert archived_bundle.tenant_id == run.tenant_id
-        assert archived_bundle.year == 2025
-        assert archived_bundle.month == 3
-        assert archived_bundle.workflow_run_count == 1
-        assert archived_bundle.row_count == 2
-        session.commit.assert_called_once()
-
    def test_index_skips_all_already_archived_runs(self):
        archiver = WorkflowRunArchiver(days=90)
        run = MagicMock()
@@ -201,9 +201,13 @@ class TestWebhookServiceLookupWithContainers:
            db_session_with_containers, app=app, node_id="node-1", status=AppTriggerStatus.RATE_LIMITED
        )

-        with pytest.raises(ValueError, match="rate limited"):
+        with pytest.raises(QuotaExceededError) as exc_info:
            WebhookService.get_webhook_trigger_and_workflow(webhook_trigger.webhook_id)

+        assert exc_info.value.feature == QuotaType.TRIGGER.value
+        assert exc_info.value.tenant_id == tenant.id
+        assert exc_info.value.required == 1
+
    def test_get_webhook_trigger_and_workflow_raises_when_app_trigger_disabled(
        self, db_session_with_containers: Session, flask_app_with_containers: Flask
    ):
@@ -0,0 +1,139 @@
+import datetime
+from unittest.mock import MagicMock
+
+import click
+import pytest
+from sqlalchemy.exc import OperationalError
+
+from commands import retention
+
+
+def _db_disconnect_error() -> OperationalError:
+    return OperationalError(
+        "select 1",
+        {},
+        RuntimeError("server closed the connection unexpectedly"),
+        connection_invalidated=True,
+    )
+
+
+def _session_context(session):
+    context = MagicMock()
+    context.__enter__.return_value = session
+    context.__exit__.return_value = False
+    return context
+
+
+def test_resolve_archive_tenant_ids_from_plan_uses_explicit_sessions(monkeypatch):
+    end_before = datetime.datetime(2025, 4, 1, tzinfo=datetime.UTC)
+    sessions = [MagicMock(name="session-a"), MagicMock(name="session-b")]
+    session_maker = MagicMock(side_effect=[_session_context(sessions[0]), _session_context(sessions[1])])
+    calls = []
+
+    def get_candidate_tenants(session, prefix, *, start_from, end_before):
+        calls.append((session, prefix, start_from, end_before))
+        return [f"{prefix}-paid", f"{prefix}-free"]
+
+    monkeypatch.setattr(retention, "_get_archive_candidate_tenant_ids_by_prefix", get_candidate_tenants)
+    monkeypatch.setattr(
+        retention,
+        "_filter_paid_workflow_archive_tenant_ids",
+        lambda tenant_ids: (["a-paid", "b-paid"], ["a-free", "b-free"]),
+    )
+
+    tenant_plan = retention._resolve_archive_tenant_ids_from_plan(
+        session_maker=session_maker,
+        tenant_ids=None,
+        tenant_prefixes=["a", "b"],
+        start_from=None,
+        end_before=end_before,
+    )
+
+    assert tenant_plan["archive_tenant_ids"] == ["a-paid", "b-paid"]
+    assert tenant_plan["paid_tenant_ids"] == ["a-paid", "b-paid"]
+    assert tenant_plan["unpaid_tenant_ids"] == ["a-free", "b-free"]
+    assert calls == [
+        (sessions[0], "a", None, end_before),
+        (sessions[1], "b", None, end_before),
+    ]
+
+
+def test_safe_remove_scoped_session_discards_registry_and_disposes_after_remove_error(monkeypatch):
+    fake_db = MagicMock()
+    fake_db.session.remove.side_effect = RuntimeError("server closed the connection unexpectedly")
+    monkeypatch.setattr(retention, "db", fake_db)
+
+    retention._safe_remove_scoped_session("archive workflow run command")
+
+    fake_db.session.remove.assert_called_once()
+    fake_db.session.registry.clear.assert_called_once()
+    fake_db.engine.dispose.assert_called_once()
+
+
+def test_archive_command_db_retry_retries_retryable_db_disconnect(monkeypatch):
+    operation = MagicMock(side_effect=[_db_disconnect_error(), "ok"])
+    sleep = MagicMock()
+    monkeypatch.setattr("services.retention.workflow_run.db_retry.time.sleep", sleep)
+
+    result = retention._run_archive_command_db_retry("archive plan", operation)
+
+    assert result == "ok"
+    assert operation.call_count == 2
+    sleep.assert_called_once_with(1.0)
+
+
+def test_archive_plan_prefix_stats_retries_count_query_with_fresh_session(monkeypatch):
+    end_before = datetime.datetime(2025, 4, 1, tzinfo=datetime.UTC)
+    sessions = [MagicMock(name="session-1"), MagicMock(name="session-2")]
+    sessions[0].scalar.side_effect = _db_disconnect_error()
+    sessions[1].scalar.side_effect = [7, 9]
+    session_maker = MagicMock(side_effect=[_session_context(sessions[0]), _session_context(sessions[1])])
+    sleep = MagicMock()
+
+    monkeypatch.setattr(
+        retention,
+        "_get_archive_candidate_tenant_ids_by_prefix",
+        lambda session, prefix, *, start_from, end_before: [f"{prefix}-tenant"],
+    )
+    monkeypatch.setattr("services.retention.workflow_run.db_retry.time.sleep", sleep)
+
+    stats = retention._get_archive_plan_prefix_stats(
+        session_maker,
+        "a",
+        start_from=None,
+        end_before=end_before,
+    )
+
+    assert stats["tenant_ids"] == ["a-tenant"]
+    assert stats["workflow_runs"] == 7
+    assert stats["workflow_node_executions"] == 9
+    assert session_maker.call_count == 2
+    sleep.assert_called_once_with(1.0)
+
+
+def test_archive_workflow_runs_raises_click_exception_when_tenant_plan_fails(monkeypatch):
+    fake_db = MagicMock()
+    monkeypatch.setattr(retention, "db", fake_db)
+    monkeypatch.setattr(
+        retention,
+        "_resolve_archive_tenant_ids_from_plan",
+        MagicMock(side_effect=RuntimeError("tenant plan failed")),
+    )
+
+    with pytest.raises(click.ClickException, match="Failed to resolve workflow archive tenant plan"):
+        retention.archive_workflow_runs.callback(
+            tenant_ids="tenant-1",
+            tenant_prefixes=None,
+            before_days=90,
+            from_days_ago=None,
+            to_days_ago=None,
+            start_from=None,
+            end_before=None,
+            batch_size=10000,
+            workers=1,
+            run_shard_index=None,
+            run_shard_total=None,
+            limit=None,
+            dry_run=True,
+            delete_after_archive=False,
+        )
@@ -13,11 +13,15 @@ from pathlib import Path
 import pytest

 REPO_ROOT = Path(__file__).resolve().parents[4]
-SCRIPT_PATH = REPO_ROOT / "scripts" / "check_no_new_getattr.py"
+SCRIPTS_DIR = REPO_ROOT / "scripts"
+SCRIPT_PATH = SCRIPTS_DIR / "check_no_new_getattr.py"
+GUARD_HELPER_PATH = SCRIPTS_DIR / "ast_grep_guard.py"


 def load_guard_module() -> types.ModuleType:
-    spec = importlib.util.spec_from_file_location("check_no_new_getattr_under_test", SCRIPT_PATH)
+    if str(SCRIPTS_DIR) not in sys.path:
+        sys.path.insert(0, str(SCRIPTS_DIR))
+    spec = importlib.util.spec_from_file_location("ast_grep_guard_under_test", GUARD_HELPER_PATH)
    assert spec is not None
    assert spec.loader is not None
    module = importlib.util.module_from_spec(spec)
@@ -1,7 +1,15 @@
+"""CLI migration tests using real command-owned SQLite sessions.
+
+Migration services and package I/O remain fakes so these cases can focus on
+CLI parsing and session lifecycle without fabricating the SQLAlchemy boundary.
+"""
+
 import json
 from pathlib import Path

 from click.testing import CliRunner
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session

 from commands import data_migration
 from commands.data_migration import (
@@ -20,24 +28,6 @@ from services.data_migration.entities import (
 )


-class FakeSessionContext:
-    session: object
-    entered: bool
-    exited: bool
-
-    def __init__(self, session: object) -> None:
-        self.session = session
-        self.entered = False
-        self.exited = False
-
-    def __enter__(self) -> object:
-        self.entered = True
-        return self.session
-
-    def __exit__(self, *_args: object) -> None:
-        self.exited = True
-
-
 def test_export_command_requires_input_and_output():
    result = CliRunner().invoke(export_migration_data, [])

@@ -99,9 +89,7 @@ def test_export_template_command_requires_overwrite_for_existing_output(tmp_path
    assert "already exists" in result.output


-def test_export_command_uses_cli_owned_session(monkeypatch, tmp_path: Path):
-    session = object()
-    session_context = FakeSessionContext(session)
+def test_export_command_uses_cli_owned_session(monkeypatch, tmp_path: Path, sqlite_engine: Engine):
    captured: dict[str, object] = {}
    input_file = tmp_path / "export-config.json"
    output_file = tmp_path / "migration-package.json"
@@ -120,7 +108,7 @@ def test_export_command_uses_cli_owned_session(monkeypatch, tmp_path: Path):
            captured["path"] = path
            captured["overwrite"] = overwrite

-    monkeypatch.setattr(data_migration.session_factory, "create_session", lambda: session_context)
+    monkeypatch.setattr(data_migration.session_factory, "create_session", lambda: Session(sqlite_engine))
    monkeypatch.setattr(data_migration, "MigrationExportService", FakeMigrationExportService)
    monkeypatch.setattr(data_migration, "MigrationPackageService", FakeMigrationPackageService)

@@ -130,17 +118,16 @@ def test_export_command_uses_cli_owned_session(monkeypatch, tmp_path: Path):
    )

    assert result.exit_code == 0
-    assert captured["session"] is session
+    captured_session = captured["session"]
+    assert isinstance(captured_session, Session)
+    assert captured_session.get_bind() is sqlite_engine
+    assert captured_session.in_transaction() is False
    assert captured["package"] is package
    assert captured["path"] == str(output_file)
    assert captured["overwrite"] is False
-    assert session_context.entered
-    assert session_context.exited


-def test_import_command_uses_cli_owned_session(monkeypatch, tmp_path: Path):
-    session = object()
-    session_context = FakeSessionContext(session)
+def test_import_command_uses_cli_owned_session(monkeypatch, tmp_path: Path, sqlite_engine: Engine):
    captured: dict[str, object] = {}
    input_file = tmp_path / "migration-package.json"
    input_file.write_text("{}")
@@ -166,7 +153,7 @@ def test_import_command_uses_cli_owned_session(monkeypatch, tmp_path: Path):
            captured["path"] = path
            return package

-    monkeypatch.setattr(data_migration.session_factory, "create_session", lambda: session_context)
+    monkeypatch.setattr(data_migration.session_factory, "create_session", lambda: Session(sqlite_engine))
    monkeypatch.setattr(data_migration, "MigrationImportService", FakeMigrationImportService)
    monkeypatch.setattr(data_migration, "MigrationPackageService", FakeMigrationPackageService)

@@ -176,10 +163,11 @@ def test_import_command_uses_cli_owned_session(monkeypatch, tmp_path: Path):
    )

    assert result.exit_code == 0
-    assert captured["session"] is session
+    captured_session = captured["session"]
+    assert isinstance(captured_session, Session)
+    assert captured_session.get_bind() is sqlite_engine
+    assert captured_session.in_transaction() is False
    assert captured["path"] == str(input_file)
    request = captured["request"]
    assert request.package is package
    assert request.options_override == ImportOptions(conflict_strategy=ConflictStrategy.SKIP)
-    assert session_context.entered
-    assert session_context.exited
@@ -75,6 +75,7 @@ def test_dify_config(monkeypatch: pytest.MonkeyPatch):
    # default values
    assert config.EDITION == "SELF_HOSTED"
    assert config.API_COMPRESSION_ENABLED is False
+    assert config.AGENT_SHELL_ENABLED is True
    assert config.SENTRY_TRACES_SAMPLE_RATE == 1.0
    assert config.TEMPLATE_TRANSFORM_MAX_LENGTH == 400_000

@@ -110,6 +111,25 @@ def test_http_timeout_defaults(monkeypatch: pytest.MonkeyPatch):
    assert config.HTTP_REQUEST_MAX_WRITE_TIMEOUT == 600


+def test_internal_files_url_falls_back_to_server_console_api_url(monkeypatch: pytest.MonkeyPatch):
+    os.environ.clear()
+    monkeypatch.setenv("SERVER_CONSOLE_API_URL", "http://api:5001")
+
+    config = DifyConfig(_env_file=None)
+
+    assert config.INTERNAL_FILES_URL == "http://api:5001"
+
+
+def test_internal_files_url_prefers_explicit_value(monkeypatch: pytest.MonkeyPatch):
+    os.environ.clear()
+    monkeypatch.setenv("INTERNAL_FILES_URL", "http://files-internal:5001")
+    monkeypatch.setenv("SERVER_CONSOLE_API_URL", "http://api:5001")
+
+    config = DifyConfig(_env_file=None)
+
+    assert config.INTERNAL_FILES_URL == "http://files-internal:5001"
+
+
 # NOTE: If there is a `.env` file in your Workspace, this test might not succeed as expected.
 # This is due to `pymilvus` loading all the variables from the `.env` file into `os.environ`.
 def test_flask_configs(monkeypatch: pytest.MonkeyPatch):
@@ -556,6 +556,7 @@ def test_app_create_api_attaches_permission_keys(app, app_module):

    with app.test_request_context("/apps", method="POST", json={}):
        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setattr(app_module.dify_config, "RBAC_ENABLED", True)
            app_module.console_ns.payload = {
                "name": "Created App",
                "description": "Summary",
@@ -571,11 +572,25 @@ def test_app_create_api_attaches_permission_keys(app, app_module):
                "batch_get",
                lambda tenant_id, account_id, app_ids, session: {"app-new": ["app.acl.view_layout", "app.acl.edit"]},
            )
+            initialize_rbac_task = MagicMock()
+            monkeypatch.setattr(
+                app_module,
+                "initialize_created_app_rbac_access_task",
+                initialize_rbac_task,
+            )
+            replace_whitelist = MagicMock()
+            monkeypatch.setattr(
+                app_module.enterprise_rbac_service.RBACService.AppAccess,
+                "replace_whitelist",
+                replace_whitelist,
+            )

            resp, status = method(app_module.AppListApi(), "tenant-1", SimpleNamespace(id="acct-1"))

    assert status == 201
    assert resp["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"]
+    assert replace_whitelist.call_args.kwargs["payload"].scope is app_module.RBACResourceWhitelistScope.ALL
+    initialize_rbac_task.delay.assert_called_once_with("tenant-1", "acct-1", "app-new")


 def test_app_list_api_attaches_permission_keys(app, app_module):
@@ -1,3 +1,4 @@
+import uuid
 from inspect import unwrap
 from unittest.mock import MagicMock, PropertyMock, patch

@@ -310,6 +311,37 @@ class TestChatApi:
            with pytest.raises(completion_module.NotFound):
                method(api, MagicMock(), user, chat_app)

+    def test_invalid_conversation_id_fails_fast_as_not_found(self, app: Flask, chat_app, user) -> None:
+        # A nonexistent conversation_id must fail fast as 404, before the streaming
+        # generator is created. Previously the lookup only ran inside the generator,
+        # so an invalid id surfaced as a hang instead of a clean error.
+        payload_patch = patch.object(
+            type(completion_module.console_ns),
+            "payload",
+            new_callable=PropertyMock,
+            return_value={"inputs": {}, "query": "hi", "conversation_id": str(uuid.uuid4())},
+        )
+        generate_mock = MagicMock(return_value={"ok": True})
+
+        api = completion_module.ChatApi()
+        method = unwrap(api.post)
+
+        with (
+            app.test_request_context("/", json={}),
+            payload_patch,
+            patch.object(
+                completion_module.ConversationService,
+                "get_conversation",
+                side_effect=completion_module.services.errors.conversation.ConversationNotExistsError(),
+            ),
+            patch.object(completion_module.AppGenerateService, "generate", generate_mock),
+        ):
+            with pytest.raises(completion_module.NotFound):
+                method(api, MagicMock(), user, chat_app)
+
+        # The lookup must run before generation, so the generator is never started.
+        generate_mock.assert_not_called()
+
    def test_app_unavailable_chat(self, app: Flask, chat_app, user, payload_patch):
        api = completion_module.ChatApi()
        method = unwrap(api.post)
@@ -49,6 +49,8 @@ def make_message():
    msg.query = "hello"
    msg.re_sign_file_url_answer = ""
    msg.user_feedback = MagicMock(rating=None)
+    msg.total_price = None
+    msg.currency = None
    msg.status = "normal"
    msg.error = None
    return msg
@@ -262,6 +262,25 @@ class TestPaginationMapping:


 class TestResourceAccessScopeBindings:
+    def test_app_whitelist_all_schedules_member_policy_sync(self, app):
+        with (
+            app.test_request_context(
+                "/workspaces/current/rbac/apps/app-1/whitelist",
+                method="PUT",
+                json={"scope": "all"},
+            ),
+            patch("controllers.console.workspace.rbac._current_ids", return_value=("tenant-1", "acct-actor")),
+            patch("controllers.console.workspace.rbac.dify_config.RBAC_ENABLED", True),
+            patch(
+                "controllers.console.workspace.rbac.svc.RBACService.AppAccess.replace_whitelist",
+                return_value=rbac_mod.svc.ResourceWhitelist(),
+            ),
+            patch("controllers.console.workspace.rbac.initialize_created_app_rbac_access_task") as mock_sync_task,
+        ):
+            inspect.unwrap(rbac_mod.RBACAppWhitelistApi.put)(rbac_mod.RBACAppWhitelistApi(), "app-1")
+
+        mock_sync_task.delay.assert_called_once_with("tenant-1", "acct-actor", "app-1")
+
    def test_app_user_access_policy_assignment_forwards_ids(self, app):
        with (
            app.test_request_context(
@@ -34,11 +34,11 @@ class DummyFile:


 class DummyToolFile:
-    def __init__(self):
+    def __init__(self, name="test.txt", mimetype="text/plain"):
        self.id = "file-id"
-        self.name = "test.txt"
+        self.name = name
        self.size = 10
-        self.mimetype = "text/plain"
+        self.mimetype = mimetype
        self.original_url = "http://original"
        self.user_id = "user-1"
        self.tenant_id = "tenant-1"
@@ -56,7 +56,7 @@ class TestPluginUploadFileApi:
        mock_get_user,
        mock_verify_signature,
    ):
-        dummy_file = DummyFile()
+        dummy_file = DummyFile(filename="report.docx", mimetype="application/octet-stream")

        module.request = fake_request(
            {
@@ -71,7 +71,10 @@ class TestPluginUploadFileApi:
        )

        tool_file_manager_instance = mock_tool_file_manager.return_value
-        tool_file_manager_instance.create_file_by_raw.return_value = DummyToolFile()
+        tool_file_manager_instance.create_file_by_raw.return_value = DummyToolFile(
+            name="report.docx",
+            mimetype="application/octet-stream",
+        )

        mock_tool_file_manager.sign_file.return_value = "signed-url"

@@ -84,10 +87,12 @@ class TestPluginUploadFileApi:
        assert result["id"] == "file-id"
        assert result["reference"] == build_file_reference(record_id="file-id")
        assert result["preview_url"] == "signed-url"
+        assert result["extension"] == ".docx"
        mock_verify_signature.assert_called_once()
        assert mock_verify_signature.call_args.kwargs["conversation_id"] == "conversation-1"
        tool_file_manager_instance.create_file_by_raw.assert_called_once()
        assert tool_file_manager_instance.create_file_by_raw.call_args.kwargs["conversation_id"] == "conversation-1"
+        mock_tool_file_manager.sign_file.assert_called_once_with(tool_file_id="file-id", extension=".docx")

    def test_missing_file(self):
        module.request = fake_request(
@@ -318,6 +318,7 @@ class TestPluginDownloadFileRequestApi:
        mock_payload.user_id = "user-id"
        mock_payload.user_from = "account"
        mock_payload.invoke_from = "debugger"
+        mock_payload.for_external = False
        reference = build_file_reference(record_id="tool-file-1")
        mock_payload.file.model_dump.return_value = {
            "transfer_method": "tool_file",
@@ -333,6 +334,7 @@ class TestPluginDownloadFileRequestApi:
            user_from="account",
            invoke_from="debugger",
            file_mapping={"transfer_method": "tool_file", "reference": reference},
+            for_external=False,
        )
        assert result["data"] == {
            "filename": "report.pdf",
@@ -42,6 +42,7 @@ from graphon.model_runtime.errors.invoke import InvokeError
 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.conversation import ConversationNotExistsError
 from services.errors.llm import InvokeRateLimitError
@@ -554,6 +555,35 @@ class TestChatApiController:
            with pytest.raises(AgentNotPublishedError):
                handler(api, session=Mock(), app_model=app_model, end_user=end_user)

+    def test_invalid_conversation_id_fails_fast_as_not_found(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
+        # A well-formed but nonexistent conversation_id must fail fast as 404, before the
+        # streaming generator is created. Previously the lookup only ran inside the generator,
+        # so an invalid id surfaced as a hang instead of a clean error.
+        monkeypatch.setattr(
+            ConversationService,
+            "get_conversation",
+            lambda *_args, **_kwargs: (_ for _ in ()).throw(ConversationNotExistsError()),
+        )
+
+        generate_mock = Mock(return_value={"text": "unused"})
+        monkeypatch.setattr(AppGenerateService, "generate", generate_mock)
+
+        api = ChatApi()
+        handler = unwrap(api.post)
+        app_model = SimpleNamespace(mode=AppMode.CHAT.value, id="app-1")
+        end_user = SimpleNamespace()
+
+        with app.test_request_context(
+            "/chat-messages",
+            method="POST",
+            json={"inputs": {}, "query": "hi", "conversation_id": str(uuid.uuid4())},
+        ):
+            with pytest.raises(NotFound):
+                handler(api, session=Mock(), app_model=app_model, end_user=end_user)
+
+        # The lookup must run before generation, so the generator is never started.
+        generate_mock.assert_not_called()
+

 class TestChatStopApiController:
    def test_wrong_mode(self, app: Flask) -> None:
@@ -26,7 +26,7 @@ from flask import Flask
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, NotFound

-from controllers.service_api.app.error import NotWorkflowAppError
+from controllers.service_api.app.error import NotWorkflowAppError, WorkflowVersionExecutionNotAllowedError
 from controllers.service_api.app.workflow import (
    AppQueueManager,
    DifyAPIRepositoryFactory,
@@ -42,11 +42,13 @@ from controllers.service_api.app.workflow import (
 )
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from core.app.entities.app_invoke_entities import InvokeFrom
+from enums.cloud_plan import CloudPlan
 from graphon.enums import WorkflowExecutionStatus
 from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom
 from models.model import App, AppMode, EndUser
 from models.workflow import WorkflowAppLog, WorkflowAppLogCreatedFrom, WorkflowRun, WorkflowType
 from services.app_generate_service import AppGenerateService
+from services.billing_service import BillingService
 from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError
 from services.workflow_app_service import LogView, LogViewDetails, WorkflowAppService
@@ -570,9 +572,117 @@ class TestWorkflowRunApi:
            with pytest.raises(InvokeRateLimitHttpError):
                handler(api, session=Mock(), app_model=app_model, end_user=end_user)

+    def test_sandbox_billing_does_not_gate_default_workflow_run(
+        self, app: Flask, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        workflow_module = sys.modules["controllers.service_api.app.workflow"]
+        monkeypatch.setattr(workflow_module.dify_config, "BILLING_ENABLED", True)
+
+        billing_get_info = Mock(return_value={"enabled": True, "subscription": {"plan": CloudPlan.SANDBOX}})
+        generate = Mock(return_value={"result": "ok"})
+        monkeypatch.setattr(BillingService, "get_info", billing_get_info)
+        monkeypatch.setattr(AppGenerateService, "generate", generate)
+
+        api = WorkflowRunApi()
+        handler = unwrap(api.post)
+
+        with app.test_request_context("/workflows/run", method="POST", json={"inputs": {}}):
+            response = handler(
+                api,
+                session=Mock(),
+                app_model=_make_app_model(),
+                end_user=_make_end_user(),
+            )
+
+        assert response.get_json() == {"result": "ok"}
+        billing_get_info.assert_not_called()
+        generate.assert_called_once()
+

 class TestWorkflowRunByIdApi:
+    def test_rejects_sandbox_plan_with_upgrade_error(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
+        workflow_module = sys.modules["controllers.service_api.app.workflow"]
+        monkeypatch.setattr(workflow_module.dify_config, "BILLING_ENABLED", True)
+
+        billing_get_info = Mock(return_value={"enabled": True, "subscription": {"plan": CloudPlan.SANDBOX}})
+        generate = Mock()
+        monkeypatch.setattr(BillingService, "get_info", billing_get_info)
+        monkeypatch.setattr(AppGenerateService, "generate", generate)
+
+        api = WorkflowRunByIdApi()
+        handler = unwrap(api.post)
+        app_model = _make_app_model()
+
+        with app.test_request_context("/workflows/w1/run", method="POST", json={"inputs": {}}):
+            with pytest.raises(WorkflowVersionExecutionNotAllowedError) as exc_info:
+                handler(
+                    api,
+                    session=Mock(),
+                    app_model=app_model,
+                    end_user=_make_end_user(),
+                    workflow_id="w1",
+                )
+
+        billing_get_info.assert_called_once_with(app_model.tenant_id, exclude_vector_space=True)
+        generate.assert_not_called()
+        assert exc_info.value.code == 403
+        assert exc_info.value.error_code == "workflow_version_execution_not_allowed"
+        assert exc_info.value.description == (
+            "Workflow version execution is not available on your current plan. Please upgrade to a paid plan."
+        )
+        assert exc_info.value.data == {
+            "code": "workflow_version_execution_not_allowed",
+            "message": exc_info.value.description,
+            "status": 403,
+        }
+
+    @pytest.mark.parametrize(
+        ("billing_config_enabled", "billing_enabled", "plan"),
+        [
+            (False, True, CloudPlan.SANDBOX),
+            (True, False, CloudPlan.SANDBOX),
+            (True, True, CloudPlan.PROFESSIONAL),
+        ],
+    )
+    def test_allows_execution_outside_enabled_sandbox_plan(
+        self,
+        app: Flask,
+        monkeypatch: pytest.MonkeyPatch,
+        billing_config_enabled: bool,
+        billing_enabled: bool,
+        plan: CloudPlan,
+    ) -> None:
+        workflow_module = sys.modules["controllers.service_api.app.workflow"]
+        monkeypatch.setattr(workflow_module.dify_config, "BILLING_ENABLED", billing_config_enabled)
+
+        billing_get_info = Mock(return_value={"enabled": billing_enabled, "subscription": {"plan": plan}})
+        generate = Mock(return_value={"result": "ok"})
+        monkeypatch.setattr(BillingService, "get_info", billing_get_info)
+        monkeypatch.setattr(AppGenerateService, "generate", generate)
+
+        api = WorkflowRunByIdApi()
+        handler = unwrap(api.post)
+        app_model = _make_app_model()
+
+        with app.test_request_context("/workflows/w1/run", method="POST", json={"inputs": {}}):
+            response = handler(
+                api,
+                session=Mock(),
+                app_model=app_model,
+                end_user=_make_end_user(),
+                workflow_id="w1",
+            )
+
+        assert response.get_json() == {"result": "ok"}
+        generate.assert_called_once()
+        if billing_config_enabled:
+            billing_get_info.assert_called_once_with(app_model.tenant_id, exclude_vector_space=True)
+        else:
+            billing_get_info.assert_not_called()
+
    def test_not_found(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
+        workflow_module = sys.modules["controllers.service_api.app.workflow"]
+        monkeypatch.setattr(workflow_module.dify_config, "BILLING_ENABLED", False)
        monkeypatch.setattr(
            AppGenerateService,
            "generate",
@@ -589,6 +699,8 @@ class TestWorkflowRunByIdApi:
                handler(api, session=Mock(), app_model=app_model, end_user=end_user, workflow_id="w1")

    def test_draft_workflow(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
+        workflow_module = sys.modules["controllers.service_api.app.workflow"]
+        monkeypatch.setattr(workflow_module.dify_config, "BILLING_ENABLED", False)
        monkeypatch.setattr(
            AppGenerateService,
            "generate",
@@ -5,6 +5,7 @@ import pytest
 from werkzeug.exceptions import NotFound, RequestEntityTooLarge

 import controllers.trigger.webhook as module
+from services.errors.app import QuotaExceededError


@pytest.fixture(autouse=True)
@@ -83,6 +84,58 @@ class TestHandleWebhook:
        assert status == 400
        assert response["error"] == "Bad Request"

+    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
+    @patch.object(module.WebhookService, "extract_and_validate_webhook_data")
+    @patch.object(
+        module.WebhookService,
+        "trigger_workflow_execution",
+        side_effect=QuotaExceededError(feature="trigger", tenant_id="tenant-1", required=1),
+    )
+    def test_quota_exceeded(self, mock_trigger, mock_extract, mock_get):
+        mock_get.return_value = (DummyWebhookTrigger(), "workflow", "node_config")
+        mock_extract.return_value = {"input": "x"}
+
+        response, status = module.handle_webhook("wh-1")
+
+        assert status == 429
+        assert response == {
+            "error": "Too Many Requests",
+            "message": "Trigger event quota exceeded. Please upgrade your plan.",
+        }
+
+    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow")
+    @patch.object(module.WebhookService, "extract_and_validate_webhook_data")
+    @patch.object(
+        module.WebhookService,
+        "trigger_workflow_execution",
+        side_effect=QuotaExceededError(feature="workflow", tenant_id="tenant-1", required=1),
+    )
+    def test_workflow_quota_exceeded(self, mock_trigger, mock_extract, mock_get):
+        mock_get.return_value = (DummyWebhookTrigger(), "workflow", "node_config")
+        mock_extract.return_value = {"input": "x"}
+
+        response, status = module.handle_webhook("wh-1")
+
+        assert status == 429
+        assert response == {
+            "error": "Too Many Requests",
+            "message": "Workflow execution quota exceeded. Please upgrade your plan.",
+        }
+
+    @patch.object(
+        module.WebhookService,
+        "get_webhook_trigger_and_workflow",
+        side_effect=QuotaExceededError(feature="trigger", tenant_id="tenant-1", required=1),
+    )
+    def test_rate_limited(self, mock_get):
+        response, status = module.handle_webhook("wh-1")
+
+        assert status == 429
+        assert response == {
+            "error": "Too Many Requests",
+            "message": "Trigger event quota exceeded. Please upgrade your plan.",
+        }
+
    @patch.object(module.WebhookService, "get_webhook_trigger_and_workflow", side_effect=ValueError("missing"))
    def test_value_error_not_found(self, mock_get):
        with pytest.raises(NotFound):
@@ -37,6 +37,8 @@ from pydantic_ai.messages import (
 from clients.agent_backend import (
    AgentBackendError,
    AgentBackendRunEventAdapter,
+    AgentBackendRunFailedError,
+    AgentBackendRunFailedInternalEvent,
    AgentBackendStreamInternalEvent,
    FakeAgentBackendRunClient,
    FakeAgentBackendScenario,
@@ -54,6 +56,7 @@ from core.app.entities.queue_entities import (
    QueueMessageEndEvent,
 )
 from core.workflow.nodes.agent_v2.ask_human_resume import AskHumanResumeOutcome
+from graphon.model_runtime.errors.invoke import InvokeRateLimitError
 from models.agent_config_entities import AgentSoulConfig
 from models.model import MessageAgentThought

@@ -1039,6 +1042,130 @@ def test_tool_result_without_call_id_matches_unique_open_tool_name(monkeypatch):
    assert rows[0].observation == "Knowledge base search results: browser skill"


+def test_repeated_tool_calls_without_call_id_or_index_create_distinct_rows(monkeypatch):
+    fake_session = _FakeDbSession()
+    monkeypatch.setattr(app_runner_module.db, "session", fake_session)
+    qm = _FakeQueueManager()
+    recorder = app_runner_module._AgentProcessRecorder(
+        dify_context=_dify_ctx(),
+        message_id="msg-1",
+        queue_manager=qm,  # type: ignore[arg-type]
+    )
+
+    recorder.handle_stream_event(
+        AgentBackendStreamInternalEvent(
+            run_id="run-1",
+            data={
+                "event_kind": "function_tool_call",
+                "part": {
+                    "part_kind": "tool-call",
+                    "tool_name": "shell_run",
+                    "args": {"script": "lookup find"},
+                },
+            },
+        )
+    )
+    recorder.handle_stream_event(
+        AgentBackendStreamInternalEvent(
+            run_id="run-1",
+            data={
+                "event_kind": "function_tool_result",
+                "part": {
+                    "part_kind": "tool-return",
+                    "tool_name": "shell_run",
+                    "content": "find output",
+                },
+            },
+        )
+    )
+    recorder.handle_stream_event(
+        AgentBackendStreamInternalEvent(
+            run_id="run-1",
+            data={
+                "event_kind": "function_tool_call",
+                "part": {
+                    "part_kind": "tool-call",
+                    "tool_name": "shell_run",
+                    "args": {"script": "lookup out"},
+                },
+            },
+        )
+    )
+    recorder.handle_stream_event(
+        AgentBackendStreamInternalEvent(
+            run_id="run-1",
+            data={
+                "event_kind": "function_tool_result",
+                "part": {
+                    "part_kind": "tool-return",
+                    "tool_name": "shell_run",
+                    "content": "out output",
+                },
+            },
+        )
+    )
+
+    rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
+    assert len(rows) == 2
+    assert rows[0].tool == "shell_run"
+    assert rows[0].tool_input == '{"script": "lookup find"}'
+    assert rows[0].observation == "find output"
+    assert rows[1].tool == "shell_run"
+    assert rows[1].tool_input == '{"script": "lookup out"}'
+    assert rows[1].observation == "out output"
+
+
+def test_repeated_tool_calls_with_placeholder_call_id_and_reused_index_create_distinct_rows(monkeypatch):
+    fake_session = _FakeDbSession()
+    monkeypatch.setattr(app_runner_module.db, "session", fake_session)
+    qm = _FakeQueueManager()
+    recorder = app_runner_module._AgentProcessRecorder(
+        dify_context=_dify_ctx(),
+        message_id="msg-1",
+        queue_manager=qm,  # type: ignore[arg-type]
+    )
+
+    for script, output in (("lookup find", "find output"), ("lookup out", "out output")):
+        recorder.handle_stream_event(
+            AgentBackendStreamInternalEvent(
+                run_id="run-1",
+                data={
+                    "event_kind": "function_tool_call",
+                    "index": 0,
+                    "part": {
+                        "part_kind": "tool-call",
+                        "tool_name": "shell_run",
+                        "tool_call_id": "None",
+                        "args": {"script": script},
+                    },
+                },
+            )
+        )
+        recorder.handle_stream_event(
+            AgentBackendStreamInternalEvent(
+                run_id="run-1",
+                data={
+                    "event_kind": "function_tool_result",
+                    "part": {
+                        "part_kind": "tool-return",
+                        "tool_name": "shell_run",
+                        "tool_call_id": "None",
+                        "content": output,
+                    },
+                },
+            )
+        )
+
+    rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
+    assert len(rows) == 2
+    assert rows[0].tool == "shell_run"
+    assert rows[0].tool_input == '{"script": "lookup find"}'
+    assert rows[0].observation == "find output"
+    assert rows[1].tool == "shell_run"
+    assert rows[1].tool_input == '{"script": "lookup out"}'
+    assert rows[1].observation == "out output"
+
+
 def test_prior_session_snapshot_is_threaded_into_request():
    prior = CompositorSessionSnapshot(layers=[])
    client = FakeAgentBackendRunClient()
@@ -1081,13 +1208,48 @@ def test_failed_run_raises_agent_backend_error():
    store = _FakeSessionStore()
    qm = _FakeQueueManager()

-    with pytest.raises(AgentBackendError):
+    with pytest.raises(AgentBackendRunFailedError, match="fake failure .*agent_run_id=fake-run-1"):
        _run(_runner(client, store), qm)
    # No message-end on failure; no snapshot saved.
    assert not [e for e in qm.events if isinstance(e, QueueMessageEndEvent)]
    assert store.saved == []


+def test_agent_backend_failure_to_exception_maps_rate_limit_reason():
+    err = app_runner_module._agent_backend_failure_to_exception(
+        AgentBackendRunFailedInternalEvent(
+            run_id="run-1",
+            error="quota exceeded",
+            reason="InvokeRateLimitError",
+        )
+    )
+
+    assert isinstance(err, InvokeRateLimitError)
+    assert str(err) == "quota exceeded"
+
+
+def test_agent_backend_failure_to_exception_preserves_unknown_reason_context():
+    err = app_runner_module._agent_backend_failure_to_exception(
+        AgentBackendRunFailedInternalEvent(
+            run_id="run-1",
+            source_event_id="event-1",
+            error="Knowledge retrieval failed",
+            reason="knowledge_retrieve_failed",
+        )
+    )
+
+    assert isinstance(err, AgentBackendRunFailedError)
+    assert err.run_id == "run-1"
+    assert err.reason == "knowledge_retrieve_failed"
+    assert err.source_event_id == "event-1"
+    assert err.detail == {
+        "error": "Knowledge retrieval failed",
+        "reason": "knowledge_retrieve_failed",
+        "source_event_id": "event-1",
+    }
+    assert str(err) == "Knowledge retrieval failed (agent_run_id=run-1)"
+
+
 def test_stopped_task_cancels_agent_backend_run_and_skips_session_save():
    client = _RecordingFakeAgentBackendRunClient()
    store = _FakeSessionStore()
@@ -116,10 +116,12 @@ class TestAgentChatAppGeneratorGenerate:
        )

        thread_obj = mocker.MagicMock()
-        mocker.patch(
+        thread_constructor = mocker.patch(
            "core.app.apps.agent_chat.app_generator.threading.Thread",
            return_value=thread_obj,
        )
+        session = mocker.MagicMock()
+        mocker.patch("core.app.apps.agent_chat.app_generator.db.session", return_value=session)

        mocker.patch(
            "core.app.apps.agent_chat.app_generator.AgentChatAppGenerateResponseConverter.convert",
@@ -144,6 +146,7 @@ class TestAgentChatAppGeneratorGenerate:

        assert result == {"result": "ok"}
        assert generate_entity.call_args.kwargs["extras"]["trace_session_id"] == "session-1"
+        assert thread_constructor.call_args.kwargs["kwargs"]["session"] is session
        thread_obj.start.assert_called_once()

    def test_generate_without_file_config(self, generator, mocker: MockerFixture):
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
L1nSn0w	abb2ea82d7	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	2026-07-13 11:57:51 +08:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	408cd12835	chore(deps): bump the storage group in /api with 5 updates (#38825 ) Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-13 11:32:27 +08:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	ca46d07ea8	build(deps): bump soupsieve from 2.8 to 2.8.4 in /api (#38631 ) Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-13 11:32:06 +08:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	e721801b37	chore(deps): bump the google group in /api with 2 updates (#38811 ) Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-13 11:31:53 +08:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	f03648435a	chore(deps): bump flask-cors from 6.0.2 to 6.0.5 in /api in the flask group (#38810 ) Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-13 11:31:37 +08:00
JoelandGitHub	41683013e3	fix: marketplace page header jitter when it in sticky status (#38832 )	2026-07-13 03:19:10 +00:00
Asuka MinatoandGitHub	0e84444809	test: use sqlite3 session in test_data_migration_commands (#38672 )	2026-07-13 03:04:37 +00:00
Asuka MinatoandGitHub	b893b53f0a	test: use sqlite3 session in test_plugin_tool (#38781 )	2026-07-13 03:03:12 +00:00
Asuka MinatoandGitHub	4c1c92eb2f	test: use sqlite3 session in test_remove_app_and_related_data_task (#38778 )	2026-07-13 03:02:30 +00:00
非法操作 GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	7311f1ba6d	chore: sandbox plan should not execution special workflow (#38662 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-13 02:55:39 +00:00
Wu TianweiandGitHub	f4dfac7214	fix: update role selector to use max height for scroll area (#38831 )	2026-07-13 02:38:15 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	6d8e390c95	chore(deps): bump the github-actions-dependencies group with 4 updates (#38807 ) Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-13 01:39:28 +00:00
Stephen Zhou GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	a84c2d36a3	style: format with vp fmt (#38803 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-12 15:57:46 +00:00
Yufeng He GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	fde08d24fe	fix: eagerly validate conversation on service-api and explore chat endpoints (#38801 ) Signed-off-by: Yufeng He <[email protected]> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-12 15:19:07 +00:00
Stephen Zhou GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	b54107e9b9	chore: migrate formatting to vp fmt (#38705 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-12 14:59:00 +00:00
yyh GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	594c21d98c	fix(web): split model selection and settings actions (#38797 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-12 14:24:27 +00:00
yyhandGitHub	6e59c723c9	fix(agent-v2): restore keyboard access to row actions (#38792 )	2026-07-12 14:23:49 +00:00
yyhandGitHub	4efab8d7a3	chore(dify-ui): enforce strict ESLint rules (#38793 )	2026-07-12 14:23:39 +00:00
yyhandGitHub	2a2193dabb	fix(web): correct knowledge dialog draft state (#38794 )	2026-07-12 14:23:28 +00:00
yyhandGitHub	1246aa7f3a	refactor(web): normalize z-index layering (#38796 )	2026-07-12 14:23:14 +00:00
yyhandGitHub	3defd50709	fix(web): improve marketplace install dialog accessibility (#38798 )	2026-07-12 14:18:39 +00:00
yyhandGitHub	6e5ba18d65	refactor(dify-ui): standardize primitive boundary names (#38669 )	2026-07-10 14:17:10 +00:00
Stephen ZhouandGitHub	e3a08b40f2	test(web): centralize i18n mocks (#38670 )	2026-07-10 14:08:33 +00:00
Stephen ZhouandGitHub	6f47950fe2	perf(web): lazy-load Vite plugins (#38663 )	2026-07-10 11:13:09 +00:00
林玮 (Jade Lin)andGitHub	4043adacd7	fix(trigger): return 429 instead of 500 when API quota is exceeded (#38664 )	2026-07-10 10:57:50 +00:00
Stephen Zhou GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	b9fde89781	refactor(web): migrate i18n to selector optimize mode (#38657 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-10 09:35:31 +00:00
github-actions[bot]GitHubclaude[bot] <41898282+claude[bot]@users.noreply.github.com>yyh	d612307658	chore(i18n): sync translations with en-US (#38658 ) Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com> Co-authored-by: yyh <[email protected]>	2026-07-10 09:34:41 +00:00
wangxiaolei GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	7e05f28a46	feat: create app init rbac scope to all and add workspace members. (#38636 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-10 09:31:41 +00:00
林玮 (Jade Lin)andGitHub	44fb074359	fix(trigger): surface webhook trigger quota exceeded as 429 response (#38656 )	2026-07-10 08:50:25 +00:00
无能的迪拜巧克力andGitHub	775ed70bac	feat(web): refine web app customization options (#38650 )	2026-07-10 08:35:26 +00:00
KVOJJJinandGitHub	3c8e0e2113	fix(web): support punctuation in markdown form field names (#38651 )	2026-07-10 07:08:07 +00:00
yyhandGitHub	489e77658e	fix(web): simplify infotip icon variants (#38644 )	2026-07-10 06:37:36 +00:00
JoelandGitHub	4a8c6773a3	fix: final answer in new agent would clear after sse if agent thought has answer (#38649 )	2026-07-10 06:33:26 +00:00
Asuka MinatoandGitHub	d0f0134314	test: use sqlite3 session (#38641 )	2026-07-10 06:11:24 +00:00
zyssyz123andGitHub	9b3650e2ca	fix(agent): surface knowledge retrieval failures (#38640 )	2026-07-10 04:40:46 +00:00
KVOJJJinandGitHub	0038bcc4bb	fix(web): agent tool credentials mismatch (#38637 )	2026-07-10 04:35:52 +00:00
yyhandGitHub	b325597d29	test(e2e): consume agent service API SSE incrementally (#38634 )	2026-07-10 04:26:04 +00:00
Stephen ZhouandGitHub	b7255e6f27	chore: migrate js-yaml to v5 (#38639 )	2026-07-10 04:03:30 +00:00
yyhandGitHub	c3fb5344cb	fix: simplify help menu upgrade item (#38635 )	2026-07-10 03:58:27 +00:00
Yunlu Wen GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Copilot Autofix powered by AI	888bea9911	feat: ee agent sandbox support (#38528 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Copilot Autofix powered by AI <[email protected]>	2026-07-10 03:56:55 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2da37a9416	test: Mock to sqlite3 (#38499 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-10 03:49:06 +00:00
chariri GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	f902551d0c	chore(api): Prohibit new direct SQLAlchemy in controller (#38624 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-10 03:10:27 +00:00
DresdenandGitHub	faba726a45	chore: Replace @patch logger.warning with caplog fixture in credit_pool_service test (#38626 )	2026-07-10 03:08:17 +00:00
wangxiaoleiandGitHub	7da3d7074a	chore: remove unused param (#38628 )	2026-07-10 02:20:16 +00:00
Stephen Zhou GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	f9dd37420b	chore: update deps (#38627 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-10 02:07:57 +00:00
yyhandGitHub	eb74946b25	fix(web): save agent configure draft on route leave (#38620 )	2026-07-09 13:55:34 +00:00
QuantumGhostandGitHub	b5e35fc2fc	fix: remove hardcoded sandbox path in configuration file (#38618 )	2026-07-09 13:30:56 +00:00
wangxiaoleiandGitHub	7bfbb2bbe8	fix: fix auth prefix duplicate (#38616 )	2026-07-09 12:55:58 +00:00
QuantumGhostandGitHub	d177998255	chore: Bump version to 1.16.0-rc1 (#38600 )	2026-07-09 11:42:09 +00:00
wangxiaolei GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	74ee665af6	fix: fix miss session param (#38612 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-09 11:26:16 +00:00
Stephen ZhouandGitHub	d655153a3d	fix(web): update docs links (#38591 )	2026-07-09 11:25:31 +00:00
zyssyz123 GitHub 盐粒 Yanli yyh Joelautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>QuantumGhost yyh 盐粒 Yanli	925f97be20	feat: daily sync (#38593 ) Co-authored-by: 盐粒 Yanli <[email protected]> Co-authored-by: yyh <[email protected]> Co-authored-by: Joel <[email protected]> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: QuantumGhost <[email protected]> Co-authored-by: yyh <[email protected]> Co-authored-by: 盐粒 Yanli <[email protected]>	2026-07-09 10:35:15 +00:00
Leilei GitHub Claude Fable 5	9252d81826	docs: remove Dify Premium on AWS Marketplace section from all READMEs (#38607 ) Co-authored-by: Claude Fable 5 <[email protected]>	2026-07-09 10:09:59 +00:00
Harsh Kashyap GitHub Harsh Kashyapautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Harsh Kashyap	9d5819a9c1	fix(api): ignore invalid utf8 cache payloads (#37835 ) Co-authored-by: Harsh Kashyap <[email protected]> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Harsh Kashyap <[email protected]>	2026-07-09 09:52:49 +00:00
Ingram Z GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	99e3b1a401	fix: harden workflow archive DB retries (#38170 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-09 09:46:28 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	45957225cd	chore: batch example #38419 (#38474 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-09 09:14:52 +00:00
Coding On Star GitHub CodingOnStar	2f99652203	fix: chunk workflow failure tracking data (#38598 ) Co-authored-by: CodingOnStar <[email protected]>	2026-07-09 08:33:47 +00:00
melooooooo GitHub Cursor	dc1131b6df	refactor(tests): replace logger mocks with caplog in trace provider tests (#38569 ) Co-authored-by: Cursor <[email protected]>	2026-07-09 08:33:28 +00:00
Stephen ZhouandGitHub	a5a7c762a3	refactor(web): split app context state atoms (#38588 )	2026-07-09 07:42:02 +00:00
Coding On Star GitHub CodingOnStarautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	063e390c5d	fix(web): preserve attribution from auth redirect (#38583 ) Co-authored-by: CodingOnStar <[email protected]> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-09 07:40:55 +00:00
JoelandGitHub	3b3c25273a	fix: guard chat tree against out-of-order parents (#38590 )	2026-07-09 07:05:05 +00:00