Revert "chore: improve lambda plugin runtime error display"

This reverts commit 6bc1508a6b.

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

@@ -12,7 +12,6 @@ Use this as the component decision guide for Dify web. Existing code is referenc
 | Question | Default | Promote or extract only when |
 | --- | --- | --- |
 | Where should code live? | Keep it local to the feature workflow, route, or owner. | Multiple verticals need the same stable primitive. |
-| How should route/tab folders be named? | Match the current route segment, tab name, or user-visible surface. | Keep a historical or broader parent only when it still owns multiple surfaces. |
 | Who owns state, data, and handlers? | The lowest component that uses them. | A parent coordinates shared loading, errors, empty UI, selection, submission, navigation, or one consistent snapshot. |
 | Should this become Jotai state? | Keep synchronous UI/form state in component or DOM state. | Siblings need one source of truth, the value drives atoms, or scoped workflow state must survive hidden/unmounted steps. |
 | Should URL state enter Jotai? | Let Next.js route params and `nuqs` own URL state and updates. | Query atoms or shared derived atoms need a read-only bridge hydrated at the route/surface boundary. |
@@ -24,9 +23,7 @@ Use this as the component decision guide for Dify web. Existing code is referenc
 
 - Search before adding UI, hooks, helpers, query utilities, or styling patterns. Reuse existing base components, feature components, hooks, utilities, and design styles when they fit.
 - Follow Dify's CSS-first Tailwind v4 contract from `packages/dify-ui/README.md` and `packages/dify-ui/AGENTS.md`. Prefer design-system tokens, utilities, and radius mappings over generic Tailwind choices.
-- Group feature code by workflow, route, or ownership area with route-aligned names: components, hooks, local types, query helpers, atoms, constants, tests, and small utilities should live near the code that changes with them.
-- For each feature module, keep a module-local `README.md` as a boundary note. Start with the module name, a brief one-sentence description, then split dependencies into `Internal Modules` and `External Modules` sections; keep both sections and write `None.` when one category is empty. `Internal Modules` lists modules inside the same overall feature using paths from that feature root, such as `shared/domain/runtime-status`; `External Modules` lists project modules outside the feature using paths from the web root without a `web/` prefix, such as `app/components/base/skeleton`. Omit npm packages, workspace package dependencies, and whitelisted plumbing modules. Do not copy caller-relative import paths into the README.
-- Module README whitelist: `@/service/client`, `@/next/*`.
+- Group feature code by workflow, route, or ownership area: components, hooks, local types, query helpers, atoms, constants, and small utilities should live near the code that changes with them.
 - Keep source/default selection, validation, dirty checks, and payload shaping close to the workflow that owns submit behavior. Do not hide flow-specific priority order, fallback behavior, or submit semantics in generic utilities.
 - Prefer direct conditionals for small branch-specific decisions, especially form source selection and request payload assembly.
 - Loading states for page sections, cards, lists, tables, forms, and drawers should be skeletons scoped to the content being loaded. Use spinners only for small inline busy indicators.
@@ -35,8 +32,6 @@ Use this as the component decision guide for Dify web. Existing code is referenc
 
 - State-heavy wizards, drawers, modals, and secondary workflows can be a small feature surface: an entry file, one feature-local state file when Jotai is actually needed, and shallow `ui/` owners that match real visual regions.
 - The entry file handles route integration, provider wiring, close behavior, and surface mounting. The composition owner handles high-level workflow branching. The closest visual owner handles section branching.
-- When a page or tab maps to a route segment, name its feature folder after that route/tab surface instead of a stale parent grouping. Remove misleading intermediate folders when only one surface remains.
-- When a tab folder grows into several independent sections or action areas, split the first level by product/visual owners. Keep the root for the entry component and cross-owner state, colocate tests with the owner folder, and put truly shared local UI under a specifically named `components/` file.
 - Repeated TanStack query calls in sibling components are acceptable when each component independently consumes the data; TanStack Query deduplicates and shares cache.
 - Pass stable domain identity across boundaries. Do not forward derived presentation state when the receiver can derive it from its own data source.
 - A component that owns a visual surface should also own data access, loading, empty, and error states for content rendered inside it unless a parent truly coordinates that state.
@@ -51,7 +46,6 @@ Use this as the component decision guide for Dify web. Existing code is referenc
 - Use uncontrolled `@langgenius/dify-ui/form` and `@langgenius/dify-ui/field` controls for edit/create forms whose fields are read only at submit time. Initialize query-backed defaults with `defaultValue` and keyed remounts.
 - Promote form state to atoms only when another component must react to in-progress values, a draft must survive unmount/remount in the scoped workflow, or multiple steps share the same editable draft before submit.
 - Treat `useParams`, route args, and `nuqs` query state as framework-owned state. When atom logic needs those values, hydrate primitive atoms at the route or surface boundary, such as with `useHydrateAtoms(..., { dangerouslyForceHydrate: true })`; keep URL updates in the route/query-state APIs instead of write atoms.
-- Within a route-owned feature, choose one source for route identity. If route params are bridged into feature atoms, use that bridge consistently for route-derived queries and actions instead of also threading the same route id through page, tab, and section props.
 - For async work tied to atom state, use `atomWithQuery` or `atomWithMutation`; write atoms should update only the inputs that drive those atoms. This applies to pure frontend async work as well as network requests, so do not hand-roll loading/error/in-flight state with `useState` or `useRef` for atom-orchestrated async behavior. For component-owned remote work, use `useQuery` or `useMutation` directly.
 - Row-local async state belongs to the row owner unless it participates in a shared Jotai workflow or needs atom-scoped reset semantics.
 - Leave query and mutation atoms unscoped so they keep shared QueryClient cache and invalidation behavior. Scope resettable primitives and explicit hydration tuples; scope a derived atom only when every dependency should be private to that surface.
@@ -66,10 +60,8 @@ Use this as the component decision guide for Dify web. Existing code is referenc
 - Type component signatures directly; do not use `FC` or `React.FC`.
 - Prefer `function` for top-level components and module helpers. Use arrow functions for local callbacks, handlers, and lambda-style APIs.
 - Prefer named exports. Use default exports only where the framework requires them, such as Next.js route files.
-- Avoid barrel files that only re-export secondary owners. `index.tsx` is acceptable for a route/tab entry component; import header controls, switches, sections, and row owners from their concrete owner files.
 - Type simple one-off props inline. Use a named `Props` type only when reused, exported, complex, or clearer.
 - Use API-generated or API-returned types at component boundaries. Keep small UI conversion helpers and one-off UI extensions beside the component that needs them.
-- Avoid `common.tsx` buckets for shared UI. Use a feature-local `components/` folder with concrete filenames that describe the shared role.
 - Do not create type aliases that only rename another type. Use aliases only for real UI concepts, refinements, or reusable local contracts.
 - Name values by their domain role and backend API contract, especially persistent IDs and route params. Normalize framework or route params at the boundary.
 - Put fallback and invariant checks in the lowest component that already handles that state. Do not extract helpers whose only behavior is hiding missing display data.
@@ -91,13 +83,11 @@ Use this as the component decision guide for Dify web. Existing code is referenc
 
 - Keep `web/contract/*` as the API shape source of truth and follow the `{ params, query?, body? }` input shape.
 - Consume generated queries with `useQuery(consoleQuery.xxx.queryOptions(...))` or `useQuery(marketplaceQuery.xxx.queryOptions(...))`.
-- If a generated query input comes from an atom, including a route-identity bridge atom, keep the query in `atomWithQuery`; do not unwrap the atom in a component just to call `useQuery`.
 - Consume owner-local mutations with `useMutation(consoleQuery.xxx.mutationOptions(...))` or `useMutation(marketplaceQuery.xxx.mutationOptions(...))` when pending/error state is not consumed by feature atoms.
 - In `atomWithQuery`, `atomWithInfiniteQuery`, and `atomWithMutation`, return generated `queryOptions()`, `infiniteOptions()`, or `mutationOptions()` directly. Pass `enabled`, `retry`, `placeholderData`, `select`, and pagination options into the generated call instead of spreading options into a hand-built object.
 - For generated oRPC options with missing required input, branch the whole input with `input: condition ? validInput : skipToken` and `enabled: Boolean(condition)`. Never place `skipToken` inside a nested placeholder payload or coerce required IDs to `''`.
 - When prefetch and render use the same request, extract local query options or a query-options atom so `prefetchQuery` and `useQuery`/`atomWithQuery` share the exact options.
 - For custom query or mutation functions, wrap options with TanStack `queryOptions(...)` or `mutationOptions(...)`.
-- Do not extract generated `queryOptions(...)` into a helper solely to share input construction; extract only when prefetch/render must share exact options or the helper owns real domain behavior.
 - Avoid pass-through hooks and thin `web/service/use-*` wrappers that only rename generated options. Keep feature hooks for real orchestration, workflow state, or shared domain behavior.
 - Put shared cache behavior in `createTanstackQueryUtils(...experimental_defaults...)`. Component or atom callbacks may handle local toasts, closing dialogs, and navigation, but should not replace shared invalidation or patch shared server state locally.
 - For overlays that may open heavier secondary content, prefetch from the trigger/menu open event with `queryClient.prefetchQuery(queryOptions)` when `onOpenChange` is available. Do not mount hidden subscribers just to warm cache.
@@ -106,7 +96,7 @@ Use this as the component decision guide for Dify web. Existing code is referenc
 
 ## Boundaries And Overlays
 
-- Use the first level below a page or tab to organize independent page sections when it adds structure or the root folder becomes noisy. This layer is layout/semantic first, not automatically the data owner.
+- Use the first level below a page or tab to organize independent page sections when it adds structure. This layer is layout/semantic first, not automatically the data owner.
 - Treat component names, semantic roles, and user- or design-marked visual regions as boundary constraints. Keep adjacent UI as a sibling owner or introduce a correctly named broader owner.
 - Keep cohesive forms, menu bodies, and one-off helpers local unless they need their own state, reuse, or semantic boundary.
 - Separate hidden secondary surfaces from the trigger's main flow. For dialogs, dropdowns, popovers, and similar branches, extract a small local component when hidden content would obscure the parent.

.github/workflows/api-tests.yml

@@ -16,7 +16,7 @@ concurrency:
 jobs:
   api-unit:
     name: API Unit Tests
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     env:
       COVERAGE_FILE: coverage-unit
     defaults:
@@ -75,7 +75,7 @@ jobs:
 
   api-integration:
     name: API Integration Tests
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     env:
       COVERAGE_FILE: coverage-integration
       STORAGE_TYPE: opendal
@@ -129,7 +129,7 @@ jobs:
 
   api-coverage:
     name: API Coverage
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     needs:
       - api-unit
       - api-integration

.github/workflows/build-push.yml

@@ -171,7 +171,7 @@ jobs:
 
   create-manifest:
     needs: build
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     if: github.repository == 'langgenius/dify'
     strategy:
       matrix:

.github/workflows/cli-release.yml

@@ -23,7 +23,7 @@ concurrency:
 jobs:
   validate:
     name: validate manifest + resolve target Dify release
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     if: github.repository == 'langgenius/dify'
     permissions:
       contents: read
@@ -87,7 +87,7 @@ jobs:
   release:
     name: build + attach standalone binaries (all targets)
     needs: validate
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     permissions:
       contents: write
     defaults:

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

@@ -9,7 +9,7 @@ concurrency:
 
 jobs:
   db-migration-test-postgres:
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
 
     steps:
       - name: Checkout code
@@ -59,7 +59,7 @@ jobs:
         run: uv run --directory api flask upgrade-db
 
   db-migration-test-mysql:
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
 
     steps:
       - name: Checkout code

.github/workflows/deploy-agent.yml

@@ -13,7 +13,7 @@ on:
 
 jobs:
   deploy:
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     if: |
       github.event.workflow_run.conclusion == 'success' &&
       github.event.workflow_run.head_branch == 'deploy/agent'

.github/workflows/deploy-dev.yml

@@ -10,7 +10,7 @@ on:
 
 jobs:
   deploy:
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     if: |
       github.event.workflow_run.conclusion == 'success' &&
       github.event.workflow_run.head_branch == 'deploy/dev'

.github/workflows/deploy-enterprise.yml

@@ -13,7 +13,7 @@ on:
 
 jobs:
   deploy:
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     if: |
       github.event.workflow_run.conclusion == 'success' &&
       github.event.workflow_run.head_branch == 'deploy/enterprise'

.github/workflows/deploy-saas.yml

@@ -13,7 +13,7 @@ on:
 
 jobs:
   deploy:
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     if: |
       github.event.workflow_run.conclusion == 'success' &&
       github.event.workflow_run.head_branch == 'deploy/saas'

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

@@ -22,7 +22,7 @@ concurrency:
 jobs:
   check-cherry-pick-provenance:
     name: Require cherry-pick provenance
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
         with:

.github/workflows/labeler.yml

@@ -7,7 +7,7 @@ jobs:
     permissions:
       contents: read
       pull-requests: write
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - uses: actions/labeler@f27b608878404679385c85cfa523b85ccb86e213 # v6.1.0
         with:

.github/workflows/main-ci.yml

@@ -23,7 +23,7 @@ concurrency:
 jobs:
   pre_job:
     name: Skip Duplicate Checks
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     outputs:
       should_skip: ${{ steps.skip_check.outputs.should_skip || 'false' }}
     steps:
@@ -39,7 +39,7 @@ jobs:
     name: Check Changed Files
     needs: pre_job
     if: needs.pre_job.outputs.should_skip != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     outputs:
       api-changed: ${{ steps.changes.outputs.api }}
       cli-changed: ${{ steps.changes.outputs.cli }}
@@ -152,7 +152,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.api-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Report skipped API tests
         run: echo "No API-related changes detected; skipping API tests."
@@ -165,7 +165,7 @@ jobs:
       - check-changes
       - api-tests-run
       - api-tests-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Finalize API Tests status
         env:
@@ -212,7 +212,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.cli-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Report skipped CLI tests
         run: echo "No CLI-related changes detected; skipping CLI tests."
@@ -225,7 +225,7 @@ jobs:
       - check-changes
       - cli-tests-run
       - cli-tests-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Finalize CLI Tests status
         env:
@@ -272,7 +272,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.web-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Report skipped web tests
         run: echo "No web-related changes detected; skipping web tests."
@@ -285,7 +285,7 @@ jobs:
       - check-changes
       - web-tests-run
       - web-tests-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Finalize Web Tests status
         env:
@@ -331,7 +331,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Report skipped web full-stack e2e
         run: echo "No E2E-related changes detected; skipping web full-stack E2E."
@@ -344,7 +344,7 @@ jobs:
       - check-changes
       - web-e2e-run
       - web-e2e-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Finalize Web Full-Stack E2E status
         env:
@@ -396,7 +396,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.vdb-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Report skipped VDB tests
         run: echo "No VDB-related changes detected; skipping VDB tests."
@@ -409,7 +409,7 @@ jobs:
       - check-changes
       - vdb-tests-run
       - vdb-tests-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Finalize VDB Tests status
         env:
@@ -455,7 +455,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.migration-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Report skipped DB migration tests
         run: echo "No migration-related changes detected; skipping DB migration tests."
@@ -468,7 +468,7 @@ jobs:
       - check-changes
       - db-migration-test-run
       - db-migration-test-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Finalize DB Migration Test status
         env:

.github/workflows/pyrefly-diff-comment.yml

@@ -12,7 +12,7 @@ permissions: {}
 jobs:
   comment:
     name: Comment PR with pyrefly diff
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     permissions:
       actions: read
       contents: read

.github/workflows/pyrefly-diff.yml

@@ -10,7 +10,7 @@ permissions:
 
 jobs:
   pyrefly-diff:
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     permissions:
       contents: read
       issues: write

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

@@ -12,7 +12,7 @@ permissions: {}
 jobs:
   comment:
     name: Comment PR with type coverage
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     permissions:
       actions: read
       contents: read

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

@@ -10,7 +10,7 @@ permissions:
 
 jobs:
   pyrefly-type-coverage:
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     permissions:
       contents: read
       issues: write

.github/workflows/semantic-pull-request.yml

@@ -16,7 +16,7 @@ jobs:
     name: Validate PR title
     permissions:
       pull-requests: read
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     steps:
       - name: Complete merge group check
         if: github.event_name == 'merge_group'

.github/workflows/stale.yml

@@ -12,7 +12,7 @@ on:
 jobs:
   stale:
 
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     permissions:
       issues: write
       pull-requests: write

.github/workflows/style.yml

@@ -15,7 +15,7 @@ permissions:
 jobs:
   python-style:
     name: Python Style
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
 
     steps:
       - name: Checkout code
@@ -53,8 +53,6 @@ jobs:
 
       - name: Run Type Checks
         if: steps.changed-files.outputs.any_changed == 'true'
-        env:
-          PYREFLY_OUTPUT_FORMAT: github
         run: make type-check-core
 
       - name: Dotenv check
@@ -63,7 +61,7 @@ jobs:
 
   web-style:
     name: Web Style
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     defaults:
       run:
         working-directory: ./web
@@ -177,7 +175,7 @@ jobs:
 
   superlinter:
     name: SuperLinter
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
 
     steps:
       - name: Checkout code

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

@@ -17,7 +17,7 @@ concurrency:
 jobs:
   build:
     name: unit test for Node.js SDK
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
 
     defaults:
       run:

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

@@ -35,7 +35,7 @@ concurrency:
 jobs:
   translate:
     if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     timeout-minutes: 120
 
     steps:

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

@@ -16,7 +16,7 @@ concurrency:
 jobs:
   trigger:
     if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     timeout-minutes: 5
 
     steps:

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

@@ -16,7 +16,7 @@ jobs:
   test:
     name: Full VDB Tests
     if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     strategy:
       matrix:
         python-version:

.github/workflows/vdb-tests.yml

@@ -13,7 +13,7 @@ concurrency:
 jobs:
   test:
     name: VDB Smoke Tests
-    runs-on: depot-ubuntu-24.04
+    runs-on: depot-ubuntu-24.04-4
     strategy:
       matrix:
         python-version:

api/app_factory.py

@@ -2,14 +2,14 @@ import logging
 import time
 
 import socketio
-from flask import request
+from flask import g, request
 from opentelemetry.trace import get_current_span
 from opentelemetry.trace.span import INVALID_SPAN_ID, INVALID_TRACE_ID
 
 from configs import dify_config
 from contexts.wrapper import RecyclableContextVar
 from controllers.console.error import UnauthorizedAndForceLogout
-from core.logging.context import init_request_context
+from core.logging.context import clear_request_context, init_request_context
 from dify_app import DifyApp
 from extensions.ext_socketio import sio
 from services.enterprise.enterprise_service import EnterpriseService
@@ -60,6 +60,7 @@ def create_flask_app_with_configs() -> DifyApp:
     def before_request():
         # Initialize logging context for this request
         init_request_context()
+        g._logging_context_cleanup_scheduled = False
         RecyclableContextVar.increment_thread_recycles()
 
         # Enterprise license validation for API endpoints (both console and webapp)
@@ -117,9 +118,26 @@ def create_flask_app_with_configs() -> DifyApp:
             logger.warning("Failed to add trace headers to response", exc_info=True)
         return response
 
+    @dify_app.after_request
+    def schedule_logging_context_cleanup(response):
+        """Keep logging context through streaming, then clear it when WSGI closes the response."""
+        if response.direct_passthrough:
+            return response
+        g._logging_context_cleanup_scheduled = True
+        response.call_on_close(clear_request_context)
+        return response
+
+    @dify_app.teardown_request
+    def clear_unscheduled_logging_context(_error: BaseException | None) -> None:
+        """Clear when no response-close callback can own cleanup."""
+        if not g.get("_logging_context_cleanup_scheduled", False):
+            clear_request_context()
+
     # Capture the decorator return values so static checkers do not treat the hooks as unused.
     _ = before_request
     _ = add_trace_headers
+    _ = schedule_logging_context_cleanup
+    _ = clear_unscheduled_logging_context
 
     return dify_app
 

api/clients/agent_backend/request_builder.py

@@ -78,22 +78,11 @@ def _filter_snapshot_to_specs(
     return CompositorSessionSnapshot(schema_version=snapshot.schema_version, layers=filtered_layers)
 
 
-def _shell_layer_deps() -> dict[str, str]:
-    return {"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID}
-
-
-def _drive_layer_deps() -> dict[str, str]:
-    return {"shell": DIFY_SHELL_LAYER_ID}
-
-
-def _shell_config_with_drive_ref(
-    shell_config: DifyShellLayerConfig | None,
-    drive_config: DifyDriveLayerConfig | None,
-) -> DifyShellLayerConfig:
-    config = shell_config or DifyShellLayerConfig()
-    if drive_config is None:
-        return config
-    return config.model_copy(update={"agent_stub_drive_ref": drive_config.drive_ref})
+def _shell_layer_deps(*, include_drive: bool) -> dict[str, str]:
+    deps = {"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID}
+    if include_drive:
+        deps["drive"] = DIFY_DRIVE_LAYER_ID
+    return deps
 
 
 class AgentBackendModelConfig(BaseModel):
@@ -274,29 +263,14 @@ class AgentBackendRunRequestBuilder:
             ]
         )
 
-        include_shell = run_input.include_shell or run_input.drive_config is not None
-        if include_shell:
-            # Sandboxed bash workspace (dify.shell). It enters before drive so
-            # drive can materialize mentioned targets with `dify-agent drive pull`
-            # in the same shell-visible filesystem used by model commands.
-            layers.append(
-                RunLayerSpec(
-                    name=DIFY_SHELL_LAYER_ID,
-                    type=DIFY_SHELL_LAYER_TYPE_ID,
-                    deps=_shell_layer_deps(),
-                    metadata=run_input.metadata,
-                    config=_shell_config_with_drive_ref(run_input.shell_config, run_input.drive_config),
-                )
-            )
-
         if run_input.drive_config is not None:
-            # Drive Skills & Files declaration (dify.drive): the catalog plus
-            # prompt-mentioned entries eagerly pulled through the shell layer.
+            # Drive Skills & Files declaration (dify.drive): a config-only index;
+            # the agent pulls listed entries through the back proxy by drive_ref.
             layers.append(
                 RunLayerSpec(
                     name=DIFY_DRIVE_LAYER_ID,
                     type=DIFY_DRIVE_LAYER_TYPE_ID,
-                    deps=_drive_layer_deps(),
+                    deps={"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID},
                     metadata=run_input.metadata,
                     config=run_input.drive_config,
                 )
@@ -338,7 +312,7 @@ class AgentBackendRunRequestBuilder:
                 )
             )
 
-        if run_input.knowledge is not None and run_input.knowledge.sets:
+        if run_input.knowledge is not None and run_input.knowledge.dataset_ids:
             layers.append(
                 RunLayerSpec(
                     name=DIFY_KNOWLEDGE_BASE_LAYER_ID,
@@ -362,6 +336,21 @@ class AgentBackendRunRequestBuilder:
                 )
             )
 
+        if run_input.include_shell:
+            # Sandboxed bash workspace (dify.shell). Depends on execution_context
+            # so the agent server can mint per-command Agent Stub env, and on
+            # drive when present so that env points at /mnt/drive/<drive_ref>.
+            # shellctl connection itself is server-injected.
+            layers.append(
+                RunLayerSpec(
+                    name=DIFY_SHELL_LAYER_ID,
+                    type=DIFY_SHELL_LAYER_TYPE_ID,
+                    deps=_shell_layer_deps(include_drive=run_input.drive_config is not None),
+                    metadata=run_input.metadata,
+                    config=run_input.shell_config or DifyShellLayerConfig(),
+                )
+            )
+
         if run_input.output is not None:
             layers.append(
                 RunLayerSpec(
@@ -456,7 +445,7 @@ class AgentBackendRunRequestBuilder:
                     name=WORKFLOW_NODE_JOB_PROMPT_LAYER_ID,
                     type=PLAIN_PROMPT_LAYER_TYPE_ID,
                     metadata={**run_input.metadata, "origin": "workflow_node_job"},
-                    config=PromptLayerConfig(user=run_input.workflow_node_job_prompt),
+                    config=PromptLayerConfig(prefix=run_input.workflow_node_job_prompt),
                 ),
                 RunLayerSpec(
                     name=WORKFLOW_USER_PROMPT_LAYER_ID,
@@ -473,29 +462,14 @@ class AgentBackendRunRequestBuilder:
             ]
         )
 
-        include_shell = run_input.include_shell or run_input.drive_config is not None
-        if include_shell:
-            # Sandboxed bash workspace (dify.shell). It enters before drive so
-            # drive can materialize mentioned targets with `dify-agent drive pull`
-            # in the same shell-visible filesystem used by model commands.
-            layers.append(
-                RunLayerSpec(
-                    name=DIFY_SHELL_LAYER_ID,
-                    type=DIFY_SHELL_LAYER_TYPE_ID,
-                    deps=_shell_layer_deps(),
-                    metadata=run_input.metadata,
-                    config=_shell_config_with_drive_ref(run_input.shell_config, run_input.drive_config),
-                )
-            )
-
         if run_input.drive_config is not None:
-            # Drive Skills & Files declaration (dify.drive): the catalog plus
-            # prompt-mentioned entries eagerly pulled through the shell layer.
+            # Drive Skills & Files declaration (dify.drive): a config-only index;
+            # the agent pulls listed entries through the back proxy by drive_ref.
             layers.append(
                 RunLayerSpec(
                     name=DIFY_DRIVE_LAYER_ID,
                     type=DIFY_DRIVE_LAYER_TYPE_ID,
-                    deps=_drive_layer_deps(),
+                    deps={"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID},
                     metadata=run_input.metadata,
                     config=run_input.drive_config,
                 )
@@ -539,7 +513,7 @@ class AgentBackendRunRequestBuilder:
                 )
             )
 
-        if run_input.knowledge is not None and run_input.knowledge.sets:
+        if run_input.knowledge is not None and run_input.knowledge.dataset_ids:
             layers.append(
                 RunLayerSpec(
                     name=DIFY_KNOWLEDGE_BASE_LAYER_ID,
@@ -563,6 +537,21 @@ class AgentBackendRunRequestBuilder:
                 )
             )
 
+        if run_input.include_shell:
+            # Sandboxed bash workspace (dify.shell). Depends on execution_context
+            # so the agent server can mint per-command Agent Stub env, and on
+            # drive when present so that env points at /mnt/drive/<drive_ref>.
+            # shellctl connection itself is server-injected.
+            layers.append(
+                RunLayerSpec(
+                    name=DIFY_SHELL_LAYER_ID,
+                    type=DIFY_SHELL_LAYER_TYPE_ID,
+                    deps=_shell_layer_deps(include_drive=run_input.drive_config is not None),
+                    metadata=run_input.metadata,
+                    config=run_input.shell_config or DifyShellLayerConfig(),
+                )
+            )
+
         if run_input.output is not None:
             layers.append(
                 RunLayerSpec(

api/commands/__init__.py

@@ -22,7 +22,7 @@ from .plugin import (
     setup_system_trigger_oauth_client,
     transform_datasource_credentials,
 )
-from .rbac import migrate_member_roles_to_rbac
+from .rbac import migrate_dataset_permissions_to_rbac, migrate_member_roles_to_rbac
 from .retention import (
     archive_workflow_runs,
     archive_workflow_runs_plan,
@@ -76,6 +76,7 @@ __all__ = [
     "legacy_model_types",
     "migrate_annotation_vector_database",
     "migrate_data_for_plugin",
+    "migrate_dataset_permissions_to_rbac",
     "migrate_knowledge_vector_database",
     "migrate_member_roles_to_rbac",
     "migrate_oss",

api/commands/data_migrate.py

@@ -7,6 +7,7 @@ from typing import cast
 
 import click
 
+from commands.rbac import migrate_dataset_permissions_to_rbac
 from extensions.ext_database import db
 from graphon.model_runtime.entities.model_entities import ModelType
 from services.legacy_model_type_migration import (
@@ -177,3 +178,4 @@ def legacy_model_types(
 
 
 data_migrate.add_command(legacy_model_types)
+data_migrate.add_command(migrate_dataset_permissions_to_rbac)

api/commands/rbac.py

@@ -1,11 +1,55 @@
 from __future__ import annotations
 
+import json
+from collections.abc import Iterator
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
 import click
 from sqlalchemy import select
 
+from configs import dify_config
 from core.db.session_factory import session_factory
-from models import TenantAccountJoin, TenantAccountRole
-from services.enterprise.rbac_service import ListOption, RBACService
+from core.rbac import RBACResourceWhitelistScope
+from models import Dataset, DatasetPermission, DatasetPermissionEnum, TenantAccountJoin, TenantAccountRole
+from services.enterprise.rbac_service import ListOption, RBACService, ReplaceMemberBindings, ReplaceUserAccessPolicies
+
+_RBAC_DEFAULT_ACCESS_POLICY_ID = "default"
+
+_LEGACY_ROLE_TO_BUILTIN_TAG = {
+    TenantAccountRole.OWNER.value: "owner",
+    TenantAccountRole.ADMIN.value: "admin",
+    TenantAccountRole.EDITOR.value: "editor",
+    TenantAccountRole.NORMAL.value: "normal",
+    TenantAccountRole.DATASET_OPERATOR.value: "dataset_operator",
+}
+
+
+def _resolve_builtin_role_ids(tenant_id: str, operator_account_id: str) -> dict[str, str]:
+    """Resolve every legacy workspace role to the current tenant's builtin RBAC role id.
+
+    The migration replays the old `TenantAccountJoin.role` values onto the
+    RBAC member-role binding API. Builtin RBAC roles are tenant-scoped and
+    identified by runtime ids, so the command must look them up per tenant.
+    """
+    roles = RBACService.Roles.list(
+        tenant_id=tenant_id,
+        account_id=operator_account_id,
+        options=ListOption(page_number=1, results_per_page=100),
+    ).data
+    role_id_by_tag = {
+        role.role_tag: role.id
+        for role in roles
+        if role.is_builtin and role.category == "global_system_default" and role.role_tag
+    }
+    resolved: dict[str, str] = {}
+    for legacy_role, expected_builtin_tag in _LEGACY_ROLE_TO_BUILTIN_TAG.items():
+        role_id = role_id_by_tag.get(expected_builtin_tag)
+        if expected_builtin_tag == "dataset_operator" and not dify_config.DATASET_OPERATOR_ENABLED:
+            continue
+        if not role_id:
+            raise ValueError(f"Builtin RBAC role not found for tenant={tenant_id}, legacy_role={legacy_role}")
+        resolved[legacy_role] = role_id
+    return resolved
 
 
 def _resolve_builtin_role_id(tenant_id: str, operator_account_id: str, legacy_role: str) -> str:
@@ -15,26 +59,86 @@ def _resolve_builtin_role_id(tenant_id: str, operator_account_id: str, legacy_ro
     RBAC member-role binding API. Builtin RBAC roles are tenant-scoped and
     identified by runtime ids, so the command must look them up per tenant.
     """
-    expected_builtin_tag = {
-        TenantAccountRole.OWNER.value: "owner",
-        TenantAccountRole.ADMIN.value: "admin",
-        TenantAccountRole.EDITOR.value: "editor",
-        TenantAccountRole.NORMAL.value: "normal",
-        TenantAccountRole.DATASET_OPERATOR.value: "dataset_operator",
-    }.get(legacy_role)
-    if not expected_builtin_tag:
+    if legacy_role not in _LEGACY_ROLE_TO_BUILTIN_TAG:
         raise ValueError(f"Unsupported legacy workspace role: {legacy_role}")
 
-    roles = RBACService.Roles.list(
+    return _resolve_builtin_role_ids(tenant_id, operator_account_id)[legacy_role]
+
+
+def _iter_tenant_member_batches(
+    tenant_id: str | None,
+    *,
+    db_batch_size: int,
+    api_batch_size: int,
+) -> Iterator[tuple[str, str, list[tuple[str, str]]]]:
+    """Yield legacy member roles in tenant-scoped API-sized batches.
+
+    Rows are projected to primitive values and streamed from the database, so
+    the command never materializes every TenantAccountJoin ORM object. The
+    iterator only keeps one tenant's API-sized batches in memory while it
+    finds that tenant's owner account.
+    """
+    with session_factory.create_session() as session:
+        stmt = (
+            select(TenantAccountJoin.tenant_id, TenantAccountJoin.account_id, TenantAccountJoin.role)
+            .order_by(TenantAccountJoin.tenant_id.asc(), TenantAccountJoin.id.asc())
+            .execution_options(yield_per=db_batch_size)
+        )
+        if tenant_id:
+            stmt = stmt.where(TenantAccountJoin.tenant_id == tenant_id)
+
+        current_tenant_id: str | None = None
+        owner_account_id: str | None = None
+        batches: list[list[tuple[str, str]]] = []
+        batch: list[tuple[str, str]] = []
+
+        def flush_current_tenant() -> Iterator[tuple[str, str, list[tuple[str, str]]]]:
+            if current_tenant_id is None:
+                return
+            if batch:
+                batches.append(batch.copy())
+            if not owner_account_id:
+                raise ValueError(f"Workspace owner not found for tenant={current_tenant_id}")
+            for item in batches:
+                yield current_tenant_id, owner_account_id, item
+
+        for row in session.execute(stmt):
+            workspace_id = str(row.tenant_id)
+            if current_tenant_id is not None and workspace_id != current_tenant_id:
+                yield from flush_current_tenant()
+                owner_account_id = None
+                batches = []
+                batch = []
+            current_tenant_id = workspace_id
+            account_id = str(row.account_id)
+            role = str(row.role)
+            if role == TenantAccountRole.OWNER.value:
+                owner_account_id = account_id
+            batch.append((account_id, role))
+            if len(batch) >= api_batch_size:
+                batches.append(batch)
+                batch = []
+
+        yield from flush_current_tenant()
+
+
+def _member_already_has_role(current_roles_by_account_id: dict[str, set[str]], account_id: str, role_id: str) -> bool:
+    return current_roles_by_account_id.get(account_id) == {role_id}
+
+
+def _replace_member_role(
+    tenant_id: str,
+    operator_account_id: str,
+    member_account_id: str,
+    role_id: str,
+) -> str:
+    RBACService.MemberRoles.replace(
         tenant_id=tenant_id,
         account_id=operator_account_id,
-        options=ListOption(page_number=1, results_per_page=100),
-    ).data
-    for role in roles:
-        if role.is_builtin and role.category == "global_system_default" and role.role_tag == expected_builtin_tag:
-            return str(role.id)
-
-    raise ValueError(f"Builtin RBAC role not found for tenant={tenant_id}, legacy_role={legacy_role}")
+        member_account_id=member_account_id,
+        role_ids=[role_id],
+    )
+    return member_account_id
 
 
 @click.command(
@@ -42,7 +146,16 @@ def _resolve_builtin_role_id(tenant_id: str, operator_account_id: str, legacy_ro
 )
 @click.option("--tenant-id", help="Only migrate a single workspace.")
 @click.option("--dry-run", is_flag=True, default=False, help="Preview the migration without writing RBAC bindings.")
-def migrate_member_roles_to_rbac(tenant_id: str | None, dry_run: bool) -> None:
+@click.option("--db-batch-size", default=5000, show_default=True, help="Rows fetched per database batch.")
+@click.option("--api-batch-size", default=200, show_default=True, help="Members checked per RBAC batch_get call.")
+@click.option("--workers", default=1, show_default=True, help="Concurrent member role replace calls per tenant batch.")
+def migrate_member_roles_to_rbac(
+    tenant_id: str | None,
+    dry_run: bool,
+    db_batch_size: int,
+    api_batch_size: int,
+    workers: int,
+) -> None:
     """Backfill RBAC member-role bindings from legacy `TenantAccountJoin.role` data.
 
     This is an offline migration command for workspaces that already have
@@ -50,63 +163,322 @@ def migrate_member_roles_to_rbac(tenant_id: str | None, dry_run: bool) -> None:
     member-role binding store.
     """
     click.echo(click.style("Starting RBAC member-role migration.", fg="green"))
+    if workers < 1:
+        raise click.BadParameter("workers must be >= 1", param_hint="--workers")
 
-    with session_factory.create_session() as session:
-        stmt = select(TenantAccountJoin).order_by(TenantAccountJoin.tenant_id.asc(), TenantAccountJoin.id.asc())
-        if tenant_id:
-            stmt = stmt.where(TenantAccountJoin.tenant_id == tenant_id)
+    tenant_count = 0
+    scanned_count = 0
+    skipped_count = 0
+    migrated_count = 0
+    current_tenant_id: str | None = None
+    role_ids_by_legacy_role: dict[str, str] = {}
 
-        joins = list(session.scalars(stmt).all())
+    for workspace_id, owner_account_id, batch in _iter_tenant_member_batches(
+        tenant_id,
+        db_batch_size=db_batch_size,
+        api_batch_size=api_batch_size,
+    ):
+        scanned_count += len(batch)
+        if workspace_id != current_tenant_id:
+            tenant_count += 1
+            current_tenant_id = workspace_id
+            role_ids_by_legacy_role = _resolve_builtin_role_ids(workspace_id, owner_account_id)
+            click.echo(f"tenant={workspace_id}")
 
-    if not joins:
+        current_roles_by_account_id: dict[str, set[str]] = {}
+        if not dry_run:
+            current_roles = RBACService.MemberRoles.batch_get(
+                tenant_id=workspace_id,
+                account_id=owner_account_id,
+                member_account_ids=[account_id for account_id, _ in batch],
+            )
+            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:
+            resolved_role_id = role_ids_by_legacy_role.get(legacy_role)
+            if not resolved_role_id:
+                raise ValueError(f"Unsupported legacy workspace role: {legacy_role}")
+
+            if dry_run:
+                click.echo(
+                    f"tenant={workspace_id} member={member_account_id} "
+                    f"legacy_role={legacy_role} -> rbac_role_id={resolved_role_id}"
+                )
+                continue
+
+            if _member_already_has_role(current_roles_by_account_id, member_account_id, resolved_role_id):
+                skipped_count += 1
+                continue
+
+            replace_jobs.append((member_account_id, resolved_role_id))
+
+        if replace_jobs:
+            if workers == 1:
+                for member_account_id, resolved_role_id in replace_jobs:
+                    _replace_member_role(workspace_id, owner_account_id, member_account_id, resolved_role_id)
+                    migrated_count += 1
+            else:
+                with ThreadPoolExecutor(max_workers=workers) as executor:
+                    futures = [
+                        executor.submit(
+                            _replace_member_role,
+                            workspace_id,
+                            owner_account_id,
+                            member_account_id,
+                            resolved_role_id,
+                        )
+                        for member_account_id, resolved_role_id in replace_jobs
+                    ]
+                    for future in as_completed(futures):
+                        future.result()
+                        migrated_count += 1
+
+        if scanned_count % 10000 == 0:
+            click.echo(
+                f"progress scanned={scanned_count} migrated={migrated_count} skipped={skipped_count}",
+                err=True,
+            )
+
+    if scanned_count == 0:
         click.echo(click.style("No workspace members found for migration.", fg="yellow"))
         return
 
-    owner_account_by_tenant: dict[str, str] = {}
-    resolved_role_ids: dict[tuple[str, str], str] = {}
-    migrated_count = 0
-
-    for join in joins:
-        workspace_id = str(join.tenant_id)
-        member_account_id = str(join.account_id)
-        legacy_role = str(join.role)
-
-        if workspace_id not in owner_account_by_tenant:
-            owner_join = next(
-                (
-                    item
-                    for item in joins
-                    if str(item.tenant_id) == workspace_id and str(item.role) == TenantAccountRole.OWNER.value
-                ),
-                None,
-            )
-            if not owner_join:
-                raise ValueError(f"Workspace owner not found for tenant={workspace_id}")
-            owner_account_by_tenant[workspace_id] = str(owner_join.account_id)
-
-        operator_account_id = owner_account_by_tenant[workspace_id]
-        cache_key = (workspace_id, legacy_role)
-        if cache_key not in resolved_role_ids:
-            resolved_role_ids[cache_key] = _resolve_builtin_role_id(workspace_id, operator_account_id, legacy_role)
-
-        resolved_role_id = resolved_role_ids[cache_key]
+    if dry_run:
         click.echo(
-            f"tenant={workspace_id} member={member_account_id} "
-            f"legacy_role={legacy_role} -> rbac_role_id={resolved_role_id}"
+            click.style(
+                f"Dry run completed. Scanned {scanned_count} members across {tenant_count} tenants. "
+                "No RBAC bindings were written.",
+                fg="yellow",
+            )
+        )
+    else:
+        click.echo(
+            click.style(
+                f"RBAC member-role migration completed. Scanned {scanned_count} members across {tenant_count} tenants, "
+                f"migrated {migrated_count}, skipped {skipped_count} already up-to-date.",
+                fg="green",
+            )
         )
 
-        if dry_run:
-            continue
 
-        RBACService.MemberRoles.replace(
-            tenant_id=workspace_id,
-            account_id=operator_account_id,
-            member_account_id=member_account_id,
-            role_ids=[resolved_role_id],
-        )
-        migrated_count += 1
+def _dataset_permission_enum(permission: DatasetPermissionEnum | str | None) -> DatasetPermissionEnum:
+    if permission is None:
+        return DatasetPermissionEnum.ONLY_ME
+    try:
+        return DatasetPermissionEnum(permission)
+    except ValueError as exc:
+        raise ValueError(f"Unsupported legacy dataset permission: {permission}") from exc
+
+
+def _rbac_dataset_scope_for_legacy_permission(permission: DatasetPermissionEnum) -> RBACResourceWhitelistScope:
+    if permission is DatasetPermissionEnum.ALL_TEAM:
+        return RBACResourceWhitelistScope.ALL
+    if permission in {DatasetPermissionEnum.ONLY_ME, DatasetPermissionEnum.PARTIAL_TEAM}:
+        return RBACResourceWhitelistScope.SPECIFIC
+    raise ValueError(f"Unsupported legacy dataset permission: {permission}")
+
+
+def _emit_dataset_permission_migration_event(payload: dict[str, object]) -> None:
+    click.echo(json.dumps(payload, sort_keys=True))
+
+
+@click.command(
+    "rbac-migrate-dataset-permissions",
+    help=(
+        "Migrate legacy dataset permission scopes and partial members into RBAC dataset access bindings. "
+        "Side effect: replacing each dataset whitelist clears existing per-user policy bindings; "
+        "the command then recreates legacy partial-member default bindings."
+    ),
+)
+@click.option("--tenant-id", help="Only migrate datasets in a single workspace.")
+@click.option("--dataset-id", help="Only migrate a single dataset.")
+@click.option("--batch-size", default=500, show_default=True, type=click.IntRange(min=1))
+@click.option(
+    "--dry-run/--apply",
+    default=True,
+    show_default=True,
+    help="Preview the migration without writing RBAC bindings. Use --apply to write changes.",
+)
+def migrate_dataset_permissions_to_rbac(
+    tenant_id: str | None,
+    dataset_id: str | None,
+    batch_size: int,
+    dry_run: bool,
+) -> None:
+    """Backfill RBAC dataset access config from legacy `Dataset.permission`.
+
+    Legacy mapping:
+    - all_team_members -> RBAC dataset whitelist scope "all"
+    - partial_members  -> RBAC dataset whitelist scope "specific" plus each partial member gets the
+      virtual default policy
+    - only_me          -> RBAC dataset whitelist scope "specific" with no member policy bindings
+
+    The command replaces each dataset's RBAC whitelist scope first. RBAC clears
+    existing per-user policy bindings during that replace, then this command
+    recreates the legacy partial-member default bindings. Re-running it is
+    therefore idempotent for a dataset's current legacy configuration.
+    """
+    click.echo(click.style("Starting RBAC dataset permission migration.", fg="green"))
+
+    scanned_count = 0
+    scope_migrated_count = 0
+    user_policy_migrated_count = 0
+    partial_dataset_count = 0
+
+    last_dataset_id: str | None = None
+    while True:
+        with session_factory.create_session() as session:
+            stmt = (
+                select(Dataset.id, Dataset.tenant_id, Dataset.permission, Dataset.created_by)
+                .order_by(Dataset.id.asc())
+                .limit(batch_size)
+            )
+            if tenant_id:
+                stmt = stmt.where(Dataset.tenant_id == tenant_id)
+            if dataset_id:
+                stmt = stmt.where(Dataset.id == dataset_id)
+            if last_dataset_id:
+                stmt = stmt.where(Dataset.id > last_dataset_id)
+
+            dataset_rows = list(session.execute(stmt).all())
+            if not dataset_rows:
+                break
+
+            dataset_ids = [str(row.id) for row in dataset_rows]
+            partial_members_by_dataset_id: dict[str, list[str]] = {item: [] for item in dataset_ids}
+            permission_rows = session.execute(
+                select(DatasetPermission.dataset_id, DatasetPermission.account_id).where(
+                    DatasetPermission.dataset_id.in_(dataset_ids)
+                )
+            ).all()
+            for row in permission_rows:
+                partial_members_by_dataset_id[str(row.dataset_id)].append(str(row.account_id))
+
+        for dataset in dataset_rows:
+            workspace_id = str(dataset.tenant_id)
+            current_dataset_id = str(dataset.id)
+            operator_account_id = str(dataset.created_by)
+            permission_value = _dataset_permission_enum(dataset.permission)
+            scope = _rbac_dataset_scope_for_legacy_permission(permission_value)
+            partial_member_ids = sorted(set(partial_members_by_dataset_id[current_dataset_id]))
+            should_bind_partial_members = permission_value is DatasetPermissionEnum.PARTIAL_TEAM
+
+            click.echo(
+                f"tenant={workspace_id} dataset={current_dataset_id} "
+                f"operator={operator_account_id} "
+                f"legacy_permission={permission_value} -> rbac_scope={scope} "
+                f"partial_members={len(partial_member_ids) if should_bind_partial_members else 0}"
+            )
+
+            scanned_count += 1
+            replace_whitelist_payload = ReplaceMemberBindings(scope=scope)
+            if dry_run:
+                _emit_dataset_permission_migration_event(
+                    {
+                        "event": "dataset_permission_migration_proposed_change",
+                        "action": "replace_whitelist",
+                        "dry_run": True,
+                        "tenant_id": workspace_id,
+                        "dataset_id": current_dataset_id,
+                        "operator_account_id": operator_account_id,
+                        "before": {
+                            "legacy_dataset_permission": permission_value.value,
+                            "legacy_partial_member_ids": partial_member_ids if should_bind_partial_members else [],
+                        },
+                        "after": {
+                            "rbac_whitelist_scope": scope.value,
+                        },
+                        "call": {
+                            "method": "RBACService.DatasetAccess.replace_whitelist",
+                            "kwargs": {
+                                "tenant_id": workspace_id,
+                                "account_id": operator_account_id,
+                                "dataset_id": current_dataset_id,
+                                "payload": replace_whitelist_payload.model_dump(mode="json"),
+                            },
+                        },
+                    }
+                )
+            if not dry_run:
+                RBACService.DatasetAccess.replace_whitelist(
+                    tenant_id=workspace_id,
+                    account_id=operator_account_id,
+                    dataset_id=current_dataset_id,
+                    payload=replace_whitelist_payload,
+                )
+                scope_migrated_count += 1
+
+            if should_bind_partial_members:
+                partial_dataset_count += 1
+                for member_account_id in partial_member_ids:
+                    replace_user_access_policies_payload = ReplaceUserAccessPolicies(
+                        access_policy_ids=[_RBAC_DEFAULT_ACCESS_POLICY_ID],
+                    )
+                    if dry_run:
+                        _emit_dataset_permission_migration_event(
+                            {
+                                "event": "dataset_permission_migration_proposed_change",
+                                "action": "replace_user_access_policies",
+                                "dry_run": True,
+                                "tenant_id": workspace_id,
+                                "dataset_id": current_dataset_id,
+                                "operator_account_id": operator_account_id,
+                                "target_account_id": member_account_id,
+                                "before": {
+                                    "legacy_dataset_permission": permission_value.value,
+                                    "legacy_partial_member_id": member_account_id,
+                                },
+                                "after": {
+                                    "rbac_user_access_policy_ids": [_RBAC_DEFAULT_ACCESS_POLICY_ID],
+                                },
+                                "call": {
+                                    "method": "RBACService.DatasetAccess.replace_user_access_policies",
+                                    "kwargs": {
+                                        "tenant_id": workspace_id,
+                                        "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", exclude_unset=True
+                                        ),
+                                    },
+                                },
+                            }
+                        )
+                        continue
+                    RBACService.DatasetAccess.replace_user_access_policies(
+                        tenant_id=workspace_id,
+                        account_id=operator_account_id,
+                        dataset_id=current_dataset_id,
+                        target_account_id=member_account_id,
+                        payload=replace_user_access_policies_payload,
+                    )
+                    user_policy_migrated_count += 1
+
+        last_dataset_id = dataset_ids[-1]
+
+        if dataset_id:
+            break
+
+    if scanned_count == 0:
+        click.echo(click.style("No datasets found for migration.", fg="yellow"))
+        return
 
     if dry_run:
-        click.echo(click.style("Dry run completed. No RBAC bindings were written.", fg="yellow"))
+        click.echo(
+            click.style(
+                f"Dry run completed. Scanned {scanned_count} datasets; "
+                f"{partial_dataset_count} partial-member datasets would be migrated.",
+                fg="yellow",
+            )
+        )
     else:
-        click.echo(click.style(f"RBAC member-role migration completed. Migrated {migrated_count} members.", fg="green"))
+        click.echo(
+            click.style(
+                "RBAC dataset permission migration completed. "
+                f"Scanned {scanned_count} datasets, migrated {scope_migrated_count} scopes, "
+                f"wrote {user_policy_migrated_count} user default-policy bindings.",
+                fg="green",
+            )
+        )

api/commands/retention.py

@@ -35,12 +35,6 @@ class WorkflowRunArchiveTenantPlan(TypedDict):
     unpaid_tenant_ids: list[str]
 
 
-def _normalize_utc_datetime(value: datetime.datetime) -> datetime.datetime:
-    if value.tzinfo is None:
-        return value.replace(tzinfo=datetime.UTC)
-    return value.astimezone(datetime.UTC)
-
-
 def _parse_tenant_prefixes(prefixes: str | None) -> list[str]:
     if not prefixes:
         return []
@@ -162,16 +156,11 @@ def _resolve_archive_time_range(
             raise click.UsageError("Choose either day offsets or explicit dates, not both.")
         if from_days_ago <= to_days_ago:
             raise click.UsageError("--from-days-ago must be greater than --to-days-ago.")
-        now = datetime.datetime.now(datetime.UTC)
+        now = datetime.datetime.now()
         start_from = now - datetime.timedelta(days=from_days_ago)
         end_before = now - datetime.timedelta(days=to_days_ago)
         before_days = 0
 
-    if start_from is not None:
-        start_from = _normalize_utc_datetime(start_from)
-    if end_before is not None:
-        end_before = _normalize_utc_datetime(end_before)
-
     if start_from and end_before and start_from >= end_before:
         raise click.UsageError("--start-from must be earlier than --end-before.")
 
@@ -413,13 +402,6 @@ def archive_workflow_runs_plan(
             fg="white",
         )
     )
-    click.echo(
-        click.style(
-            "fixed_archive_window="
-            f"{start_from.isoformat() if start_from else 'unbounded'},{plan_end_before.isoformat()}",
-            fg="white",
-        )
-    )
     click.echo("tenant_prefix,total_tenants,workflow_runs,workflow_node_executions,paid_tenants,unpaid_tenants")
     for row in rows:
         click.echo(
@@ -469,7 +451,7 @@ def archive_workflow_runs_plan(
     default=None,
     help="Archive runs created before this timestamp (UTC if no timezone).",
 )
-@click.option("--batch-size", default=10000, show_default=True, help="Maximum workflow runs per archive bundle.")
+@click.option("--batch-size", default=100, show_default=True, help="Maximum workflow runs per archive bundle.")
 @click.option(
     "--workers",
     default=1,
@@ -539,7 +521,6 @@ def archive_workflow_runs(
         )
     )
 
-    uses_relative_window = start_from is None and end_before is None
     try:
         before_days, start_from, end_before = _resolve_archive_time_range(
             before_days=before_days,
@@ -565,14 +546,6 @@ def archive_workflow_runs(
     if delete_after_archive:
         click.echo(click.style("delete-after-archive is not supported by bundle archive.", fg="red"))
         return
-    if uses_relative_window:
-        click.echo(
-            click.style(
-                "Relative archive windows are evaluated at command start. For multi-day prefix/shard rollout, "
-                "reuse absolute --start-from/--end-before values from archive-workflow-runs-plan.",
-                fg="yellow",
-            )
-        )
 
     try:
         tenant_plan = _resolve_archive_tenant_ids_from_plan(

api/configs/enterprise/__init__.py

@@ -34,6 +34,12 @@ class EnterpriseFeatureConfig(BaseSettings):
         default=False,
     )
 
+    ENTERPRISE_RBAC_REQUEST_TIMEOUT: int = Field(
+        ge=1,
+        description="Maximum timeout in seconds for inner RBAC requests.",
+        default=30,
+    )
+
 
 class EnterpriseTelemetryConfig(BaseSettings):
     """

api/configs/extra/agent_backend_config.py

@@ -36,8 +36,8 @@ class AgentBackendConfig(BaseSettings):
         description=(
             "Inject the dify.drive layer (Skills & Files drive manifest declaration) "
             "into Agent runs. The declaration is an index only — the agent backend "
-            "pulls the actual SKILL.md / files through the back proxy. Set this to "
-            "false only when temporarily rolling back the drive integration."
+            "pulls the actual SKILL.md / files through the back proxy. Keep it off "
+            "until the agent backend registers the dify.drive layer type."
         ),
-        default=True,
+        default=False,
     )

api/controllers/common/fields.py

@@ -183,7 +183,6 @@ class Site(BaseModel):
     description: str | None = None
     copyright: str | None = None
     privacy_policy: str | None = None
-    input_placeholder: str | None = None
     custom_disclaimer: str | None = None
     default_language: str
     show_workflow_steps: bool

api/controllers/console/agent/app_helpers.py

@@ -6,15 +6,5 @@ from services.agent.roster_service import AgentRosterService
 
 
 def resolve_agent_app_model(*, tenant_id: str, agent_id: UUID) -> App:
-    """Resolve a roster Agent's public Agent App."""
+    """Resolve the hidden Agent App backing an Agent Console resource."""
     return AgentRosterService(db.session).get_agent_app_model(tenant_id=tenant_id, agent_id=str(agent_id))
-
-
-def resolve_agent_runtime_app_model(*, tenant_id: str, agent_id: UUID) -> App:
-    """Resolve the App that backs an Agent runtime surface.
-
-    This accepts both roster Agent Apps and workflow-only inline Agents with a
-    hidden backing App.
-    """
-
-    return AgentRosterService(db.session).get_agent_runtime_app_model(tenant_id=tenant_id, agent_id=str(agent_id))

api/controllers/console/agent/composer.py

@@ -1,10 +1,10 @@
 from uuid import UUID
 
-from flask import request
 from flask_restx import Resource
 
-from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
+from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
+from controllers.console.agent.app_helpers import resolve_agent_app_model
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import (
     RBACPermission,
@@ -28,15 +28,9 @@ from libs.login import login_required
 from models.model import App, AppMode
 from services.agent.composer_service import AgentComposerService
 from services.agent.composer_validator import ComposerConfigValidator
-from services.entities.agent_entities import (
-    ComposerSavePayload,
-    WorkflowAgentComposerQuery,
-    WorkflowComposerCopyFromRosterPayload,
-)
+from services.entities.agent_entities import ComposerSavePayload, WorkflowComposerCopyFromRosterPayload
 
-register_schema_models(
-    console_ns, ComposerSavePayload, WorkflowAgentComposerQuery, WorkflowComposerCopyFromRosterPayload
-)
+register_schema_models(console_ns, ComposerSavePayload, WorkflowComposerCopyFromRosterPayload)
 register_response_schema_models(
     console_ns,
     AgentAppComposerResponse,
@@ -47,26 +41,27 @@ register_response_schema_models(
 )
 
 
+def _resolve_agent_app_id(*, tenant_id: str, agent_id: UUID) -> str:
+    return resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id).id
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/agent-composer")
 class WorkflowAgentComposerApi(Resource):
     @console_ns.response(
         200, "Workflow agent composer state", console_ns.models[WorkflowAgentComposerResponse.__name__]
     )
-    @console_ns.doc(params=query_params_from_model(WorkflowAgentComposerQuery))
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.WORKFLOW, AppMode.ADVANCED_CHAT])
     @with_current_tenant_id
     def get(self, tenant_id: str, app_model: App, node_id: str):
-        query = WorkflowAgentComposerQuery.model_validate(request.args.to_dict(flat=True))
         return dump_response(
             WorkflowAgentComposerResponse,
             AgentComposerService.load_workflow_composer(
                 tenant_id=tenant_id,
                 app_id=app_model.id,
                 node_id=node_id,
-                snapshot_id=query.snapshot_id,
             ),
         )
 
@@ -142,7 +137,6 @@ class WorkflowAgentComposerValidateApi(Resource):
     def post(self, tenant_id: str, app_model: App, node_id: str):
         payload = ComposerSavePayload.model_validate(console_ns.payload or {})
         ComposerConfigValidator.validate_publish_payload(payload)
-        AgentComposerService.validate_knowledge_datasets(tenant_id=tenant_id, agent_soul=payload.agent_soul)
         findings = AgentComposerService.collect_validation_findings(
             tenant_id=tenant_id,
             payload=payload,
@@ -234,9 +228,10 @@ class AgentComposerApi(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def get(self, tenant_id: str, agent_id: UUID):
+        app_id = _resolve_agent_app_id(tenant_id=tenant_id, agent_id=agent_id)
         return dump_response(
             AgentAppComposerResponse,
-            AgentComposerService.load_agent_composer(tenant_id=tenant_id, agent_id=str(agent_id)),
+            AgentComposerService.load_agent_app_composer(tenant_id=tenant_id, app_id=app_id),
         )
 
     @console_ns.expect(console_ns.models[ComposerSavePayload.__name__])
@@ -249,12 +244,13 @@ class AgentComposerApi(Resource):
     @with_current_user_id
     @with_current_tenant_id
     def put(self, tenant_id: str, account_id: str, agent_id: UUID):
+        app_id = _resolve_agent_app_id(tenant_id=tenant_id, agent_id=agent_id)
         payload = ComposerSavePayload.model_validate(console_ns.payload or {})
         return dump_response(
             AgentAppComposerResponse,
-            AgentComposerService.save_agent_composer(
+            AgentComposerService.save_agent_app_composer(
                 tenant_id=tenant_id,
-                agent_id=str(agent_id),
+                app_id=app_id,
                 account_id=account_id,
                 payload=payload,
             ),
@@ -272,10 +268,9 @@ class AgentComposerValidateApi(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def post(self, tenant_id: str, agent_id: UUID):
-        AgentComposerService.load_agent_composer(tenant_id=tenant_id, agent_id=str(agent_id))
+        _resolve_agent_app_id(tenant_id=tenant_id, agent_id=agent_id)
         payload = ComposerSavePayload.model_validate(console_ns.payload or {})
         ComposerConfigValidator.validate_publish_payload(payload)
-        AgentComposerService.validate_knowledge_datasets(tenant_id=tenant_id, agent_soul=payload.agent_soul)
         findings = AgentComposerService.collect_validation_findings(
             tenant_id=tenant_id,
             payload=payload,
@@ -295,11 +290,12 @@ class AgentComposerCandidatesApi(Resource):
     @with_current_user_id
     @with_current_tenant_id
     def get(self, tenant_id: str, current_user_id: str, agent_id: UUID):
+        app_id = _resolve_agent_app_id(tenant_id=tenant_id, agent_id=agent_id)
         return dump_response(
             AgentComposerCandidatesResponse,
             AgentComposerService.get_agent_app_candidates(
                 tenant_id=tenant_id,
-                agent_id=str(agent_id),
+                app_id=app_id,
                 user_id=current_user_id,
             ),
         )

api/controllers/console/agent/roster.py

@@ -7,7 +7,7 @@ from sqlalchemy import func, select
 
 from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
 from controllers.console import console_ns
-from controllers.console.agent.app_helpers import resolve_agent_app_model, resolve_agent_runtime_app_model
+from controllers.console.agent.app_helpers import resolve_agent_app_model
 from controllers.console.apikey import ApiKeyItem, ApiKeyList, BaseApiKeyListResource, BaseApiKeyResource
 from controllers.console.app.app import (
     AppDetailWithSite as GenericAppDetailWithSite,
@@ -54,10 +54,8 @@ from libs.datetime_utils import parse_time_range
 from libs.helper import dump_response
 from libs.login import login_required
 from models import Account
-from models.agent import Agent, AgentStatus
 from models.enums import ApiTokenType
 from models.model import ApiToken, App, IconType
-from services.agent.composer_service import AgentComposerService
 from services.agent.errors import AgentNotFoundError
 from services.agent.observability_service import (
     AgentLogQueryParams,
@@ -67,7 +65,7 @@ from services.agent.observability_service import (
 from services.agent.roster_service import AgentRosterService
 from services.app_service import AppListParams, AppService, CreateAppParams
 from services.enterprise.enterprise_service import EnterpriseService
-from services.entities.agent_entities import ComposerSavePayload, RosterListQuery
+from services.entities.agent_entities import RosterListQuery
 from services.feature_service import FeatureService
 
 
@@ -234,8 +232,6 @@ class AgentStatisticsQuery(BaseModel):
 
 class AgentAppPartial(GenericAppPartial):
     app_id: str | None = None
-    backing_app_id: str | None = None
-    hidden_app_backed: bool = False
     debug_conversation_id: str | None = None
     role: str | None = None
     active_config_is_published: bool = False
@@ -245,8 +241,6 @@ class AgentAppPartial(GenericAppPartial):
 
 class AgentAppDetailWithSite(GenericAppDetailWithSite):
     app_id: str | None = None
-    backing_app_id: str | None = None
-    hidden_app_backed: bool = False
     debug_conversation_id: str | None = None
     role: str | None = None
     active_config_is_published: bool = False
@@ -256,36 +250,6 @@ class AgentDebugConversationRefreshResponse(BaseModel):
     debug_conversation_id: str
 
 
-class AgentPublishPayload(BaseModel):
-    version_note: str | None = Field(default=None, description="Optional note for this published Agent version")
-
-
-class AgentPublishResponse(BaseModel):
-    result: str
-    active_config_snapshot_id: str
-    active_config_snapshot: dict[str, object] | None = None
-    draft: dict[str, object] | None = None
-
-
-class AgentBuildDraftCheckoutPayload(BaseModel):
-    force: bool = Field(default=False, description="Overwrite the existing current-user build draft")
-
-
-class AgentBuildDraftResponse(BaseModel):
-    variant: str
-    draft: dict[str, object]
-    agent_soul: dict[str, object]
-
-
-class AgentBuildDraftApplyResponse(BaseModel):
-    result: str
-    draft: dict[str, object]
-
-
-class AgentSimpleResultResponse(BaseModel):
-    result: str
-
-
 class AgentAppPagination(GenericAppPagination):
     data: list[AgentAppPartial] = Field(  # type: ignore[assignment]  # pyrefly: ignore[bad-override-mutable-attribute]
         validation_alias=AliasChoices("items", "data")
@@ -297,9 +261,6 @@ register_schema_models(
     AgentAppCreatePayload,
     AgentAppUpdatePayload,
     AgentAppCopyPayload,
-    AgentPublishPayload,
-    AgentBuildDraftCheckoutPayload,
-    ComposerSavePayload,
     AgentApiStatusPayload,
     AgentInviteOptionsQuery,
     AgentLogsQuery,
@@ -316,10 +277,6 @@ register_response_schema_models(
     AgentAppDetailWithSite,
     AgentAppPartial,
     AgentDebugConversationRefreshResponse,
-    AgentPublishResponse,
-    AgentBuildDraftResponse,
-    AgentBuildDraftApplyResponse,
-    AgentSimpleResultResponse,
     AgentConfigSnapshotDetailResponse,
     AgentConfigSnapshotListResponse,
     AgentConfigSnapshotRestoreResponse,
@@ -337,7 +294,7 @@ def _agent_roster_service() -> AgentRosterService:
     return AgentRosterService(db.session)
 
 
-def _serialize_agent_app_detail(app_model, *, current_user: Account, agent_id: str | None = None) -> dict:
+def _serialize_agent_app_detail(app_model, *, current_user: Account) -> dict:
     """Serialize an Agent App detail using roster-only DTOs.
 
     `/agent` responses are roster-shaped rather than raw app-shaped: `id`
@@ -354,23 +311,11 @@ def _serialize_agent_app_detail(app_model, *, current_user: Account, agent_id: s
 
     roster_service = _agent_roster_service()
     payload = AgentAppDetailWithSite.model_validate(app_model, from_attributes=True).model_dump(mode="json")
-    agent = (
-        db.session.scalar(
-            select(Agent).where(
-                Agent.tenant_id == app_model.tenant_id,
-                Agent.id == agent_id,
-                Agent.status == AgentStatus.ACTIVE,
-            )
-        )
-        if agent_id
-        else roster_service.get_app_backing_agent(tenant_id=app_model.tenant_id, app_id=str(app_model.id))
-    )
+    agent = roster_service.get_app_backing_agent(tenant_id=app_model.tenant_id, app_id=str(app_model.id))
     if not agent:
         raise AgentNotFoundError()
     payload.pop("bound_agent_id", None)
-    payload["app_id"] = agent.app_id
-    payload["backing_app_id"] = roster_service.runtime_backing_app_id(agent)
-    payload["hidden_app_backed"] = bool(agent.backing_app_id and agent.backing_app_id != agent.app_id)
+    payload["app_id"] = str(app_model.id)
     payload["id"] = agent.id
     payload["debug_conversation_id"] = roster_service.get_or_create_agent_app_debug_conversation_id(
         tenant_id=app_model.tenant_id,
@@ -420,8 +365,6 @@ def _serialize_agent_app_pagination(app_pagination, *, tenant_id: str, current_u
         agent = agents_by_app_id.get(app_id)
         if agent:
             item["app_id"] = app_id
-            item["backing_app_id"] = agent.backing_app_id or app_id
-            item["hidden_app_backed"] = False
             item["id"] = agent.id
             item["debug_conversation_id"] = debug_conversation_ids_by_agent_id.get(agent.id)
             item["role"] = agent.role or ""
@@ -573,8 +516,8 @@ class AgentAppApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def get(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
-        return _serialize_agent_app_detail(app_model, current_user=current_user, agent_id=str(agent_id))
+        app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        return _serialize_agent_app_detail(app_model, current_user=current_user)
 
     @console_ns.expect(console_ns.models[AgentAppUpdatePayload.__name__])
     @console_ns.response(200, "Agent app updated successfully", console_ns.models[AgentAppDetailWithSite.__name__])
@@ -640,112 +583,6 @@ class AgentDebugConversationRefreshApi(Resource):
         )
 
 
-@console_ns.route("/agent/<uuid:agent_id>/publish")
-class AgentPublishApi(Resource):
-    @console_ns.expect(console_ns.models[AgentPublishPayload.__name__])
-    @console_ns.response(200, "Agent draft published", console_ns.models[AgentPublishResponse.__name__])
-    @console_ns.response(403, "Insufficient permissions")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    @with_current_user
-    @with_current_tenant_id
-    def post(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        args = AgentPublishPayload.model_validate(console_ns.payload or {})
-        return AgentComposerService.publish_agent_app_draft(
-            tenant_id=tenant_id,
-            agent_id=str(agent_id),
-            account_id=current_user.id,
-            version_note=args.version_note,
-        )
-
-
-@console_ns.route("/agent/<uuid:agent_id>/build-draft/checkout")
-class AgentBuildDraftCheckoutApi(Resource):
-    @console_ns.expect(console_ns.models[AgentBuildDraftCheckoutPayload.__name__])
-    @console_ns.response(200, "Agent build draft checked out", console_ns.models[AgentBuildDraftResponse.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    @with_current_user
-    @with_current_tenant_id
-    def post(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        args = AgentBuildDraftCheckoutPayload.model_validate(console_ns.payload or {})
-        return AgentComposerService.checkout_agent_app_build_draft(
-            tenant_id=tenant_id,
-            agent_id=str(agent_id),
-            account_id=current_user.id,
-            force=args.force,
-        )
-
-
-@console_ns.route("/agent/<uuid:agent_id>/build-draft")
-class AgentBuildDraftApi(Resource):
-    @console_ns.response(200, "Agent build draft", console_ns.models[AgentBuildDraftResponse.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    @with_current_user
-    @with_current_tenant_id
-    def get(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        return AgentComposerService.load_agent_app_build_draft(
-            tenant_id=tenant_id,
-            agent_id=str(agent_id),
-            account_id=current_user.id,
-        )
-
-    @console_ns.expect(console_ns.models[ComposerSavePayload.__name__])
-    @console_ns.response(200, "Agent build draft saved", console_ns.models[AgentBuildDraftResponse.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    @with_current_user
-    @with_current_tenant_id
-    def put(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        payload = ComposerSavePayload.model_validate(console_ns.payload or {})
-        return AgentComposerService.save_agent_app_build_draft(
-            tenant_id=tenant_id,
-            agent_id=str(agent_id),
-            account_id=current_user.id,
-            payload=payload,
-        )
-
-    @console_ns.response(200, "Agent build draft discarded", console_ns.models[AgentSimpleResultResponse.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    @with_current_user
-    @with_current_tenant_id
-    def delete(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        return AgentComposerService.discard_agent_app_build_draft(
-            tenant_id=tenant_id,
-            agent_id=str(agent_id),
-            account_id=current_user.id,
-        )
-
-
-@console_ns.route("/agent/<uuid:agent_id>/build-draft/apply")
-class AgentBuildDraftApplyApi(Resource):
-    @console_ns.response(200, "Agent build draft applied", console_ns.models[AgentBuildDraftApplyResponse.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    @with_current_user
-    @with_current_tenant_id
-    def post(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        return AgentComposerService.apply_agent_app_build_draft(
-            tenant_id=tenant_id,
-            agent_id=str(agent_id),
-            account_id=current_user.id,
-        )
-
-
 @console_ns.route("/agent/<uuid:agent_id>/copy")
 class AgentAppCopyApi(Resource):
     @console_ns.expect(console_ns.models[AgentAppCopyPayload.__name__])
@@ -875,7 +712,7 @@ class AgentLogsApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def get(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         query_data: dict[str, object] = dict(request.args.to_dict(flat=True))
         query_data["sources"] = _multi_query_values("sources", "source")
         query_data["statuses"] = _multi_query_values("statuses", "status")
@@ -912,7 +749,7 @@ class AgentLogMessagesApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def get(self, tenant_id: str, current_user: Account, agent_id: UUID, conversation_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         query_data: dict[str, object] = dict(request.args.to_dict(flat=True))
         query_data["sources"] = _multi_query_values("sources", "source")
         query_data["statuses"] = _multi_query_values("statuses", "status")
@@ -949,7 +786,7 @@ class AgentLogSourcesApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def get(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         payload = _agent_observability_service().list_log_sources(app=app_model, agent_id=str(agent_id))
         return dump_response(AgentLogSourceListResponse, payload)
 
@@ -968,7 +805,7 @@ class AgentStatisticsSummaryApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def get(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         query = AgentStatisticsQuery.model_validate(request.args.to_dict(flat=True))
         timezone = current_user.timezone or "UTC"
         start, end = _parse_observability_time_range(query.start, query.end, current_user)

api/controllers/console/app/agent.py

@@ -13,7 +13,7 @@ from controllers.common.schema import (
     register_schema_models,
 )
 from controllers.console import console_ns
-from controllers.console.agent.app_helpers import resolve_agent_runtime_app_model
+from controllers.console.agent.app_helpers import resolve_agent_app_model
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import (
     RBACPermission,
@@ -351,7 +351,7 @@ class AgentSkillUploadByAgentApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def post(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         return _upload_skill_for_app(current_user=current_user, app_model=app_model)
 
 
@@ -394,7 +394,7 @@ class AgentDriveFilesByAgentApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def post(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         return _commit_drive_file_for_app(current_user=current_user, app_model=app_model, allow_node_id=False)
 
     @console_ns.doc("delete_agent_drive_file_by_agent")
@@ -407,7 +407,7 @@ class AgentDriveFilesByAgentApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def delete(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         return _delete_drive_file_for_app(current_user=current_user, app_model=app_model, allow_node_id=False)
 
 
@@ -454,7 +454,7 @@ class AgentSkillByAgentApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def delete(self, tenant_id: str, current_user: Account, agent_id: UUID, slug: str):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         return _delete_skill_for_app(current_user=current_user, app_model=app_model, slug=slug, allow_node_id=False)
 
 
@@ -494,7 +494,7 @@ class AgentSkillInferToolsByAgentApi(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def post(self, tenant_id: str, agent_id: UUID, slug: str):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         return _infer_skill_tools_for_app(app_model=app_model, slug=slug)
 
 

api/controllers/console/app/agent_app_feature.py

@@ -17,7 +17,7 @@ from pydantic import BaseModel, Field
 from controllers.common.fields import SimpleResultResponse
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
-from controllers.console.agent.app_helpers import resolve_agent_runtime_app_model
+from controllers.console.agent.app_helpers import resolve_agent_app_model
 from controllers.console.wraps import (
     RBACPermission,
     RBACResourceScope,
@@ -87,7 +87,7 @@ class AgentAppFeatureConfigResource(Resource):
     @with_current_user
     @with_current_tenant_id
     def post(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         args = AgentAppFeaturesPayload.model_validate(console_ns.payload or {})
 
         new_app_model_config = AgentAppFeatureConfigService.update_features(

api/controllers/console/app/agent_app_sandbox.py

@@ -22,7 +22,7 @@ from controllers.common.schema import (
     register_schema_models,
 )
 from controllers.console import console_ns
-from controllers.console.agent.app_helpers import resolve_agent_runtime_app_model
+from controllers.console.agent.app_helpers import resolve_agent_app_model
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required, with_current_tenant_id
 from fields.base import ResponseModel
@@ -144,7 +144,7 @@ class AgentAppSandboxListResource(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def get(self, tenant_id: str, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         query = query_params_from_request(AgentSandboxListQuery)
         try:
             result = AgentAppSandboxService().list_files(
@@ -169,7 +169,7 @@ class AgentAppSandboxReadResource(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def get(self, tenant_id: str, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         query = query_params_from_request(AgentSandboxFileQuery)
         try:
             result = AgentAppSandboxService().read_file(
@@ -194,7 +194,7 @@ class AgentAppSandboxUploadResource(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def post(self, tenant_id: str, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         payload = AgentSandboxUploadPayload.model_validate(request.get_json(silent=True) or {})
         try:
             result = AgentAppSandboxService().upload_file(

api/controllers/console/app/agent_drive_inspector.py

@@ -25,7 +25,7 @@ from controllers.common.schema import (
     register_response_schema_models,
 )
 from controllers.console import console_ns
-from controllers.console.agent.app_helpers import resolve_agent_runtime_app_model
+from controllers.console.agent.app_helpers import resolve_agent_app_model
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required, with_current_tenant_id
 from fields.base import ResponseModel
@@ -182,7 +182,7 @@ class AgentDriveListByAgentApi(Resource):
     @with_current_tenant_id
     def get(self, tenant_id: str, agent_id: UUID):
         query = query_params_from_request(AgentDriveListByAgentQuery)
-        resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         try:
             items = AgentDriveService().manifest(tenant_id=tenant_id, agent_id=str(agent_id), prefix=query.prefix)
         except AgentDriveError as exc:
@@ -201,7 +201,7 @@ class AgentDriveSkillListByAgentApi(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def get(self, tenant_id: str, agent_id: UUID):
-        resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         try:
             items = AgentDriveService().list_skills(tenant_id=tenant_id, agent_id=str(agent_id))
         except AgentDriveError as exc:
@@ -220,7 +220,7 @@ class AgentDriveSkillInspectByAgentApi(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def get(self, tenant_id: str, agent_id: UUID, skill_path: str):
-        resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         try:
             return _json_response(
                 AgentDriveService().inspect_skill(
@@ -245,7 +245,7 @@ class AgentDrivePreviewByAgentApi(Resource):
     @with_current_tenant_id
     def get(self, tenant_id: str, agent_id: UUID):
         query = query_params_from_request(AgentDriveFileByAgentQuery)
-        resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         try:
             return AgentDriveService().preview(tenant_id=tenant_id, agent_id=str(agent_id), key=query.key)
         except AgentDriveError as exc:
@@ -264,7 +264,7 @@ class AgentDriveDownloadByAgentApi(Resource):
     @with_current_tenant_id
     def get(self, tenant_id: str, agent_id: UUID):
         query = query_params_from_request(AgentDriveFileByAgentQuery)
-        resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         try:
             url = AgentDriveService().download_url(tenant_id=tenant_id, agent_id=str(agent_id), key=query.key)
         except AgentDriveError as exc:

api/controllers/console/app/app.py

@@ -43,6 +43,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
@@ -69,6 +70,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"]
 
@@ -331,7 +333,7 @@ class ModelConfig(ResponseModel):
         return to_timestamp(value)
 
 
-class AppDetailSiteResponse(ResponseModel):
+class Site(ResponseModel):
     access_token: str | None = Field(default=None, validation_alias="code")
     code: str | None = None
     title: str | None = None
@@ -345,7 +347,6 @@ class AppDetailSiteResponse(ResponseModel):
     customize_domain: str | None = None
     copyright: str | None = None
     privacy_policy: str | None = None
-    input_placeholder: str | None = None
     custom_disclaimer: str | None = None
     customize_token_strategy: str | None = None
     prompt_public: bool | None = None
@@ -462,7 +463,7 @@ class AppDetailWithSite(AppDetail):
     api_base_url: str | None = None
     max_active_requests: int | None = None
     deleted_tools: list[DeletedTool] = Field(default_factory=list)
-    site: AppDetailSiteResponse | None = None
+    site: Site | None = None
     # For Agent App type: the roster Agent backing this app (None otherwise).
     bound_agent_id: str | None = None
     # For Agent App responses exposed through /agent.
@@ -547,7 +548,7 @@ register_schema_models(
     WorkflowPartial,
     ModelConfigPartial,
     ModelConfig,
-    AppDetailSiteResponse,
+    Site,
     DeletedTool,
     AppDetail,
     AppExportResponse,
@@ -667,6 +668,14 @@ class AppListApi(Resource):
 
         app_service = AppService()
         app = app_service.create_app(current_tenant_id, params, current_user)
+        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=app.id)
         permission_keys_map = enterprise_rbac_service.RBACService.AppPermissions.batch_get(
             str(current_tenant_id),
             current_user.id,

api/controllers/console/app/completion.py

@@ -11,7 +11,7 @@ import services
 from controllers.common.fields import GeneratedAppResponse, SimpleResultResponse
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
-from controllers.console.agent.app_helpers import resolve_agent_runtime_app_model
+from controllers.console.agent.app_helpers import resolve_agent_app_model
 from controllers.console.app.error import (
     AppUnavailableError,
     CompletionRequestError,
@@ -93,10 +93,6 @@ class ChatMessagePayload(BaseMessagePayload):
     query: str = Field(..., description="User query")
     conversation_id: str | None = Field(default=None, description="Conversation ID")
     parent_message_id: str | None = Field(default=None, description="Parent message ID")
-    draft_type: Literal["draft", "debug_build"] = Field(
-        default="draft",
-        description="Agent App debug config source. Use debug_build while the Agent is in build mode.",
-    )
 
     @field_validator("conversation_id", "parent_message_id")
     @classmethod
@@ -222,7 +218,7 @@ class AgentChatMessageApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def post(self, current_tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
         return _create_chat_message(
             current_tenant_id=current_tenant_id,
             current_user=current_user,
@@ -258,7 +254,7 @@ class AgentChatMessageStopApi(Resource):
     @with_current_user_id
     @with_current_tenant_id
     def post(self, current_tenant_id: str, current_user_id: str, agent_id: UUID, task_id: str):
-        app_model = resolve_agent_runtime_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
         return _stop_chat_message(current_user_id=current_user_id, app_model=app_model, task_id=task_id)
 
 

api/controllers/console/app/message.py

@@ -13,7 +13,7 @@ from controllers.common.controller_schemas import MessageFeedbackPayload as _Mes
 from controllers.common.fields import SimpleResultResponse, TextFileResponse
 from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
 from controllers.console import console_ns
-from controllers.console.agent.app_helpers import resolve_agent_runtime_app_model
+from controllers.console.agent.app_helpers import resolve_agent_app_model
 from controllers.console.app.error import (
     CompletionRequestError,
     ProviderModelCurrentlyNotSupportError,
@@ -167,16 +167,12 @@ register_schema_models(
     ChatMessagesQuery,
     MessageFeedbackPayload,
     FeedbackExportQuery,
-)
-register_response_schema_models(
-    console_ns,
     AnnotationCountResponse,
     SuggestedQuestionsResponse,
     MessageDetailResponse,
     MessageInfiniteScrollPaginationResponse,
-    SimpleResultResponse,
-    TextFileResponse,
 )
+register_response_schema_models(console_ns, SimpleResultResponse, TextFileResponse)
 
 
 @console_ns.route("/apps/<uuid:app_id>/chat-messages")
@@ -214,7 +210,7 @@ class AgentChatMessageListApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def get(self, current_tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
         return _list_chat_messages(app_model=app_model, current_user=current_user)
 
 
@@ -250,7 +246,7 @@ class AgentMessageFeedbackApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def post(self, current_tenant_id: str, current_user: Account, agent_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
         return _update_message_feedback(current_user=current_user, app_model=app_model)
 
 
@@ -315,7 +311,7 @@ class AgentMessageSuggestedQuestionApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def get(self, current_tenant_id: str, current_user: Account, agent_id: UUID, message_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
         return _get_message_suggested_questions(current_user=current_user, app_model=app_model, message_id=message_id)
 
 
@@ -393,7 +389,7 @@ class AgentMessageApi(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def get(self, current_tenant_id: str, agent_id: UUID, message_id: UUID):
-        app_model = resolve_agent_runtime_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
+        app_model = resolve_agent_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
         return _get_message_detail(app_model=app_model, message_id=message_id)
 
 

api/controllers/console/app/site.py

@@ -40,7 +40,6 @@ class AppSiteUpdatePayload(BaseModel):
     customize_domain: str | None = Field(default=None)
     copyright: str | None = Field(default=None)
     privacy_policy: str | None = Field(default=None)
-    input_placeholder: str | None = Field(default=None)
     custom_disclaimer: str | None = Field(default=None)
     customize_token_strategy: Literal["must", "allow", "not_allow"] | None = Field(default=None)
     prompt_public: bool | None = Field(default=None)
@@ -67,7 +66,6 @@ class AppSiteResponse(ResponseModel):
     customize_domain: str | None = None
     copyright: str | None = None
     privacy_policy: str | None = None
-    input_placeholder: str | None = None
     custom_disclaimer: str | None = None
     customize_token_strategy: str
     prompt_public: bool
@@ -112,7 +110,6 @@ class AppSite(Resource):
             "customize_domain",
             "copyright",
             "privacy_policy",
-            "input_placeholder",
             "custom_disclaimer",
             "customize_token_strategy",
             "prompt_public",

api/controllers/console/app/wraps.py

@@ -19,7 +19,8 @@ from controllers.console.app.error import AppNotFoundError
 from core.db.session_factory import session_factory
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant
-from models import App, AppMode
+from models import App, AppMode, TrialApp
+from services.recommended_app_service import RecommendedAppService
 
 
 def _load_app_model(session: Session, app_id: str) -> App | None:
@@ -41,7 +42,10 @@ def _load_app_model_from_scoped_session(app_id: str) -> App | None:
 
 
 def _load_app_model_with_trial(app_id: str) -> App | None:
-    app_model = db.session.scalar(select(App).where(App.id == app_id, App.status == "normal").limit(1))
+    """Load a normal app through its trial registration without applying current-tenant scope."""
+    app_model = db.session.scalar(
+        select(App).join(TrialApp, TrialApp.app_id == App.id).where(App.id == app_id, App.status == "normal").limit(1)
+    )
     return app_model
 
 
@@ -195,6 +199,8 @@ def get_app_model_with_trial[**P, R](
     *,
     mode: AppMode | list[AppMode] | None = None,
 ) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
+    """Inject an app registered for trial or available from the recommended catalog."""
+
     def decorator(view_func: Callable[P, R]) -> Callable[P, R]:
         @wraps(view_func)
         def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
@@ -207,6 +213,8 @@ def get_app_model_with_trial[**P, R](
             del kwargs["app_id"]
 
             app_model = _load_app_model_with_trial(app_id)
+            if app_model is None:
+                app_model = RecommendedAppService.get_app(app_id, session=db.session())
 
             if not app_model:
                 raise AppNotFoundError()

api/controllers/console/auth/email_register.py

@@ -23,9 +23,9 @@ from libs.password import valid_password
 from models import Account
 from services.account_service import AccountService
 from services.billing_service import BillingService
-from services.errors.account import AccountRegisterError
+from services.errors.account import AccountRegisterError, SeatsLimitExceededError
 
-from ..error import AccountInFreezeError, EmailSendIpLimitError
+from ..error import AccountInFreezeError, EmailSendIpLimitError, SeatsLimitExceeded
 from ..wraps import email_password_login_enabled, email_register_enabled, setup_required
 
 
@@ -208,5 +208,7 @@ class EmailRegisterResetApi(Resource):
                 timezone=timezone,
                 session=db.session,
             )
+        except SeatsLimitExceededError:
+            raise SeatsLimitExceeded()
         except AccountRegisterError:
             raise AccountInFreezeError()

api/controllers/console/auth/login.py

@@ -25,6 +25,7 @@ from controllers.console.error import (
     AccountNotFound,
     EmailSendIpLimitError,
     NotAllowedCreateWorkspace,
+    SeatsLimitExceeded,
     WorkspacesLimitExceeded,
 )
 from controllers.console.wraps import (
@@ -51,7 +52,7 @@ from models.account import Account
 from services.account_service import AccountService, InvitationDetailDict, RegisterService, TenantService
 from services.billing_service import BillingService
 from services.entities.auth_entities import LoginFailureReason, LoginPayloadBase
-from services.errors.account import AccountRegisterError
+from services.errors.account import AccountRegisterError, SeatsLimitExceededError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService
 
@@ -317,6 +318,8 @@ class EmailCodeLoginApi(Resource):
                 )
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()
+            except SeatsLimitExceededError:
+                raise SeatsLimitExceeded()
             except AccountRegisterError:
                 _log_console_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_IN_FREEZE)
                 raise AccountInFreezeError()

api/controllers/console/auth/oauth.py

@@ -25,7 +25,7 @@ from libs.token import (
 from models import Account, AccountStatus
 from services.account_service import AccountService, RegisterService, TenantService
 from services.billing_service import BillingService
-from services.errors.account import AccountNotFoundError, AccountRegisterError
+from services.errors.account import AccountNotFoundError, AccountRegisterError, SeatsLimitExceededError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkSpaceNotFoundError
 from services.feature_service import FeatureService
 
@@ -38,6 +38,7 @@ class OAuthLoginQuery(BaseModel):
     invite_token: str | None = Field(default=None, description="Optional invitation token")
     timezone: str | None = Field(default=None, description="Preferred timezone")
     language: str | None = Field(default=None, description="Preferred interface language")
+    redirect_url: str | None = Field(default=None, description="Relative page to resume after login")
 
 
 class OAuthCallbackQuery(BaseModel):
@@ -87,6 +88,36 @@ def _validated_language(value: str | None) -> str | None:
     return None
 
 
+def _url_origin(url: str) -> tuple[str, str, int] | None:
+    parsed_url = urllib.parse.urlsplit(url)
+    if parsed_url.scheme not in {"http", "https"} or parsed_url.hostname is None:
+        return None
+
+    try:
+        port = parsed_url.port
+    except ValueError:
+        return None
+
+    if port is None:
+        port = 443 if parsed_url.scheme == "https" else 80
+    return parsed_url.scheme, parsed_url.hostname, port
+
+
+def _get_redirect_target(redirect_url: str | None) -> str:
+    if not redirect_url:
+        return dify_config.CONSOLE_WEB_URL
+
+    parsed_url = urllib.parse.urlsplit(redirect_url)
+    normalized_path = redirect_url.lstrip().replace("\\", "/")
+    if not parsed_url.scheme and not parsed_url.netloc and not normalized_path.startswith("//"):
+        return redirect_url
+
+    redirect_origin = _url_origin(redirect_url)
+    if redirect_origin is not None and redirect_origin == _url_origin(dify_config.CONSOLE_WEB_URL):
+        return redirect_url
+    return dify_config.CONSOLE_WEB_URL
+
+
 def _preferred_interface_language(language: str | None = None) -> str:
     if language:
         return language
@@ -109,6 +140,7 @@ class OAuthLogin(Resource):
         invite_token = request.args.get("invite_token") or None
         timezone = _validated_timezone(request.args.get("timezone") or None)
         language = _validated_language(request.args.get("language") or None)
+        redirect_url = request.args.get("redirect_url") or None
         OAUTH_PROVIDERS = get_oauth_providers()
         with current_app.app_context():
             oauth_provider = OAUTH_PROVIDERS.get(provider)
@@ -119,6 +151,7 @@ class OAuthLogin(Resource):
             invite_token=invite_token,
             timezone=timezone,
             language=language,
+            redirect_url=redirect_url,
         )
         return redirect(auth_url)
 
@@ -144,6 +177,7 @@ class OAuthCallback(Resource):
         invite_token = oauth_state.get("invite_token")
         timezone = _validated_timezone(oauth_state.get("timezone"))
         language = _validated_language(oauth_state.get("language"))
+        redirect_url = oauth_state.get("redirect_url")
 
         if not code:
             return {"error": "Authorization code is required"}, 400
@@ -182,6 +216,8 @@ class OAuthCallback(Resource):
                 f"{dify_config.CONSOLE_WEB_URL}/signin"
                 "?message=Workspace not found, please contact system admin to invite you to join in a workspace."
             )
+        except SeatsLimitExceededError:
+            return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Licensed seats limit exceeded.")
         except AccountRegisterError as e:
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message={e.description}")
 
@@ -210,9 +246,9 @@ class OAuthCallback(Resource):
             ip_address=extract_remote_ip(request),
         )
 
-        base_url = dify_config.CONSOLE_WEB_URL
-        query_char = "&" if "?" in base_url else "?"
-        target_url = f"{base_url}{query_char}oauth_new_user={str(oauth_new_user).lower()}"
+        target_url = _get_redirect_target(redirect_url)
+        query_char = "&" if "?" in target_url else "?"
+        target_url = f"{target_url}{query_char}oauth_new_user={str(oauth_new_user).lower()}"
         response = redirect(target_url)
 
         set_access_token_to_cookie(request, response, token_pair.access_token)

api/controllers/console/billing/billing.py

@@ -16,7 +16,6 @@ from controllers.console.wraps import (
     with_current_user,
 )
 from enums.cloud_plan import CloudPlan
-from extensions.ext_database import db
 from libs.login import login_required
 from models import Account
 from services.billing_service import BillingService
@@ -51,7 +50,7 @@ class Subscription(Resource):
     @with_current_tenant_id
     def get(self, current_tenant_id: str, current_user: Account):
         args = SubscriptionQuery.model_validate(request.args.to_dict(flat=True))
-        BillingService.is_tenant_owner_or_admin(db.session, current_user)
+        BillingService.is_tenant_owner_or_admin(current_user)
         return BillingService.get_subscription(args.plan, args.interval, current_user.email, current_tenant_id)
 
 
@@ -65,7 +64,7 @@ class Invoices(Resource):
     @with_current_user
     @with_current_tenant_id
     def get(self, current_tenant_id: str, current_user: Account):
-        BillingService.is_tenant_owner_or_admin(db.session, current_user)
+        BillingService.is_tenant_owner_or_admin(current_user)
         return BillingService.get_invoices(current_user.email, current_tenant_id)
 
 

api/controllers/console/datasets/datasets.py

@@ -50,6 +50,8 @@ from models.provider_ids import ModelProviderID
 from services.api_token_service import ApiTokenCache
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 from services.enterprise import rbac_service as enterprise_rbac_service
+from services.enterprise.rbac_service import RBACResourceWhitelistScope, ReplaceMemberBindings
+from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
 
 register_response_schema_models(console_ns, ApiBaseUrlResponse, SimpleResultResponse, UsageCheckResponse)
 
@@ -565,6 +567,16 @@ class DatasetListApi(Resource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
+        if dify_config.RBAC_ENABLED:
+            if permission == DatasetPermissionEnum.ALL_TEAM:
+                enterprise_rbac_service.RBACService.DatasetAccess.replace_whitelist(
+                    current_tenant_id,
+                    current_user.id,
+                    dataset.id,
+                    ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
+                )
+                initialize_created_app_rbac_access_task.delay(current_tenant_id, current_user.id, dataset_id=dataset.id)
+
         permission_keys_map = enterprise_rbac_service.RBACService.DatasetPermissions.batch_get(
             str(current_tenant_id),
             current_user.id,
@@ -602,7 +614,7 @@ class DatasetApi(Resource):
         if dataset is None:
             raise NotFound("Dataset not found.")
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         permissions = enterprise_rbac_service.RBACService.MyPermissions.get(
@@ -774,7 +786,7 @@ class DatasetQueryApi(Resource):
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -915,7 +927,7 @@ class DatasetRelatedAppListApi(Resource):
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -1194,7 +1206,7 @@ class DatasetPermissionUserListApi(Resource):
         if dataset is None:
             raise NotFound("Dataset not found.")
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 

api/controllers/console/datasets/datasets_document.py

@@ -186,7 +186,7 @@ class DocumentResource(Resource):
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -206,7 +206,7 @@ class DocumentResource(Resource):
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -247,7 +247,7 @@ class GetProcessRuleApi(Resource):
                 raise NotFound("Dataset not found.")
 
             try:
-                DatasetService.check_dataset_permission(dataset, current_user, db.session)
+                DatasetService.check_dataset_permission(dataset, current_user)
             except services.errors.account.NoPermissionError as e:
                 raise Forbidden(str(e))
 
@@ -322,7 +322,7 @@ class DatasetDocumentListApi(Resource):
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -431,7 +431,7 @@ class DatasetDocumentListApi(Resource):
             raise Forbidden()
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -1173,7 +1173,7 @@ class DocumentStatusApi(DocumentResource):
         DatasetService.check_dataset_model_setting(dataset)
 
         # check user's permission
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         document_ids = request.args.getlist("document_id")
 
@@ -1440,7 +1440,7 @@ class DocumentGenerateSummaryApi(Resource):
             raise Forbidden()
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -1537,7 +1537,7 @@ class DocumentSummaryStatusApi(DocumentResource):
 
         # Check permissions
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 

api/controllers/console/datasets/datasets_segments.py

@@ -181,7 +181,7 @@ class DatasetDocumentSegmentListApi(Resource):
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -302,7 +302,7 @@ class DatasetDocumentSegmentListApi(Resource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         SegmentService.delete_segments(segment_ids, document, dataset)
@@ -345,7 +345,7 @@ class DatasetDocumentSegmentApi(Resource):
             raise Forbidden()
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
@@ -421,7 +421,7 @@ class DatasetDocumentSegmentAddApi(Resource):
             except ProviderTokenNotInitError as ex:
                 raise ProviderNotInitializeError(ex.description)
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         # validate args
@@ -494,7 +494,7 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         # validate args
@@ -550,7 +550,7 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         SegmentService.delete_segment(segment, document, dataset)
@@ -687,7 +687,7 @@ class ChildChunkAddApi(Resource):
             except ProviderTokenNotInitError as ex:
                 raise ProviderNotInitializeError(ex.description)
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         # validate args
@@ -789,7 +789,7 @@ class ChildChunkAddApi(Resource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         # validate args
@@ -862,7 +862,7 @@ class ChildChunkUpdateApi(Resource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         try:
@@ -930,7 +930,7 @@ class ChildChunkUpdateApi(Resource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         # validate args

api/controllers/console/datasets/external.py

@@ -382,7 +382,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 

api/controllers/console/datasets/hit_testing_base.py

@@ -90,7 +90,7 @@ class DatasetsHitTestingBase:
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 

api/controllers/console/datasets/metadata.py

@@ -64,7 +64,7 @@ class DatasetMetadataCreateApi(Resource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata = MetadataService.create_metadata(
             db.session(), dataset_id_str, metadata_args, current_user, current_tenant_id
@@ -108,7 +108,7 @@ class DatasetMetadataApi(Resource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata = MetadataService.update_metadata_name(
             db.session(), dataset_id_str, metadata_id_str, name, current_user, current_tenant_id
@@ -128,7 +128,7 @@ class DatasetMetadataApi(Resource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         MetadataService.delete_metadata(db.session(), dataset_id_str, metadata_id_str)
         # Frontend callers only await success and invalidate metadata caches; no response body is consumed.
@@ -165,7 +165,7 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         match action:
             case "enable":
@@ -194,7 +194,7 @@ class DocumentMetadataEditApi(Resource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata_args = MetadataOperationData.model_validate(console_ns.payload or {})
 

api/controllers/console/error.py

@@ -58,6 +58,12 @@ class WorkspacesLimitExceeded(BaseHTTPException):
     code = 400
 
 
+class SeatsLimitExceeded(BaseHTTPException):
+    error_code = "limit_exceeded"
+    description = "Unable to create account because the licensed seats limit was exceeded"
+    code = 400
+
+
 class AccountBannedError(BaseHTTPException):
     error_code = "account_banned"
     description = "Account is banned."

api/controllers/console/explore/installed_app.py

@@ -85,7 +85,6 @@ def _published_app_filter():
 class InstalledAppInfoResponse(ResponseModel):
     id: str
     name: str | None = None
-    description: str | None = None
     mode: str | None = None
     icon_type: str | None = None
     icon: str | None = None
@@ -124,7 +123,6 @@ class InstalledAppResponse(ResponseModel):
         return {
             "id": _safe_primitive(getattr(value, "id", "")) or "",
             "name": _safe_primitive(getattr(value, "name", None)),
-            "description": _safe_primitive(getattr(value, "description", None)),
             "mode": _safe_primitive(getattr(value, "mode", None)),
             "icon_type": _safe_primitive(getattr(value, "icon_type", None)),
             "icon": _safe_primitive(getattr(value, "icon", None)),

api/controllers/console/explore/recommended_app.py

@@ -106,7 +106,7 @@ class RecommendedAppListApi(Resource):
         language_prefix = _resolve_language(args.language, current_user)
 
         return RecommendedAppListResponse.model_validate(
-            RecommendedAppService.get_recommended_apps_and_categories(db.session, language_prefix),
+            RecommendedAppService.get_recommended_apps_and_categories(language_prefix, session=db.session()),
             from_attributes=True,
         ).model_dump(mode="json")
 

api/controllers/console/explore/trial.py

@@ -44,7 +44,9 @@ from controllers.console.explore.error import (
     NotWorkflowAppError,
 )
 from controllers.console.explore.wraps import TrialAppResource, trial_feature_enable
-from controllers.console.wraps import with_current_user
+from controllers.console.files import FILE_UPLOAD_PARAMS, upload_file_from_request
+from controllers.console.remote_files import RemoteFileUploadPayload, upload_remote_file_from_request
+from controllers.console.wraps import cloud_edition_billing_resource_check, with_current_user
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -64,6 +66,7 @@ from fields.app_fields import (
     tag_fields,
 )
 from fields.dataset_fields import dataset_fields
+from fields.file_fields import FileResponse, FileWithSignedUrl
 from fields.member_fields import simple_account_fields
 from fields.message_fields import SuggestedQuestionsResponse
 from fields.workflow_fields import (
@@ -75,8 +78,8 @@ from fields.workflow_fields import (
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
-from libs.helper import uuid_value
-from models import Account
+from libs.helper import dump_response, uuid_value
+from models import Account, App
 from models.account import TenantStatus
 from models.model import AppMode, Site
 from models.workflow import Workflow
@@ -199,6 +202,36 @@ register_response_schema_models(
 )
 
 
+class TrialAppFileUploadApi(TrialAppResource):
+    @trial_feature_enable
+    @cloud_edition_billing_resource_check("documents")
+    @console_ns.doc(consumes=["multipart/form-data"], params=FILE_UPLOAD_PARAMS)
+    @console_ns.response(201, "File uploaded successfully", console_ns.models[FileResponse.__name__])
+    @with_current_user
+    def post(self, current_user: Account, app_model: App):
+        """Upload a file into the tenant that owns the trial app."""
+        upload_file = upload_file_from_request(
+            current_user=current_user,
+            resource_tenant_id=app_model.tenant_id,
+        )
+        return dump_response(FileResponse, upload_file), 201
+
+
+class TrialAppRemoteFileUploadApi(TrialAppResource):
+    @trial_feature_enable
+    @cloud_edition_billing_resource_check("documents")
+    @console_ns.expect(console_ns.models[RemoteFileUploadPayload.__name__])
+    @console_ns.response(201, "File uploaded successfully", console_ns.models[FileWithSignedUrl.__name__])
+    @with_current_user
+    def post(self, current_user: Account, app_model: App):
+        """Upload a remote file into the tenant that owns the trial app."""
+        remote_file = upload_remote_file_from_request(
+            current_user=current_user,
+            resource_tenant_id=app_model.tenant_id,
+        )
+        return remote_file.model_dump(mode="json"), 201
+
+
 class TrialAppWorkflowRunApi(TrialAppResource):
     @trial_feature_enable
     @console_ns.expect(console_ns.models[WorkflowRunRequest.__name__])
@@ -605,6 +638,18 @@ class DatasetListApi(Resource):
 
 console_ns.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
 
+console_ns.add_resource(
+    TrialAppFileUploadApi,
+    "/trial-apps/<uuid:app_id>/files/upload",
+    endpoint="trial_app_file_upload",
+)
+
+console_ns.add_resource(
+    TrialAppRemoteFileUploadApi,
+    "/trial-apps/<uuid:app_id>/remote-files/upload",
+    endpoint="trial_app_remote_file_upload",
+)
+
 console_ns.add_resource(
     TrialMessageSuggestedQuestionApi,
     "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",

api/controllers/console/files.py

@@ -28,7 +28,7 @@ from controllers.console.wraps import (
 from extensions.ext_database import db
 from fields.file_fields import FileResponse, UploadConfig
 from libs.login import login_required
-from models import Account
+from models import Account, UploadFile
 from services.file_service import FileService
 
 from . import console_ns
@@ -38,6 +38,59 @@ register_response_schema_models(console_ns, AllowedExtensionsResponse, TextConte
 
 PREVIEW_WORDS_LIMIT = 3000
 
+FILE_UPLOAD_PARAMS = {
+    "file": {
+        "description": "File to upload",
+        "in": "formData",
+        "type": "file",
+        "required": True,
+    },
+    "source": {
+        "description": "Optional upload source",
+        "in": "formData",
+        "type": "string",
+        "enum": ["datasets"],
+        "required": False,
+    },
+}
+
+
+def upload_file_from_request(*, current_user: Account, resource_tenant_id: str | None = None) -> UploadFile:
+    """Validate the multipart request and persist the file under the requested resource tenant."""
+    source_str = request.form.get("source")
+    source: Literal["datasets"] | None = "datasets" if source_str == "datasets" else None
+
+    if "file" not in request.files:
+        raise NoFileUploadedError()
+
+    if len(request.files) > 1:
+        raise TooManyFilesError()
+    file = request.files["file"]
+
+    if not file.filename:
+        raise FilenameNotExistsError
+    if source == "datasets" and not current_user.is_dataset_editor:
+        raise Forbidden()
+
+    if source not in ("datasets", None):
+        source = None
+
+    try:
+        return FileService(db.engine).upload_file(
+            filename=file.filename,
+            content=file.stream.read(),
+            mimetype=file.mimetype,
+            user=current_user,
+            tenant_id=resource_tenant_id,
+            source=source,
+        )
+    except services.errors.file.FileTooLargeError as file_too_large_error:
+        raise FileTooLargeError(file_too_large_error.description)
+    except services.errors.file.UnsupportedFileTypeError:
+        raise UnsupportedFileTypeError()
+    except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
+        raise BlockedFileExtensionError(blocked_extension_error.description)
+
 
 @console_ns.route("/files/upload")
 class FileApi(Resource):
@@ -64,41 +117,11 @@ class FileApi(Resource):
     @login_required
     @account_initialization_required
     @cloud_edition_billing_resource_check("documents")
+    @console_ns.doc(consumes=["multipart/form-data"], params=FILE_UPLOAD_PARAMS)
     @console_ns.response(201, "File uploaded successfully", console_ns.models[FileResponse.__name__])
     @with_current_user
     def post(self, current_user: Account):
-        source_str = request.form.get("source")
-        source: Literal["datasets"] | None = "datasets" if source_str == "datasets" else None
-
-        if "file" not in request.files:
-            raise NoFileUploadedError()
-
-        if len(request.files) > 1:
-            raise TooManyFilesError()
-        file = request.files["file"]
-
-        if not file.filename:
-            raise FilenameNotExistsError
-        if source == "datasets" and not current_user.is_dataset_editor:
-            raise Forbidden()
-
-        if source not in ("datasets", None):
-            source = None
-
-        try:
-            upload_file = FileService(db.engine).upload_file(
-                filename=file.filename,
-                content=file.stream.read(),
-                mimetype=file.mimetype,
-                user=current_user,
-                source=source,
-            )
-        except services.errors.file.FileTooLargeError as file_too_large_error:
-            raise FileTooLargeError(file_too_large_error.description)
-        except services.errors.file.UnsupportedFileTypeError:
-            raise UnsupportedFileTypeError()
-        except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
-            raise BlockedFileExtensionError(blocked_extension_error.description)
+        upload_file = upload_file_from_request(current_user=current_user)
 
         response = FileResponse.model_validate(upload_file, from_attributes=True)
         return response.model_dump(mode="json"), 201

api/controllers/console/remote_files.py

@@ -46,6 +46,61 @@ class GetRemoteFileInfo(Resource):
         ).model_dump(mode="json")
 
 
+def upload_remote_file_from_request(
+    *,
+    current_user: Account,
+    resource_tenant_id: str | None = None,
+) -> FileWithSignedUrl:
+    """Validate the JSON request, fetch its remote file, and persist it under the requested tenant."""
+    payload = RemoteFileUploadPayload.model_validate(console_ns.payload)
+    url = payload.url
+
+    # Try to fetch remote file metadata/content first
+    try:
+        resp = remote_fetcher.make_request("HEAD", url=url)
+        if resp.status_code != httpx.codes.OK:
+            resp = remote_fetcher.make_request("GET", url=url, timeout=3, follow_redirects=True)
+        if resp.status_code != httpx.codes.OK:
+            # Normalize into a user-friendly error message expected by tests
+            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
+    except httpx.RequestError as e:
+        raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
+
+    file_info = helpers.guess_file_info_from_response(resp)
+
+    # Enforce file size limit with 400 (Bad Request) per tests' expectation
+    if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
+        raise FileTooLargeError()
+
+    # Load content if needed
+    content = resp.content if resp.request.method == "GET" else remote_fetcher.make_request("GET", url).content
+
+    try:
+        upload_file = FileService(db.engine).upload_file(
+            filename=file_info.filename,
+            content=content,
+            mimetype=file_info.mimetype,
+            user=current_user,
+            tenant_id=resource_tenant_id,
+            source_url=url,
+        )
+    except services.errors.file.FileTooLargeError as file_too_large_error:
+        raise FileTooLargeError(file_too_large_error.description)
+    except services.errors.file.UnsupportedFileTypeError:
+        raise UnsupportedFileTypeError()
+
+    return FileWithSignedUrl(
+        id=upload_file.id,
+        name=upload_file.name,
+        size=upload_file.size,
+        extension=upload_file.extension,
+        url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
+        mime_type=upload_file.mime_type,
+        created_by=upload_file.created_by,
+        created_at=int(upload_file.created_at.timestamp()),
+    )
+
+
 @console_ns.route("/remote-files/upload")
 class RemoteFileUpload(Resource):
     @console_ns.expect(console_ns.models[RemoteFileUploadPayload.__name__])
@@ -53,53 +108,8 @@ class RemoteFileUpload(Resource):
     @login_required
     @with_current_user
     def post(self, current_user: Account):
-        payload = RemoteFileUploadPayload.model_validate(console_ns.payload)
-        url = payload.url
-
-        # Try to fetch remote file metadata/content first
-        try:
-            resp = remote_fetcher.make_request("HEAD", url=url)
-            if resp.status_code != httpx.codes.OK:
-                resp = remote_fetcher.make_request("GET", url=url, timeout=3, follow_redirects=True)
-            if resp.status_code != httpx.codes.OK:
-                # Normalize into a user-friendly error message expected by tests
-                raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
-        except httpx.RequestError as e:
-            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
-
-        file_info = helpers.guess_file_info_from_response(resp)
-
-        # Enforce file size limit with 400 (Bad Request) per tests' expectation
-        if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
-            raise FileTooLargeError()
-
-        # Load content if needed
-        content = resp.content if resp.request.method == "GET" else remote_fetcher.make_request("GET", url).content
-
-        try:
-            upload_file = FileService(db.engine).upload_file(
-                filename=file_info.filename,
-                content=content,
-                mimetype=file_info.mimetype,
-                user=current_user,
-                source_url=url,
-            )
-        except services.errors.file.FileTooLargeError as file_too_large_error:
-            raise FileTooLargeError(file_too_large_error.description)
-        except services.errors.file.UnsupportedFileTypeError:
-            raise UnsupportedFileTypeError()
-
-        # Success: return created resource with 201 status
+        remote_file = upload_remote_file_from_request(current_user=current_user)
         return (
-            FileWithSignedUrl(
-                id=upload_file.id,
-                name=upload_file.name,
-                size=upload_file.size,
-                extension=upload_file.extension,
-                url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
-                mime_type=upload_file.mime_type,
-                created_by=upload_file.created_by,
-                created_at=int(upload_file.created_at.timestamp()),
-            ).model_dump(mode="json"),
+            remote_file.model_dump(mode="json"),
             201,
         )

api/controllers/console/workspace/members.py

@@ -36,7 +36,7 @@ from libs.login import current_account_with_tenant, login_required
 from models.account import Account, TenantAccountJoin, TenantAccountRole
 from services.account_service import AccountService, RegisterService, TenantService
 from services.enterprise import rbac_service as enterprise_rbac_service
-from services.errors.account import AccountAlreadyInTenantError
+from services.errors.account import AccountAlreadyInTenantError, SeatsLimitExceededError
 from services.feature_service import FeatureService
 
 
@@ -291,6 +291,14 @@ class MemberInviteEmailApi(Resource):
                             "message": "Account already in workspace.",
                         }
                     )
+                except SeatsLimitExceededError:
+                    invitation_results.append(
+                        {
+                            "status": "failed",
+                            "email": invitee_email,
+                            "message": "Licensed seats limit exceeded.",
+                        }
+                    )
                 except Exception as e:
                     invitation_results.append({"status": "failed", "email": invitee_email, "message": str(e)})
 

api/controllers/console/workspace/model_providers.py

@@ -18,7 +18,6 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user,
 )
-from extensions.ext_database import db
 from fields.base import ResponseModel
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
@@ -353,7 +352,7 @@ class ModelProviderPaymentCheckoutUrlApi(Resource):
     def get(self, current_tenant_id: str, current_user: Account, provider: str):
         if provider != "anthropic":
             raise ValueError(f"provider name {provider} is invalid")
-        BillingService.is_tenant_owner_or_admin(db.session, current_user)
+        BillingService.is_tenant_owner_or_admin(current_user)
         data = BillingService.get_model_provider_payment_link(
             provider_name=provider,
             tenant_id=current_tenant_id,

api/controllers/console/workspace/rbac.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-from enum import StrEnum
 from typing import Any
 
 from flask import request
@@ -14,9 +13,11 @@ from controllers.common.schema import register_response_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import RBACPermission, RBACResourceScope, rbac_permission_required
 from core.db.session_factory import session_factory
+from core.rbac import RBACResourceWhitelistScope
 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]):
@@ -511,14 +512,8 @@ class RBACAccessPolicyBindingUnlockApi(Resource):
 # ---------------------------------------------------------------------------
 
 
-class _AccessScope(StrEnum):
-    ALL = "all"
-    SPECIFIC = "specific"
-    ONLY_ME = "only_me"
-
-
 class _ResourceAccessScopeRequest(BaseModel):
-    scope: _AccessScope
+    scope: RBACResourceWhitelistScope
 
 
 class _ReplaceBindingsRequest(BaseModel):
@@ -584,14 +579,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")
@@ -616,8 +612,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,
             )
         )

api/controllers/files/__init__.py

@@ -14,12 +14,11 @@ api = ExternalApi(
 
 files_ns = Namespace("files", description="File operations", path="/")
 
-from . import agent_drive_archive, image_preview, tool_files, upload
+from . import image_preview, tool_files, upload
 
 api.add_namespace(files_ns)
 
 __all__ = [
-    "agent_drive_archive",
     "api",
     "bp",
     "files_ns",

api/controllers/files/agent_drive_archive.py

@@ -1,67 +0,0 @@
-from urllib.parse import quote
-
-from flask import Response, request
-from flask_restx import Resource
-from pydantic import BaseModel, Field
-from werkzeug.exceptions import Forbidden, NotFound
-
-from controllers.common.file_response import enforce_download_for_html
-from controllers.common.schema import register_schema_models
-from controllers.files import files_ns
-from models.agent import AgentDriveFileKind
-from services.agent_drive_service import AgentDriveError, AgentDriveService
-
-
-class AgentDriveArchiveMemberQuery(BaseModel):
-    tenant_id: str = Field(..., description="Tenant ID")
-    agent_id: str = Field(..., description="Agent ID")
-    key: str = Field(..., description="Virtual drive key")
-    archive_file_kind: AgentDriveFileKind = Field(..., description="Archive file kind")
-    archive_file_id: str = Field(..., description="Archive file id")
-    member_path: str = Field(..., description="Zip member path")
-    timestamp: str = Field(..., description="Unix timestamp")
-    nonce: str = Field(..., description="Random nonce")
-    sign: str = Field(..., description="HMAC signature")
-    as_attachment: bool = Field(default=False, description="Download as attachment")
-
-
-register_schema_models(files_ns, AgentDriveArchiveMemberQuery)
-
-
-@files_ns.route("/agent-drive/archive-member")
-class AgentDriveArchiveMemberApi(Resource):
-    @files_ns.doc("get_agent_drive_archive_member")
-    @files_ns.doc(description="Download a lazily resolved Agent Skill archive member by signed parameters")
-    def get(self):
-        args = AgentDriveArchiveMemberQuery.model_validate(request.args.to_dict(flat=True))
-        if not AgentDriveService.verify_archive_member_signature(
-            tenant_id=args.tenant_id,
-            agent_id=args.agent_id,
-            key=args.key,
-            archive_file_kind=args.archive_file_kind,
-            archive_file_id=args.archive_file_id,
-            member_path=args.member_path,
-            timestamp=args.timestamp,
-            nonce=args.nonce,
-            sign=args.sign,
-        ):
-            raise Forbidden("Invalid request.")
-        try:
-            payload, mime_type, filename = AgentDriveService().load_archive_member_for_signed_request(
-                tenant_id=args.tenant_id,
-                agent_id=args.agent_id,
-                key=args.key,
-                archive_file_kind=args.archive_file_kind,
-                archive_file_id=args.archive_file_id,
-                member_path=args.member_path,
-            )
-        except AgentDriveError as exc:
-            raise NotFound(exc.message) from exc
-
-        response = Response(payload, mimetype=mime_type, direct_passthrough=True, headers={})
-        response.headers["Content-Length"] = str(len(payload))
-        if args.as_attachment and filename:
-            encoded_filename = quote(filename)
-            response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
-        enforce_download_for_html(response, mime_type=mime_type, filename=filename, extension="")
-        return response

api/controllers/openapi/__init__.py

@@ -20,7 +20,7 @@ openapi_ns = Namespace("openapi", description="User-scoped operations", path="/"
 
 # Register response/query models BEFORE importing controller modules so that
 # @openapi_ns.response / @openapi_ns.expect decorators can resolve model names.
-from controllers.common.fields import EventStreamResponse, SimpleResultResponse
+from controllers.common.fields import EventStreamResponse
 from controllers.common.schema import register_enum_models, register_response_schema_models, register_schema_models
 from controllers.openapi._models import (
     AccountPayload,
@@ -95,7 +95,6 @@ register_response_schema_models(
     openapi_ns,
     ErrorBody,
     EventStreamResponse,
-    SimpleResultResponse,
     UsageInfo,
     MessageMetadata,
     AppListRow,

api/controllers/openapi/app_run.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Callable, Generator
+from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from typing import Any
 
@@ -61,7 +61,7 @@ logger = logging.getLogger(__name__)
 
 
 @contextmanager
-def _translate_service_errors() -> Generator[None, None, None]:
+def _translate_service_errors() -> Iterator[None]:
     try:
         yield
     except WorkflowNotFoundError as ex:
@@ -166,7 +166,6 @@ class AppRunApi(Resource):
             surface="apps",
         )
 
-        # response-contract:ignore compact_generate_response
         return helper.compact_generate_response(stream_obj)
 
 

api/controllers/openapi/apps.py

@@ -25,13 +25,12 @@ from controllers.openapi._models import (
     AppListRow,
 )
 from controllers.openapi.auth.composition import auth_router
-from controllers.openapi.auth.data import AuthData, CallerKind, RBACRequirement
+from controllers.openapi.auth.data import AuthData, RBACRequirement
 from controllers.service_api.app.error import AppUnavailableError
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from extensions.ext_database import db
 from libs.oauth_bearer import Scope, TokenType
 from models import App
-from models.enums import AppStatus
 from models.model import AppMode
 from services.account_service import TenantService
 from services.app_service import AppListParams, AppService
@@ -167,9 +166,7 @@ class AppListApi(Resource):
         # an empty set or list means the caller has no accessible apps.
         # End-users bypass RBAC here — their access is controlled by scope upstream.
         apply_rbac_filter = (
-            dify_config.RBAC_ENABLED
-            and auth_data.caller_kind != CallerKind.END_USER
-            and auth_data.account_id is not None
+            dify_config.RBAC_ENABLED and auth_data.caller_kind != "end_user" and auth_data.account_id is not None
         )
         access_filter = AppAccessFilter.unrestricted()
         if apply_rbac_filter:
@@ -206,7 +203,7 @@ class AppListApi(Resource):
             limit=query.limit,
             mode=query.mode.value if query.mode else "all",  # type:ignore
             name=query.name,
-            status=AppStatus.NORMAL,
+            status="normal",
             # Visibility gate pushed into the query — pagination.total stays
             # consistent across pages because invisible rows never count.
             openapi_visible=True,

api/controllers/openapi/apps_permitted_external.py

@@ -25,7 +25,6 @@ from controllers.openapi.auth.data import AuthData, Edition
 from extensions.ext_database import db
 from libs.oauth_bearer import Scope, TokenType
 from models import App
-from models.enums import AppStatus
 from services.account_service import TenantService
 from services.app_service import AppService
 from services.enterprise.app_permitted_service import list_permitted_apps
@@ -63,7 +62,7 @@ class PermittedExternalAppsListApi(Resource):
         items: list[AppListRow] = []
         for app_id in page_result.app_ids:
             app = apps_by_id.get(app_id)
-            if not app or app.status != AppStatus.NORMAL:
+            if not app or app.status != "normal":
                 continue
             tenant = tenants_by_id.get(str(app.tenant_id))
             items.append(

api/controllers/openapi/auth/data.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import uuid
 from enum import StrEnum
+from typing import Literal
 
 from pydantic import BaseModel, ConfigDict, Field
 from werkzeug.exceptions import InternalServerError
@@ -20,11 +21,6 @@ class Edition(StrEnum):
     SAAS = "saas"
 
 
-class CallerKind(StrEnum):
-    ACCOUNT = "account"
-    END_USER = "end_user"
-
-
 def current_edition() -> Edition:
     if dify_config.EDITION == "CLOUD":
         return Edition.SAAS
@@ -82,9 +78,9 @@ class AuthData(BaseModel):
     tenant_role: TenantAccountRole | None = None
 
     caller: Account | EndUser | None = None
-    caller_kind: CallerKind | None = None
+    caller_kind: Literal["account", "end_user"] | None = None
 
-    def require_app_context(self) -> tuple[App, Account | EndUser, CallerKind]:
+    def require_app_context(self) -> tuple[App, Account | EndUser, Literal["account", "end_user"]]:
         if self.app is None or self.caller is None or self.caller_kind is None:
             raise InternalServerError("pipeline_invariant_violated: app context missing")
         return self.app, self.caller, self.caller_kind

api/controllers/openapi/auth/prepare.py

@@ -5,10 +5,10 @@ import uuid
 from flask import request
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound, Unauthorized
 
-from controllers.openapi.auth.data import AuthData, CallerKind
+from controllers.openapi.auth.data import AuthData
 from extensions.ext_database import db
-from models.account import AccountStatus, TenantStatus
-from models.enums import AppStatus, EndUserType
+from models.account import TenantStatus
+from models.enums import EndUserType
 from services.account_service import AccountService, TenantService
 from services.app_service import AppService
 from services.end_user_service import EndUserService
@@ -24,7 +24,7 @@ def load_app(data: AuthData) -> None:
     except ValueError:
         raise NotFound("app not found")
     app = AppService.get_app_by_id(db.session, app_id)
-    if not app or app.status != AppStatus.NORMAL:
+    if not app or app.status != "normal":
         raise NotFound("app not found")
     data.app = app
 
@@ -65,7 +65,7 @@ def load_account(data: AuthData) -> None:
     if data.tenant:
         account.current_tenant = data.tenant
     data.caller = account
-    data.caller_kind = CallerKind.ACCOUNT
+    data.caller_kind = "account"
 
 
 def load_workspace_role(data: AuthData) -> None:
@@ -73,7 +73,7 @@ def load_workspace_role(data: AuthData) -> None:
         return
     if data.tenant is None or data.account_id is None:
         return
-    if data.caller is not None and getattr(data.caller, "status", None) != AccountStatus.ACTIVE:
+    if data.caller is not None and getattr(data.caller, "status", None) != "active":
         return
     role = TenantService.get_account_role_in_tenant(db.session, str(data.account_id), str(data.tenant.id))
     if role is None:
@@ -91,7 +91,7 @@ def resolve_external_user(data: AuthData) -> None:
         user_id=data.external_identity.email,
     )
     data.caller = end_user
-    data.caller_kind = CallerKind.END_USER
+    data.caller_kind = "end_user"
 
 
 def load_app_access_mode(data: AuthData) -> None:

api/controllers/openapi/auth/verify.py

@@ -5,7 +5,7 @@ from werkzeug.exceptions import Forbidden, NotFound, UnprocessableEntity
 
 from configs import dify_config
 from controllers.common.wraps import enforce_rbac_access
-from controllers.openapi.auth.data import AuthData, CallerKind
+from controllers.openapi.auth.data import AuthData
 from extensions.ext_database import db
 from libs.oauth_bearer import Scope, TokenType
 from services.account_service import AccountService, TenantService
@@ -58,7 +58,7 @@ def check_rbac_permission(data: AuthData) -> None:
     if not dify_config.RBAC_ENABLED:
         return
     # Only account callers are subject to RBAC; end_user access is scope-controlled.
-    if data.caller_kind != CallerKind.ACCOUNT:
+    if data.caller_kind != "account":
         return
     if data.account_id is None or data.tenant is None:
         raise Forbidden("rbac context missing")

api/controllers/openapi/human_input_form.py

@@ -22,7 +22,7 @@ from controllers.openapi._contract import accepts, returns
 from controllers.openapi._errors import HumanInputFormNotFound, RecipientSurfaceMismatch
 from controllers.openapi._models import FormSubmitResponse, HumanInputFormDefinitionResponse
 from controllers.openapi.auth.composition import auth_router
-from controllers.openapi.auth.data import AuthData, CallerKind, RBACRequirement
+from controllers.openapi.auth.data import AuthData, RBACRequirement
 from core.workflow.human_input_policy import (
     HumanInputSurface,
     is_recipient_type_allowed_for_surface,
@@ -98,7 +98,7 @@ class OpenApiWorkflowHumanInputFormApi(Resource):
 
         submission_user_id: str | None = None
         submission_end_user_id: str | None = None
-        if caller_kind == CallerKind.ACCOUNT:
+        if caller_kind == "account":
             submission_user_id = caller.id
         else:
             submission_end_user_id = caller.id

api/controllers/openapi/workflow_events.py

@@ -22,7 +22,7 @@ from controllers.common.schema import query_params_from_model
 from controllers.common.wraps import RBACPermission, RBACResourceScope
 from controllers.openapi import openapi_ns
 from controllers.openapi.auth.composition import auth_router
-from controllers.openapi.auth.data import AuthData, CallerKind, RBACRequirement
+from controllers.openapi.auth.data import AuthData, RBACRequirement
 from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
 from core.app.apps.base_app_generator import BaseAppGenerator
 from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
@@ -70,7 +70,7 @@ class OpenApiWorkflowEventsApi(Resource):
         if workflow_run.app_id != app_model.id:
             raise NotFound("Workflow run not found")
 
-        if caller_kind == CallerKind.ACCOUNT:
+        if caller_kind == "account":
             if workflow_run.created_by_role != CreatorUserRole.ACCOUNT or workflow_run.created_by != caller.id:
                 raise NotFound("Workflow run not found")
         else:

api/controllers/openapi/workspaces.py

@@ -48,6 +48,7 @@ from services.errors.account import (
     MemberNotInTenantError,
     NoPermissionError,
     RoleAlreadyAssignedError,
+    SeatsLimitExceededError,
 )
 from services.feature_service import FeatureService
 
@@ -190,6 +191,8 @@ class WorkspaceMembersApi(Resource):
             raise BadRequest(str(exc))
         except NoPermissionError as exc:
             raise BadRequest(str(exc))
+        except SeatsLimitExceededError:
+            raise BadRequest("licensed seats limit exceeded")
         except AccountRegisterError as exc:
             raise BadRequest(str(exc))
 

api/controllers/service_api/dataset/dataset.py

@@ -43,6 +43,7 @@ from models.dataset import DatasetPermissionEnum
 from models.enums import TagType
 from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
+from services.enterprise.rbac_service import RBACResourceWhitelistScope, RBACService, ReplaceMemberBindings
 from services.entities.knowledge_entities.knowledge_entities import (
     ExternalRetrievalModel,
     KnowledgeProvider,
@@ -58,6 +59,7 @@ from services.tag_service import (
 from services.tag_service import (
     UpdateTagPayload as UpdateTagServicePayload,
 )
+from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
 
 register_enum_models(service_api_ns, DatasetPermissionEnum)
 
@@ -523,6 +525,15 @@ class DatasetListApi(DatasetApiResource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
+        if payload.permission == DatasetPermissionEnum.ALL_TEAM and dify_config.RBAC_ENABLED:
+            RBACService.DatasetAccess.replace_whitelist(
+                tenant_id,
+                current_user.id,
+                dataset.id,
+                ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
+            )
+            initialize_created_app_rbac_access_task.delay(tenant_id, current_user.id, dataset_id=dataset.id)
+
         return _dump_service_dataset_detail(dataset), 200
 
 
@@ -565,7 +576,7 @@ class DatasetApi(DatasetApiResource):
         if dataset is None:
             raise NotFound("Dataset not found.")
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         data = _dump_service_dataset_detail(dataset)
@@ -819,7 +830,7 @@ class DocumentStatusApi(DatasetApiResource):
 
         # Check user's permission
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 

api/controllers/service_api/dataset/metadata.py

@@ -84,7 +84,7 @@ class DatasetMetadataCreateServiceApi(DatasetApiResource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata = MetadataService.create_metadata(db.session(), dataset_id_str, metadata_args)
         return dump_response(DatasetMetadataResponse, metadata), 201
@@ -157,7 +157,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata = MetadataService.update_metadata_name(db.session(), dataset_id_str, metadata_id_str, payload.name)
         return dump_response(DatasetMetadataResponse, metadata), 200
@@ -192,7 +192,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         MetadataService.delete_metadata(db.session(), dataset_id_str, metadata_id_str)
         return "", 204
@@ -260,7 +260,7 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         match action:
             case "enable":
@@ -306,7 +306,7 @@ class DocumentMetadataEditServiceApi(DatasetApiResource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata_args = MetadataOperationData.model_validate(service_api_ns.payload or {})
 

api/controllers/service_api/wraps.py

@@ -12,6 +12,7 @@ from flask_restx import Resource
 from flask_restx.utils import merge
 from pydantic import BaseModel
 from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
 from configs import dify_config
@@ -269,8 +270,8 @@ def cloud_edition_billing_rate_limit_check[**P, R](
                             subscription_plan=knowledge_rate_limit.subscription_plan,
                             operation="knowledge",
                         )
-                        db.session.add(rate_limit_log)
-                        db.session.commit()
+                        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
+                            session.add(rate_limit_log)
                         raise Forbidden(
                             "Sorry, you have reached the knowledge base request rate limit of your subscription."
                         )

api/controllers/trigger/webhook.py

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

api/controllers/web/login.py

@@ -29,6 +29,7 @@ from controllers.console.wraps import (
 )
 from controllers.web import web_ns
 from controllers.web.wraps import decode_jwt_token
+from core.logging.context import set_identity_context
 from libs.helper import EmailStr, extract_remote_ip
 from libs.passport import PassportService
 from libs.password import valid_password
@@ -160,7 +161,12 @@ class LoginStatusApi(Resource):
                 user_logged_in = False
 
         try:
-            _ = decode_jwt_token(app_code=app_code, user_id=user_id)
+            _, end_user = decode_jwt_token(app_code=app_code, user_id=user_id)
+            set_identity_context(
+                tenant_id=end_user.tenant_id,
+                user_id=end_user.id,
+                user_type=end_user.type or "end_user",
+            )
             app_logged_in = True
         except Exception:
             app_logged_in = False

api/controllers/web/site.py

@@ -14,7 +14,7 @@ from fields.base import ResponseModel
 from libs.helper import AppIconUrlField
 from models.account import TenantStatus
 from models.model import App, EndUser, Site
-from services.feature_service import FeatureModel, FeatureService
+from services.feature_service import FeatureService
 
 
 class AppSiteModelConfigResponse(ResponseModel):
@@ -38,7 +38,6 @@ class AppSiteResponse(ResponseModel):
     description: str | None = None
     copyright: str | None = None
     privacy_policy: str | None = None
-    input_placeholder: str | None = None
     custom_disclaimer: str | None = None
     default_language: str | None = None
     prompt_public: bool | None = None
@@ -85,7 +84,6 @@ class AppSiteApi(WebApiResource):
         "description": fields.String,
         "copyright": fields.String,
         "privacy_policy": fields.String,
-        "input_placeholder": fields.String,
         "custom_disclaimer": fields.String,
         "default_language": fields.String,
         "prompt_public": fields.Boolean,
@@ -129,15 +127,9 @@ class AppSiteApi(WebApiResource):
         if app_model.tenant and app_model.tenant.status == TenantStatus.ARCHIVE:
             raise Forbidden()
 
-        features = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True)
+        can_replace_logo = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True).can_replace_logo
 
-        return AppSiteInfo(
-            app_model.tenant,
-            app_model,
-            serialize_runtime_site(site, features),
-            end_user.id,
-            features.can_replace_logo,
-        )
+        return AppSiteInfo(app_model.tenant, app_model, site, end_user.id, can_replace_logo)
 
 
 class AppSiteInfo:
@@ -172,23 +164,7 @@ def serialize_site(site: Site) -> dict[str, Any]:
     return cast(dict[str, Any], marshal(site, AppSiteApi.site_fields))
 
 
-def serialize_runtime_site(site: Site, features: FeatureModel) -> dict[str, Any]:
-    site_payload = serialize_site(site)
-    if not features.billing.enabled or features.webapp_copyright_enabled:
-        return site_payload
-
-    site_payload["copyright"] = None
-    site_payload["input_placeholder"] = None
-    return site_payload
-
-
 def serialize_app_site_payload(app_model: App, site: Site, end_user_id: str | None) -> dict[str, Any]:
-    features = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True)
-    app_site_info = AppSiteInfo(
-        app_model.tenant,
-        app_model,
-        serialize_runtime_site(site, features),
-        end_user_id,
-        features.can_replace_logo,
-    )
+    can_replace_logo = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True).can_replace_logo
+    app_site_info = AppSiteInfo(app_model.tenant, app_model, site, end_user_id, can_replace_logo)
     return cast(dict[str, Any], marshal(app_site_info, AppSiteApi.app_fields))

api/controllers/web/wraps.py

@@ -11,6 +11,7 @@ from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
 
 from constants import HEADER_NAME_APP_CODE
 from controllers.web.error import WebAppAuthAccessDeniedError, WebAppAuthRequiredError
+from core.logging.context import set_identity_context
 from extensions.ext_database import db
 from libs.passport import PassportService
 from libs.token import extract_webapp_passport
@@ -28,6 +29,11 @@ def validate_jwt_token[**P, R](
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
             app_model, end_user = decode_jwt_token()
+            set_identity_context(
+                tenant_id=end_user.tenant_id,
+                user_id=end_user.id,
+                user_type=end_user.type or "end_user",
+            )
             return view(app_model, end_user, *args, **kwargs)
 
         return decorated

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

@@ -47,6 +47,7 @@ from core.repositories import DifyCoreRepositoryFactory
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
+from graphon.filters import ResponseStreamFilter
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from graphon.runtime import GraphRuntimeState
@@ -231,6 +232,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                 workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
             workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
                 session_factory=session_factory,
+                tenant_id=app_model.tenant_id,
                 user=user,
                 app_id=application_generate_entity.app_config.app_id,
                 triggered_from=workflow_triggered_from,
@@ -238,6 +240,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             # Create workflow node execution repository
             workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
                 session_factory=session_factory,
+                tenant_id=app_model.tenant_id,
                 user=user,
                 app_id=application_generate_entity.app_config.app_id,
                 triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
@@ -268,6 +271,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         graph_runtime_state: GraphRuntimeState,
         pause_state_config: PauseStateLayerConfig | None = None,
+        response_stream_filter: ResponseStreamFilter | None = None,
     ):
         """
         Resume a paused advanced chat execution.
@@ -297,6 +301,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             stream=application_generate_entity.stream,
             pause_state_config=pause_state_config,
             graph_runtime_state=graph_runtime_state,
+            response_stream_filter=response_stream_filter,
         )
 
     def single_iteration_generate(
@@ -356,6 +361,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         # Create workflow execution(aka workflow run) repository
         workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
             session_factory=session_factory,
+            tenant_id=app_model.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
@@ -363,6 +369,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         # Create workflow node execution repository
         workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
             session_factory=session_factory,
+            tenant_id=app_model.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
@@ -443,6 +450,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         # Create workflow execution(aka workflow run) repository
         workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
             session_factory=session_factory,
+            tenant_id=app_model.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
@@ -450,6 +458,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         # Create workflow node execution repository
         workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
             session_factory=session_factory,
+            tenant_id=app_model.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
@@ -491,6 +500,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         pause_state_config: PauseStateLayerConfig | None = None,
         graph_runtime_state: GraphRuntimeState | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        response_stream_filter: ResponseStreamFilter | None = None,
     ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
         """
         Generate App response.
@@ -538,12 +548,14 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             )
 
             graph_layers: list[GraphEngineLayer] = list(graph_engine_layers)
+            resolved_response_stream_filter = response_stream_filter or ResponseStreamFilter()
             if pause_state_config is not None:
                 graph_layers.append(
                     PauseStatePersistenceLayer(
                         session_factory=pause_state_config.session_factory,
                         generate_entity=application_generate_entity,
                         state_owner_user_id=pause_state_config.state_owner_user_id,
+                        response_stream_filter=resolved_response_stream_filter,
                     )
                 )
 
@@ -564,6 +576,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                     "workflow_node_execution_repository": workflow_node_execution_repository,
                     "graph_engine_layers": tuple(graph_layers),
                     "graph_runtime_state": graph_runtime_state,
+                    "response_stream_filter": resolved_response_stream_filter,
                 },
             )
 
@@ -585,7 +598,11 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                 message=message_snapshot,
                 user=user,
                 stream=stream,
-                draft_var_saver_factory=self._get_draft_var_saver_factory(invoke_from, account=user),
+                draft_var_saver_factory=self._get_draft_var_saver_factory(
+                    invoke_from,
+                    account=user,
+                    tenant_id=application_generate_entity.app_config.tenant_id,
+                ),
             )
 
             return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
@@ -603,6 +620,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         graph_runtime_state: GraphRuntimeState | None = None,
+        response_stream_filter: ResponseStreamFilter | None = None,
     ):
         """
         Generate worker in a new thread.
@@ -662,6 +680,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                 workflow_node_execution_repository=workflow_node_execution_repository,
                 graph_engine_layers=graph_engine_layers,
                 graph_runtime_state=graph_runtime_state,
+                response_stream_filter=response_stream_filter,
             )
 
             try:

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

@@ -39,6 +39,7 @@ from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from extensions.otel import WorkflowAppRunnerHandler, trace_span
 from graphon.enums import WorkflowType
+from graphon.filters import ResponseStreamFilter
 from graphon.graph_engine.command_channels import RedisChannel
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.runtime import GraphRuntimeState, VariablePool
@@ -73,6 +74,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         graph_runtime_state: GraphRuntimeState | None = None,
+        response_stream_filter: ResponseStreamFilter | None = None,
     ):
         super().__init__(
             queue_manager=queue_manager,
@@ -90,6 +92,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         self._workflow_execution_repository = workflow_execution_repository
         self._workflow_node_execution_repository = workflow_node_execution_repository
         self._resume_graph_runtime_state = graph_runtime_state
+        self._response_stream_filter = response_stream_filter
 
     @trace_span(WorkflowAppRunnerHandler)
     def run(self):
@@ -227,6 +230,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             variable_pool=variable_pool,
             graph_runtime_state=graph_runtime_state,
             command_channel=command_channel,
+            response_stream_filter=self._response_stream_filter,
         )
 
         self._queue_manager.graph_runtime_state = graph_runtime_state

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

@@ -16,7 +16,7 @@ from collections.abc import Generator, Mapping
 from typing import Any
 
 from flask import Flask, current_app
-from sqlalchemy import and_, or_, select
+from sqlalchemy import select
 
 from clients.agent_backend import AgentBackendRunEventAdapter
 from clients.agent_backend.factory import create_agent_backend_run_client
@@ -42,15 +42,7 @@ from core.app.llm.model_access import build_dify_model_access
 from core.ops.ops_trace_manager import TraceQueueManager
 from extensions.ext_database import db
 from models import Account, App, EndUser, Message
-from models.agent import (
-    Agent,
-    AgentConfigDraft,
-    AgentConfigDraftType,
-    AgentConfigSnapshot,
-    AgentScope,
-    AgentSource,
-    AgentStatus,
-)
+from models.agent import Agent, AgentConfigSnapshot, AgentScope, AgentSource, AgentStatus
 from models.agent_config_entities import AgentSoulConfig
 from services.conversation_service import ConversationService
 
@@ -81,15 +73,10 @@ class AgentAppGenerator(MessageBasedAppGenerator):
         inputs = args["inputs"]
 
         # Resolve the bound roster Agent + its current Agent Soul snapshot.
-        agent, agent_config_id, agent_soul = self._resolve_agent(
-            app_model,
-            invoke_from=invoke_from,
-            draft_type=args.get("draft_type"),
-            user=user,
-        )
+        agent, snapshot, agent_soul = self._resolve_agent(app_model)
         runtime_session_snapshot_id = self._runtime_session_snapshot_id(
             invoke_from=invoke_from,
-            snapshot_id=agent_config_id,
+            snapshot_id=snapshot.id,
         )
 
         conversation = None
@@ -136,7 +123,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
             call_depth=0,
             trace_manager=trace_manager,
             agent_id=agent.id,
-            agent_config_snapshot_id=agent_config_id,
+            agent_config_snapshot_id=snapshot.id,
             agent_runtime_session_snapshot_id=runtime_session_snapshot_id,
         )
 
@@ -192,12 +179,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
         persisted to the conversation. Live streaming to a reconnected client is
         out of scope here — the message is persisted and can be re-fetched.
         """
-        agent, agent_config_id, agent_soul = self._resolve_agent(
-            app_model,
-            invoke_from=invoke_from,
-            draft_type="draft",
-            user=user,
-        )
+        agent, snapshot, agent_soul = self._resolve_agent(app_model)
         conversation = ConversationService.get_conversation(
             app_model=app_model, conversation_id=conversation_id, user=user
         )
@@ -244,7 +226,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
             call_depth=0,
             trace_manager=trace_manager,
             agent_id=agent.id,
-            agent_config_snapshot_id=agent_config_id,
+            agent_config_snapshot_id=snapshot.id,
         )
 
         conversation, message = self._init_generate_records(application_generate_entity, conversation)
@@ -439,135 +421,50 @@ class AgentAppGenerator(MessageBasedAppGenerator):
 
         return False, query
 
-    def _resolve_agent(
-        self,
-        app_model: App,
-        *,
-        invoke_from: InvokeFrom,
-        draft_type: Any,
-        user: Account | EndUser,
-    ) -> tuple[Agent, str, AgentSoulConfig]:
+    def _resolve_agent(self, app_model: App) -> tuple[Agent, AgentConfigSnapshot, AgentSoulConfig]:
         agent = db.session.scalar(
-            select(Agent)
-            .where(
-                Agent.tenant_id == app_model.tenant_id,
+            select(Agent).where(
+                Agent.app_id == app_model.id,
+                Agent.scope == AgentScope.ROSTER,
+                Agent.source == AgentSource.AGENT_APP,
                 Agent.status == AgentStatus.ACTIVE,
-                or_(
-                    and_(
-                        Agent.app_id == app_model.id,
-                        Agent.scope == AgentScope.ROSTER,
-                        Agent.source == AgentSource.AGENT_APP,
-                    ),
-                    Agent.backing_app_id == app_model.id,
-                ),
             )
-            .order_by(Agent.created_at.desc())
-            .limit(1)
         )
         if agent is None:
             raise AgentAppGeneratorError("Agent App has no bound Agent")
-        if invoke_from == InvokeFrom.DEBUGGER:
-            draft = self._resolve_debug_draft(
-                tenant_id=app_model.tenant_id,
-                agent=agent,
-                draft_type=draft_type,
-                account_id=user.id if isinstance(user, Account) else None,
-            )
-            agent_soul = AgentSoulConfig.model_validate(draft.config_snapshot_dict)
-            return agent, draft.id, agent_soul
-        _, snapshot, agent_soul = self._resolve_agent_by_id(
-            tenant_id=app_model.tenant_id,
-            agent_id=agent.id,
-            snapshot_id=agent.active_config_snapshot_id,
+        return self._resolve_agent_by_id(
+            tenant_id=app_model.tenant_id, agent_id=agent.id, snapshot_id=agent.active_config_snapshot_id
         )
-        return agent, snapshot.id, agent_soul
 
     @staticmethod
     def _runtime_session_snapshot_id(*, invoke_from: InvokeFrom, snapshot_id: str) -> str | None:
         """Return the session scope snapshot id for Agent App runtime state.
 
-        Console preview/debug chat uses a stable Agent draft row id; build mode
-        uses the current user's build-draft row id. Published/web/API runs use
-        immutable published snapshot ids. This keeps runtime session continuity
-        inside one editable surface without mixing draft/build/published state.
+        Console preview/debug chat is an editing workspace: saving Agent Soul
+        creates replacement snapshots, but the user expects the same preview
+        conversation to keep context while trying prompt changes. Use a stable
+        NULL snapshot scope for debugger runs so each turn can use the latest
+        Agent Soul while reusing the conversation history. Published/web/API
+        runs keep snapshot-scoped sessions for reproducible runtime state.
         """
+        if invoke_from == InvokeFrom.DEBUGGER:
+            return None
         return snapshot_id
 
-    @staticmethod
-    def _resolve_debug_draft(
-        *, tenant_id: str, agent: Agent, draft_type: Any, account_id: str | None
-    ) -> AgentConfigDraft:
-        effective_draft_type = (
-            AgentConfigDraftType.DEBUG_BUILD
-            if draft_type == AgentConfigDraftType.DEBUG_BUILD.value
-            else AgentConfigDraftType.DRAFT
-        )
-        stmt = select(AgentConfigDraft).where(
-            AgentConfigDraft.tenant_id == tenant_id,
-            AgentConfigDraft.agent_id == agent.id,
-            AgentConfigDraft.draft_type == effective_draft_type,
-        )
-        if effective_draft_type == AgentConfigDraftType.DEBUG_BUILD:
-            if not account_id:
-                raise AgentAppGeneratorError("Build draft requires an account user")
-            stmt = stmt.where(AgentConfigDraft.account_id == account_id)
-        else:
-            stmt = stmt.where(AgentConfigDraft.account_id.is_(None))
-        draft = db.session.scalar(stmt.order_by(AgentConfigDraft.updated_at.desc()).limit(1))
-        if draft is not None:
-            return draft
-        if effective_draft_type == AgentConfigDraftType.DEBUG_BUILD:
-            raise AgentAppGeneratorError("Agent build draft not found")
-        _, snapshot, agent_soul = AgentAppGenerator._resolve_agent_by_id(
-            tenant_id=tenant_id,
-            agent_id=agent.id,
-            snapshot_id=agent.active_config_snapshot_id,
-        )
-        draft = AgentConfigDraft(
-            tenant_id=tenant_id,
-            agent_id=agent.id,
-            draft_type=AgentConfigDraftType.DRAFT,
-            account_id=None,
-            draft_owner_key="",
-            base_snapshot_id=snapshot.id,
-            config_snapshot=agent_soul,
-            created_by=agent.created_by,
-            updated_by=agent.updated_by,
-        )
-        db.session.add(draft)
-        db.session.flush()
-        return draft
-
     @staticmethod
     def _resolve_agent_by_id(
         *, tenant_id: str, agent_id: str, snapshot_id: str | None
-    ) -> tuple[Agent, AgentConfigSnapshot | AgentConfigDraft, AgentSoulConfig]:
+    ) -> tuple[Agent, AgentConfigSnapshot, AgentSoulConfig]:
         agent = db.session.scalar(select(Agent).where(Agent.id == agent_id, Agent.tenant_id == tenant_id))
         if agent is None:
             raise AgentAppGeneratorError("Agent not found")
         if not snapshot_id:
             raise AgentAppGeneratorError("Agent has no published version")
-        snapshot = db.session.scalar(
-            select(AgentConfigSnapshot).where(
-                AgentConfigSnapshot.tenant_id == tenant_id,
-                AgentConfigSnapshot.agent_id == agent_id,
-                AgentConfigSnapshot.id == snapshot_id,
-            )
-        )
-        if snapshot is not None:
-            agent_soul = AgentSoulConfig.model_validate(snapshot.config_snapshot_dict)
-            return agent, snapshot, agent_soul
-        draft = db.session.scalar(
-            select(AgentConfigDraft).where(
-                AgentConfigDraft.tenant_id == tenant_id,
-                AgentConfigDraft.agent_id == agent_id,
-                AgentConfigDraft.id == snapshot_id,
-            )
-        )
-        if draft is None:
+        snapshot = db.session.scalar(select(AgentConfigSnapshot).where(AgentConfigSnapshot.id == snapshot_id))
+        if snapshot is None:
             raise AgentAppGeneratorError("Agent published version not found")
-        agent_soul = AgentSoulConfig.model_validate(draft.config_snapshot_dict)
-        return agent, draft, agent_soul
+        agent_soul = AgentSoulConfig.model_validate(snapshot.config_snapshot_dict)
+        return agent, snapshot, agent_soul
 
 
 __all__ = ["AgentAppGenerator", "AgentAppGeneratorError"]

api/core/app/apps/base_app_generator.py

@@ -32,6 +32,7 @@ class _DebuggerDraftVariableSaver:
         self,
         *,
         account: Account,
+        tenant_id: str,
         app_id: str,
         node_id: str,
         node_type: NodeType,
@@ -39,6 +40,7 @@ class _DebuggerDraftVariableSaver:
         enclosing_node_id: str | None = None,
     ) -> None:
         self._account = account
+        self._tenant_id = tenant_id
         self._app_id = app_id
         self._node_id = node_id
         self._node_type = node_type
@@ -49,6 +51,7 @@ class _DebuggerDraftVariableSaver:
         with Session(db.engine) as session, session.begin():
             DraftVariableSaverImpl(
                 session=session,
+                tenant_id=self._tenant_id,
                 app_id=self._app_id,
                 node_id=self._node_id,
                 node_type=self._node_type,
@@ -287,7 +290,12 @@ class BaseAppGenerator:
 
     @final
     @staticmethod
-    def _get_draft_var_saver_factory(invoke_from: InvokeFrom, account: Account | EndUser) -> DraftVariableSaverFactory:
+    def _get_draft_var_saver_factory(
+        invoke_from: InvokeFrom,
+        account: Account | EndUser,
+        *,
+        tenant_id: str,
+    ) -> DraftVariableSaverFactory:
         if invoke_from == InvokeFrom.DEBUGGER:
             assert isinstance(account, Account)
 
@@ -300,6 +308,7 @@ class BaseAppGenerator:
             ) -> DraftVariableSaver:
                 return _DebuggerDraftVariableSaver(
                     account=account,
+                    tenant_id=tenant_id,
                     app_id=app_id,
                     node_id=node_id,
                     node_type=node_type,

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

@@ -214,6 +214,7 @@ class PipelineGenerator(BaseAppGenerator):
             session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
             workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
                 session_factory=session_factory,
+                tenant_id=pipeline.tenant_id,
                 user=user,
                 app_id=application_generate_entity.app_config.app_id,
                 triggered_from=workflow_triggered_from,
@@ -221,6 +222,7 @@ class PipelineGenerator(BaseAppGenerator):
 
             workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
                 session_factory=session_factory,
+                tenant_id=pipeline.tenant_id,
                 user=user,
                 app_id=application_generate_entity.app_config.app_id,
                 triggered_from=WorkflowNodeExecutionTriggeredFrom.RAG_PIPELINE_RUN,
@@ -341,6 +343,7 @@ class PipelineGenerator(BaseAppGenerator):
             draft_var_saver_factory = self._get_draft_var_saver_factory(
                 invoke_from,
                 user,
+                tenant_id=pipeline.tenant_id,
             )
             # return response or stream generator
             response = self._handle_response(
@@ -417,6 +420,7 @@ class PipelineGenerator(BaseAppGenerator):
 
         workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
             session_factory=session_factory,
+            tenant_id=pipeline.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowRunTriggeredFrom.RAG_PIPELINE_DEBUGGING,
@@ -424,6 +428,7 @@ class PipelineGenerator(BaseAppGenerator):
 
         workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
             session_factory=session_factory,
+            tenant_id=pipeline.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
@@ -513,6 +518,7 @@ class PipelineGenerator(BaseAppGenerator):
 
         workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
             session_factory=session_factory,
+            tenant_id=pipeline.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowRunTriggeredFrom.RAG_PIPELINE_DEBUGGING,
@@ -520,6 +526,7 @@ class PipelineGenerator(BaseAppGenerator):
 
         workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
             session_factory=session_factory,
+            tenant_id=pipeline.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,

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

@@ -42,6 +42,7 @@ from core.repositories import DifyCoreRepositoryFactory
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
+from graphon.filters import ResponseStreamFilter
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from graphon.runtime import GraphRuntimeState
@@ -241,6 +242,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
             workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
                 session_factory=session_factory,
+                tenant_id=app_model.tenant_id,
                 user=user,
                 app_id=application_generate_entity.app_config.app_id,
                 triggered_from=workflow_triggered_from,
@@ -248,6 +250,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
             # Create workflow node execution repository
             workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
                 session_factory=session_factory,
+                tenant_id=app_model.tenant_id,
                 user=user,
                 app_id=application_generate_entity.app_config.app_id,
                 triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
@@ -280,6 +283,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         pause_state_config: PauseStateLayerConfig | None = None,
         variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
+        response_stream_filter: ResponseStreamFilter | None = None,
     ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
         """
         Resume a paused workflow execution using the persisted runtime state.
@@ -310,6 +314,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
             graph_engine_layers=graph_engine_layers,
             graph_runtime_state=graph_runtime_state,
             pause_state_config=pause_state_config,
+            response_stream_filter=response_stream_filter,
         )
 
     def _generate(
@@ -328,6 +333,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         graph_runtime_state: GraphRuntimeState | None = None,
         pause_state_config: PauseStateLayerConfig | None = None,
+        response_stream_filter: ResponseStreamFilter | None = None,
     ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
         """
         Generate App response.
@@ -356,12 +362,14 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 app_mode=app_model.mode,
             )
 
+            resolved_response_stream_filter = response_stream_filter or ResponseStreamFilter()
             if pause_state_config is not None:
                 graph_layers.append(
                     PauseStatePersistenceLayer(
                         session_factory=pause_state_config.session_factory,
                         generate_entity=application_generate_entity,
                         state_owner_user_id=pause_state_config.state_owner_user_id,
+                        response_stream_filter=resolved_response_stream_filter,
                     )
                 )
 
@@ -384,12 +392,17 @@ class WorkflowAppGenerator(BaseAppGenerator):
                     "workflow_node_execution_repository": workflow_node_execution_repository,
                     "graph_engine_layers": tuple(graph_layers),
                     "graph_runtime_state": graph_runtime_state,
+                    "response_stream_filter": resolved_response_stream_filter,
                 },
             )
 
             worker_thread.start()
 
-            draft_var_saver_factory = self._get_draft_var_saver_factory(invoke_from, user)
+            draft_var_saver_factory = self._get_draft_var_saver_factory(
+                invoke_from,
+                user,
+                tenant_id=app_model.tenant_id,
+            )
 
             # return response or stream generator
             response = self._handle_response(
@@ -459,6 +472,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         # Create workflow execution(aka workflow run) repository
         workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
             session_factory=session_factory,
+            tenant_id=app_model.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
@@ -466,6 +480,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         # Create workflow node execution repository
         workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
             session_factory=session_factory,
+            tenant_id=app_model.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
@@ -546,6 +561,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         # Create workflow execution(aka workflow run) repository
         workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
             session_factory=session_factory,
+            tenant_id=app_model.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
@@ -553,6 +569,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         # Create workflow node execution repository
         workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
             session_factory=session_factory,
+            tenant_id=app_model.tenant_id,
             user=user,
             app_id=application_generate_entity.app_config.app_id,
             triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
@@ -590,6 +607,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         graph_runtime_state: GraphRuntimeState | None = None,
+        response_stream_filter: ResponseStreamFilter | None = None,
     ) -> None:
         """
         Generate worker in a new thread.
@@ -638,6 +656,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 root_node_id=root_node_id,
                 graph_engine_layers=graph_engine_layers,
                 graph_runtime_state=graph_runtime_state,
+                response_stream_filter=response_stream_filter,
             )
 
             try:

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

@@ -18,6 +18,7 @@ from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_redis import redis_client
 from extensions.otel import WorkflowAppRunnerHandler, trace_span
 from graphon.enums import WorkflowType
+from graphon.filters import ResponseStreamFilter
 from graphon.graph_engine.command_channels import RedisChannel
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.runtime import GraphRuntimeState, VariablePool
@@ -46,6 +47,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         graph_runtime_state: GraphRuntimeState | None = None,
+        response_stream_filter: ResponseStreamFilter | None = None,
     ):
         super().__init__(
             queue_manager=queue_manager,
@@ -60,6 +62,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         self._workflow_execution_repository = workflow_execution_repository
         self._workflow_node_execution_repository = workflow_node_execution_repository
         self._resume_graph_runtime_state = graph_runtime_state
+        self._response_stream_filter = response_stream_filter
 
     @trace_span(WorkflowAppRunnerHandler)
     def run(self):
@@ -163,6 +166,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             variable_pool=variable_pool,
             graph_runtime_state=graph_runtime_state,
             command_channel=command_channel,
+            response_stream_filter=self._response_stream_filter,
         )
 
         persistence_layer = WorkflowPersistenceLayer(

api/core/app/layers/pause_state_persist_layer.py

@@ -7,6 +7,7 @@ from sqlalchemy.orm import Session, sessionmaker
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
 from core.workflow.system_variables import SystemVariableKey, get_system_text
+from graphon.filters import ResponseStreamFilter
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import GraphEngineEvent, GraphRunPausedEvent
 from models.model import AppMode
@@ -41,6 +42,10 @@ class WorkflowResumptionContext(BaseModel):
     # Only workflow / chatflow could be paused.
     generate_entity: _GenerateEntityUnion
     serialized_graph_runtime_state: str
+    # Optional so that a workflow run paused before this field existed still
+    # loads: it just degrades to fresh-filter behavior on resume for that one
+    # stale run.
+    serialized_response_stream_filter_state: str | None = None
 
     def dumps(self) -> str:
         return self.model_dump_json()
@@ -52,6 +57,12 @@ class WorkflowResumptionContext(BaseModel):
     def get_generate_entity(self) -> WorkflowAppGenerateEntity | AdvancedChatAppGenerateEntity:
         return self.generate_entity.entity
 
+    def get_response_stream_filter(self) -> ResponseStreamFilter:
+        response_stream_filter = ResponseStreamFilter()
+        if self.serialized_response_stream_filter_state is not None:
+            response_stream_filter.loads(self.serialized_response_stream_filter_state)
+        return response_stream_filter
+
 
 @dataclass(frozen=True)
 class PauseStateLayerConfig:
@@ -67,11 +78,17 @@ class PauseStatePersistenceLayer(GraphEngineLayer):
         session_factory: Engine | sessionmaker[Session],
         generate_entity: WorkflowAppGenerateEntity | AdvancedChatAppGenerateEntity,
         state_owner_user_id: str,
+        response_stream_filter: ResponseStreamFilter,
     ):
         """Create a PauseStatePersistenceLayer.
 
         The `state_owner_user_id` is used when creating state file for pause.
         It generally should id of the creator of workflow.
+
+        `response_stream_filter` must be the exact same instance that
+        `WorkflowEntry` is using to stream this run's events — this layer
+        dumps its state on pause, and a different instance would silently
+        persist the wrong (empty) filter state.
         """
         if isinstance(session_factory, Engine):
             session_factory = sessionmaker(session_factory)
@@ -79,6 +96,7 @@ class PauseStatePersistenceLayer(GraphEngineLayer):
         self._session_maker = session_factory
         self._state_owner_user_id = state_owner_user_id
         self._generate_entity = generate_entity
+        self._response_stream_filter = response_stream_filter
 
     def _get_repo(self) -> APIWorkflowRunRepository:
         return DifyAPIRepositoryFactory.create_api_workflow_run_repository(self._session_maker)
@@ -119,6 +137,7 @@ class PauseStatePersistenceLayer(GraphEngineLayer):
         state = WorkflowResumptionContext(
             serialized_graph_runtime_state=self.graph_runtime_state.dumps(),
             generate_entity=entity_wrapper,
+            serialized_response_stream_filter_state=self._response_stream_filter.dumps(),
         )
 
         workflow_run_id = get_system_text(

api/core/callback_handler/workflow_tool_callback_handler.py

@@ -1,6 +1,7 @@
 from collections.abc import Generator, Iterable, Mapping
 from typing import Any
 
+from configs import dify_config
 from core.callback_handler.agent_tool_callback_handler import DifyAgentCallbackHandler, print_text
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.tools.entities.tool_entities import ToolInvokeMessage
@@ -19,8 +20,9 @@ class DifyWorkflowCallbackHandler(DifyAgentCallbackHandler):
         trace_manager: TraceQueueManager | None = None,
     ) -> Generator[ToolInvokeMessage, None, None]:
         for tool_output in tool_outputs:
-            print_text("\n[on_tool_execution]\n", color=self.color)
-            print_text("Tool: " + tool_name + "\n", color=self.color)
-            print_text("Outputs: " + tool_output.model_dump_json()[:1000] + "\n", color=self.color)
-            print_text("\n")
+            if dify_config.DEBUG:
+                print_text("\n[on_tool_execution]\n", color=self.color)
+                print_text("Tool: " + tool_name + "\n", color=self.color)
+                print_text("Outputs: " + tool_output.model_dump_json()[:1000] + "\n", color=self.color)
+                print_text("\n")
             yield tool_output

api/core/entities/provider_configuration.py

@@ -22,10 +22,7 @@ from core.entities.provider_entities import (
     SystemConfigurationStatus,
 )
 from core.helper import encrypter
-from core.helper.model_provider_cache import (
-    ProviderCredentialsCache,
-    ProviderCredentialsCacheType,
-)
+from core.helper.model_provider_cache import ProviderCredentialsCache, ProviderCredentialsCacheType
 from core.plugin.impl.model_runtime_factory import create_model_type_instance, create_plugin_model_assembly
 from graphon.model_runtime.entities.model_entities import AIModelEntity, FetchFrom, ModelType
 from graphon.model_runtime.entities.provider_entities import (
@@ -476,39 +473,6 @@ class ProviderConfiguration(BaseModel):
             provider_names.append(model_provider_id.provider_name)
         return provider_names
 
-    def _invalidate_provider_configuration_cache(
-        self,
-        *,
-        provider_models: bool = False,
-        preferred_model_providers: bool = False,
-        provider_model_settings: bool = False,
-        provider_model_credentials: bool = False,
-        provider_credentials: bool = False,
-        provider_load_balancing_configs: bool = False,
-    ) -> None:
-        """Invalidate tenant-scoped provider snapshots after committing configuration writes."""
-        from core.provider_manager import ProviderConfigurationCacheSource, ProviderManager
-
-        sources: list[ProviderConfigurationCacheSource] = []
-        if provider_models:
-            sources.append(ProviderConfigurationCacheSource.PROVIDER_MODELS)
-        if preferred_model_providers:
-            sources.append(ProviderConfigurationCacheSource.PREFERRED_MODEL_PROVIDERS)
-        if provider_model_settings:
-            sources.append(ProviderConfigurationCacheSource.PROVIDER_MODEL_SETTINGS)
-        if provider_model_credentials:
-            sources.append(ProviderConfigurationCacheSource.PROVIDER_MODEL_CREDENTIALS)
-        if provider_credentials:
-            sources.append(ProviderConfigurationCacheSource.PROVIDER_CREDENTIALS)
-        if provider_load_balancing_configs:
-            sources.append(ProviderConfigurationCacheSource.PROVIDER_LOAD_BALANCING_CONFIGS)
-
-        if not sources:
-            logger.warning("No provider configuration cache source selected for invalidation")
-            return
-
-        ProviderManager.invalidate_configurations_cache(self.tenant_id, sources=sources)
-
     def create_provider_credential(self, credentials: dict[str, Any], credential_name: str | None):
         """
         Add custom provider credentials.
@@ -525,7 +489,6 @@ class ProviderConfiguration(BaseModel):
 
         credentials = self.validate_provider_credentials(credentials=credentials)
 
-        preferred_model_providers_changed = False
         with Session(db.engine) as session:
             provider_record = self._get_provider_record(session)
             try:
@@ -555,9 +518,7 @@ class ProviderConfiguration(BaseModel):
                     )
                     provider_model_credentials_cache.delete()
 
-                    preferred_model_providers_changed = self.switch_preferred_provider_type(
-                        provider_type=ProviderType.CUSTOM, session=session
-                    )
+                    self.switch_preferred_provider_type(provider_type=ProviderType.CUSTOM, session=session)
                 else:
                     provider_record.is_valid = True
 
@@ -572,18 +533,12 @@ class ProviderConfiguration(BaseModel):
                         )
                         provider_model_credentials_cache.delete()
 
-                        preferred_model_providers_changed = self.switch_preferred_provider_type(
-                            provider_type=ProviderType.CUSTOM, session=session
-                        )
+                        self.switch_preferred_provider_type(provider_type=ProviderType.CUSTOM, session=session)
 
                 session.commit()
             except Exception:
                 session.rollback()
                 raise
-        self._invalidate_provider_configuration_cache(
-            preferred_model_providers=preferred_model_providers_changed,
-            provider_credentials=True,
-        )
 
     def update_provider_credential(
         self,
@@ -607,7 +562,6 @@ class ProviderConfiguration(BaseModel):
 
         credentials = self.validate_provider_credentials(credentials=credentials, credential_id=credential_id)
 
-        load_balancing_configs_changed = False
         with Session(db.engine) as session:
             provider_record = self._get_provider_record(session)
             stmt = select(ProviderCredential).where(
@@ -634,7 +588,7 @@ class ProviderConfiguration(BaseModel):
                     )
                     provider_model_credentials_cache.delete()
 
-                load_balancing_configs_changed = self._update_load_balancing_configs_with_credential(
+                self._update_load_balancing_configs_with_credential(
                     credential_id=credential_id,
                     credential_record=credential_record,
                     credential_source=CredentialSourceType.PROVIDER,
@@ -643,10 +597,6 @@ class ProviderConfiguration(BaseModel):
             except Exception:
                 session.rollback()
                 raise
-        self._invalidate_provider_configuration_cache(
-            provider_credentials=True,
-            provider_load_balancing_configs=load_balancing_configs_changed,
-        )
 
     def _update_load_balancing_configs_with_credential(
         self,
@@ -654,7 +604,7 @@ class ProviderConfiguration(BaseModel):
         credential_record: ProviderCredential | ProviderModelCredential,
         credential_source: str,
         session: Session,
-    ) -> bool:
+    ):
         """
         Update load balancing configurations that reference the given credential_id.
 
@@ -675,7 +625,7 @@ class ProviderConfiguration(BaseModel):
         load_balancing_configs = session.execute(stmt).scalars().all()
 
         if not load_balancing_configs:
-            return False
+            return
 
         # Update each load balancing config with the new credentials
         for lb_config in load_balancing_configs:
@@ -693,7 +643,6 @@ class ProviderConfiguration(BaseModel):
             lb_credentials_cache.delete()
 
         session.commit()
-        return True
 
     def delete_provider_credential(self, credential_id: str):
         """
@@ -702,8 +651,6 @@ class ProviderConfiguration(BaseModel):
         :param credential_id: credential id
         :return:
         """
-        preferred_model_providers_changed = False
-        load_balancing_configs_changed = False
         with Session(db.engine) as session:
             stmt = select(ProviderCredential).where(
                 ProviderCredential.id == credential_id,
@@ -724,7 +671,6 @@ class ProviderConfiguration(BaseModel):
                 LoadBalancingModelConfig.credential_source_type == CredentialSourceType.PROVIDER,
             )
             lb_configs_using_credential = session.execute(lb_stmt).scalars().all()
-            load_balancing_configs_changed = bool(lb_configs_using_credential)
             try:
                 for lb_config in lb_configs_using_credential:
                     lb_credentials_cache = ProviderCredentialsCache(
@@ -757,9 +703,7 @@ class ProviderConfiguration(BaseModel):
                         cache_type=ProviderCredentialsCacheType.PROVIDER,
                     )
                     provider_model_credentials_cache.delete()
-                    preferred_model_providers_changed = self.switch_preferred_provider_type(
-                        provider_type=ProviderType.SYSTEM, session=session
-                    )
+                    self.switch_preferred_provider_type(provider_type=ProviderType.SYSTEM, session=session)
                 elif provider_record and provider_record.credential_id == credential_id:
                     provider_record.credential_id = None
                     provider_record.updated_at = naive_utc_now()
@@ -770,19 +714,12 @@ class ProviderConfiguration(BaseModel):
                         cache_type=ProviderCredentialsCacheType.PROVIDER,
                     )
                     provider_model_credentials_cache.delete()
-                    preferred_model_providers_changed = self.switch_preferred_provider_type(
-                        provider_type=ProviderType.SYSTEM, session=session
-                    )
+                    self.switch_preferred_provider_type(provider_type=ProviderType.SYSTEM, session=session)
 
                 session.commit()
             except Exception:
                 session.rollback()
                 raise
-        self._invalidate_provider_configuration_cache(
-            preferred_model_providers=preferred_model_providers_changed,
-            provider_credentials=True,
-            provider_load_balancing_configs=load_balancing_configs_changed,
-        )
 
     def switch_active_provider_credential(self, credential_id: str):
         """
@@ -791,7 +728,6 @@ class ProviderConfiguration(BaseModel):
         :param credential_id: credential id
         :return:
         """
-        preferred_model_providers_changed = False
         with Session(db.engine) as session:
             stmt = select(ProviderCredential).where(
                 ProviderCredential.id == credential_id,
@@ -817,14 +753,10 @@ class ProviderConfiguration(BaseModel):
                     cache_type=ProviderCredentialsCacheType.PROVIDER,
                 )
                 provider_model_credentials_cache.delete()
-                preferred_model_providers_changed = self.switch_preferred_provider_type(
-                    ProviderType.CUSTOM, session=session
-                )
+                self.switch_preferred_provider_type(ProviderType.CUSTOM, session=session)
             except Exception:
                 session.rollback()
                 raise
-        if preferred_model_providers_changed:
-            self._invalidate_provider_configuration_cache(preferred_model_providers=True)
 
     def _get_custom_model_record(
         self,
@@ -1085,10 +1017,6 @@ class ProviderConfiguration(BaseModel):
             except Exception:
                 session.rollback()
                 raise
-        self._invalidate_provider_configuration_cache(
-            provider_models=True,
-            provider_model_credentials=True,
-        )
 
     def update_custom_model_credential(
         self,
@@ -1125,7 +1053,6 @@ class ProviderConfiguration(BaseModel):
             credential_id=credential_id,
         )
 
-        load_balancing_configs_changed = False
         with Session(db.engine) as session:
             provider_model_record = self._get_custom_model_record(model_type=model_type, model=model, session=session)
 
@@ -1155,7 +1082,7 @@ class ProviderConfiguration(BaseModel):
                     )
                     provider_model_credentials_cache.delete()
 
-                load_balancing_configs_changed = self._update_load_balancing_configs_with_credential(
+                self._update_load_balancing_configs_with_credential(
                     credential_id=credential_id,
                     credential_record=credential_record,
                     credential_source=CredentialSourceType.CUSTOM_MODEL,
@@ -1164,11 +1091,6 @@ class ProviderConfiguration(BaseModel):
             except Exception:
                 session.rollback()
                 raise
-        self._invalidate_provider_configuration_cache(
-            provider_models=True,
-            provider_model_credentials=True,
-            provider_load_balancing_configs=load_balancing_configs_changed,
-        )
 
     def delete_custom_model_credential(self, model_type: ModelType, model: str, credential_id: str):
         """
@@ -1177,7 +1099,6 @@ class ProviderConfiguration(BaseModel):
         :param credential_id: credential id
         :return:
         """
-        load_balancing_configs_changed = False
         with Session(db.engine) as session:
             stmt = select(ProviderModelCredential).where(
                 ProviderModelCredential.id == credential_id,
@@ -1197,7 +1118,6 @@ class ProviderConfiguration(BaseModel):
                 LoadBalancingModelConfig.credential_source_type == CredentialSourceType.CUSTOM_MODEL,
             )
             lb_configs_using_credential = session.execute(lb_stmt).scalars().all()
-            load_balancing_configs_changed = bool(lb_configs_using_credential)
 
             try:
                 for lb_config in lb_configs_using_credential:
@@ -1241,11 +1161,6 @@ class ProviderConfiguration(BaseModel):
             except Exception:
                 session.rollback()
                 raise
-        self._invalidate_provider_configuration_cache(
-            provider_models=True,
-            provider_model_credentials=True,
-            provider_load_balancing_configs=load_balancing_configs_changed,
-        )
 
     def add_model_credential_to_model(self, model_type: ModelType, model: str, credential_id: str):
         """
@@ -1298,7 +1213,6 @@ class ProviderConfiguration(BaseModel):
 
             session.add(provider_model_record)
             session.commit()
-        self._invalidate_provider_configuration_cache(provider_models=True)
 
     def switch_custom_model_credential(self, model_type: ModelType, model: str, credential_id: str):
         """
@@ -1337,7 +1251,6 @@ class ProviderConfiguration(BaseModel):
                 cache_type=ProviderCredentialsCacheType.MODEL,
             )
             provider_model_credentials_cache.delete()
-        self._invalidate_provider_configuration_cache(provider_models=True)
 
     def delete_custom_model(self, model_type: ModelType, model: str):
         """
@@ -1346,7 +1259,6 @@ class ProviderConfiguration(BaseModel):
         :param model: model name
         :return:
         """
-        provider_models_changed = False
         with Session(db.engine) as session:
             # get provider model
             provider_model_record = self._get_custom_model_record(model_type=model_type, model=model, session=session)
@@ -1355,7 +1267,6 @@ class ProviderConfiguration(BaseModel):
             if provider_model_record:
                 session.delete(provider_model_record)
                 session.commit()
-                provider_models_changed = True
 
                 provider_model_credentials_cache = ProviderCredentialsCache(
                     tenant_id=self.tenant_id,
@@ -1364,8 +1275,6 @@ class ProviderConfiguration(BaseModel):
                 )
 
                 provider_model_credentials_cache.delete()
-        if provider_models_changed:
-            self._invalidate_provider_configuration_cache(provider_models=True)
 
     def _get_provider_model_setting(
         self, model_type: ModelType, model: str, session: Session
@@ -1405,7 +1314,6 @@ class ProviderConfiguration(BaseModel):
                 )
                 session.add(model_setting)
             session.commit()
-        self._invalidate_provider_configuration_cache(provider_model_settings=True)
 
         return model_setting
 
@@ -1432,7 +1340,6 @@ class ProviderConfiguration(BaseModel):
                 )
                 session.add(model_setting)
             session.commit()
-        self._invalidate_provider_configuration_cache(provider_model_settings=True)
 
         return model_setting
 
@@ -1485,7 +1392,6 @@ class ProviderConfiguration(BaseModel):
                 )
                 session.add(model_setting)
             session.commit()
-        self._invalidate_provider_configuration_cache(provider_model_settings=True)
 
         return model_setting
 
@@ -1513,7 +1419,6 @@ class ProviderConfiguration(BaseModel):
                 )
                 session.add(model_setting)
             session.commit()
-        self._invalidate_provider_configuration_cache(provider_model_settings=True)
 
         return model_setting
 
@@ -1549,19 +1454,19 @@ class ProviderConfiguration(BaseModel):
             credentials=credentials or {},
         )
 
-    def switch_preferred_provider_type(self, provider_type: ProviderType, session: Session | None = None) -> bool:
+    def switch_preferred_provider_type(self, provider_type: ProviderType, session: Session | None = None):
         """
         Switch preferred provider type.
         :param provider_type:
         :return:
         """
         if provider_type == self.preferred_provider_type:
-            return False
+            return
 
         if provider_type == ProviderType.SYSTEM and not self.system_configuration.enabled:
-            return False
+            return
 
-        def _switch(s: Session) -> bool:
+        def _switch(s: Session):
             stmt = select(TenantPreferredModelProvider).where(
                 TenantPreferredModelProvider.tenant_id == self.tenant_id,
                 TenantPreferredModelProvider.provider_name.in_(self._get_provider_names()),
@@ -1578,16 +1483,12 @@ class ProviderConfiguration(BaseModel):
                 )
                 s.add(preferred_model_provider)
             s.commit()
-            return True
 
         if session:
             return _switch(session)
         else:
             with Session(db.engine) as session:
-                changed = _switch(session)
-            if changed:
-                self._invalidate_provider_configuration_cache(preferred_model_providers=True)
-            return changed
+                return _switch(session)
 
     def extract_secret_variables(self, credential_form_schemas: list[CredentialFormSchema]) -> list[str]:
         """

api/core/helper/code_executor/code_executor.py

@@ -13,7 +13,7 @@ from core.helper.code_executor.jinja2.jinja2_transformer import Jinja2TemplateTr
 from core.helper.code_executor.python3.python3_transformer import Python3TemplateTransformer
 from core.helper.code_executor.template_transformer import TemplateTransformer
 from core.helper.http_client_pooling import get_pooled_http_client
-from graphon.nodes.code.entities import CodeLanguage
+from graphon.nodes.code.entities import CodeLanguage as CodeLanguage  # noqa: PLC0414
 
 logger = logging.getLogger(__name__)
 code_execution_endpoint_url = URL(str(dify_config.CODE_EXECUTION_ENDPOINT))
@@ -133,7 +133,9 @@ class CodeExecutor:
         return response_code.data.stdout or ""
 
     @classmethod
-    def execute_workflow_code_template(cls, language: CodeLanguage, code: str, inputs: Mapping[str, Any]):
+    def execute_workflow_code_template(
+        cls, language: CodeLanguage, code: str, inputs: Mapping[str, Any]
+    ) -> dict[str, Any]:
         """
         Execute code
         :param language: code language

api/core/helper/code_executor/jinja2/jinja2_transformer.py

@@ -11,7 +11,7 @@ class Jinja2TemplateTransformer(TemplateTransformer):
 
     @classmethod
     @override
-    def transform_response(cls, response: str):
+    def transform_response(cls, response: str) -> dict[str, Any]:
         """
         Transform response to dict
         :param response: response

api/core/helper/code_executor/template_transformer.py

@@ -36,14 +36,14 @@ class TemplateTransformer(ABC):
         return runner_script, preload_script
 
     @classmethod
-    def extract_result_str_from_response(cls, response: str):
+    def extract_result_str_from_response(cls, response: str) -> str:
         result = re.search(rf"{cls._result_tag}(.*){cls._result_tag}", response, re.DOTALL)
         if not result:
             raise ValueError(f"Failed to parse result: no result tag found in response. Response: {response[:200]}...")
         return result.group(1)
 
     @classmethod
-    def transform_response(cls, response: str) -> Mapping[str, Any]:
+    def transform_response(cls, response: str) -> dict[str, Any]:
         """
         Transform response to dict
         :param response: response
@@ -71,7 +71,7 @@ class TemplateTransformer(ABC):
         return result
 
     @classmethod
-    def _post_process_result(cls, result: dict[Any, Any]) -> dict[Any, Any]:
+    def _post_process_result(cls, result: dict[str, Any]) -> dict[str, Any]:
         """
         Post-process the result to convert scientific notation strings back to numbers
         """
@@ -89,7 +89,7 @@ class TemplateTransformer(ABC):
                     return [convert_scientific_notation(v) for v in value]
             return value
 
-        return convert_scientific_notation(result)
+        return {key: convert_scientific_notation(value) for key, value in result.items()}
 
     @classmethod
     @abstractmethod

api/core/helper/creators.py

@@ -24,7 +24,7 @@ def upload_dsl(dsl_file_bytes: bytes, filename: str = "template.yaml") -> str:
     response.raise_for_status()
     data = response.json()
     claim_code = data.get("data", {}).get("claim_code")
-    if not claim_code:
+    if not isinstance(claim_code, str) or not claim_code:
         raise ValueError("Creators Platform did not return a valid claim_code")
     return claim_code
 

api/core/helper/credential_utils.py

@@ -10,18 +10,21 @@ def is_credential_exists(credential_id: str, credential_type: "PluginCredentialT
     """
     Check if the credential still exists in the database.
 
+    Uses the configured SQLAlchemy session factory instead of Flask-SQLAlchemy's
+    ``db.engine`` because workflow graph node construction may run without an
+    active Flask application context.
+
     :param credential_id: The credential ID to check
     :param credential_type: The type of credential (MODEL or TOOL)
     :return: True if credential exists, False otherwise
     """
     from sqlalchemy import select
-    from sqlalchemy.orm import Session
 
-    from extensions.ext_database import db
+    from core.db import session_factory
     from models.provider import ProviderCredential, ProviderModelCredential
     from models.tools import BuiltinToolProvider
 
-    with Session(db.engine) as session:
+    with session_factory.create_session() as session:
         if credential_type == PluginCredentialType.MODEL:
             # Check both pre-defined and custom model credentials using a single UNION query
             stmt = (
@@ -42,7 +45,7 @@ def is_credential_exists(credential_id: str, credential_type: "PluginCredentialT
 
 def runtime_check_credential_policy_compliance(
     credential_id: str, provider: str, credential_type: "PluginCredentialType", check_existence: bool = True
-):
+) -> None:
     if dify_config.ENTERPRISE_DISABLE_RUNTIME_CREDENTIAL_CHECK:
         return
     check_credential_policy_compliance(

api/core/helper/download.py

@@ -1,4 +1,7 @@
-def download_with_size_limit(url, max_download_size: int, **kwargs):
+from typing import Any
+
+
+def download_with_size_limit(url: str, max_download_size: int, **kwargs: Any) -> bytes:
     from core.file import remote_fetcher
 
     response = remote_fetcher.make_request("GET", url, follow_redirects=True, **kwargs)

api/core/helper/encrypter.py

@@ -1,5 +1,7 @@
 import base64
 
+from Crypto.PublicKey import RSA
+
 from libs import rsa
 
 
@@ -11,13 +13,13 @@ def obfuscated_token(token: str) -> str:
     return token[:6] + "*" * 12 + token[-2:]
 
 
-def full_mask_token(token_length=20):
+def full_mask_token(token_length: int = 20) -> str:
     return "*" * token_length
 
 
-def encrypt_token(tenant_id: str, token: str):
-    from extensions.ext_database import db
+def encrypt_token(tenant_id: str, token: str) -> str:
     from models.account import Tenant
+    from models.engine import db
 
     if not (tenant := db.session.get(Tenant, tenant_id)):
         raise ValueError(f"Tenant with id {tenant_id} not found")
@@ -30,15 +32,15 @@ def decrypt_token(tenant_id: str, token: str) -> str:
     return rsa.decrypt(base64.b64decode(token), tenant_id)
 
 
-def batch_decrypt_token(tenant_id: str, tokens: list[str]):
+def batch_decrypt_token(tenant_id: str, tokens: list[str]) -> list[str]:
     rsa_key, cipher_rsa = rsa.get_decrypt_decoding(tenant_id)
 
     return [rsa.decrypt_token_with_decoding(base64.b64decode(token), rsa_key, cipher_rsa) for token in tokens]
 
 
-def get_decrypt_decoding(tenant_id: str):
+def get_decrypt_decoding(tenant_id: str) -> tuple[RSA.RsaKey, object]:
     return rsa.get_decrypt_decoding(tenant_id)
 
 
-def decrypt_token_with_decoding(token: str, rsa_key, cipher_rsa):
+def decrypt_token_with_decoding(token: str, rsa_key: RSA.RsaKey, cipher_rsa: object) -> str:
     return rsa.decrypt_token_with_decoding(base64.b64decode(token), rsa_key, cipher_rsa)