Revert "chore: improve lambda plugin runtime error display"

This reverts commit 6bc1508a6b.

.agents/skills/frontend-testing/SKILL.md

@@ -102,11 +102,11 @@ describe('ComponentName', () => {
     })
   })
 
-  // Props tests (REQUIRED)
+  // Props tests (REQUIRED when props change observable behavior)
   describe('Props', () => {
-    it('should apply custom className', () => {
-      render(<Component className="custom" />)
-      expect(screen.getByRole('button')).toHaveClass('custom')
+    it('should disable the action when disabled', () => {
+      render(<Component disabled />)
+      expect(screen.getByRole('button')).toBeDisabled()
     })
   })
 
@@ -220,6 +220,7 @@ Every test should clearly separate:
 ### 2. Black-Box Testing
 
 - Test observable behavior, not implementation details
+- Test product contracts, not cosmetic implementation. Do not add or expand unit tests only to lock pure style classes, spacing, colors, backgrounds, or layout micro-adjustments. Cover visual-only fixes with browser/manual verification, screenshots, or E2E/visual checks when risk justifies it. Add unit tests only when the change affects user-observable behavior, accessibility semantics, state, data flow, routing, or a stable component API contract.
 - Use semantic queries (`getByRole` with accessible `name`, `getByLabelText`, `getByPlaceholderText`, `getByText`, and scoped `within(...)`)
 - Treat `getByTestId` as a last resort. If a control cannot be found by role/name, label, landmark, or dialog scope, fix the component accessibility first instead of adding or relying on `data-testid`.
 - Remove production `data-testid` attributes when semantic selectors can cover the behavior. Keep them only for non-visual mocked boundaries, editor/browser shims such as Monaco, canvas/chart output, or third-party widgets with no accessible DOM in the test environment.
@@ -273,7 +274,7 @@ it('should disable input when isReadOnly is true')
 ### Always Required (All Components)
 
 1. **Rendering**: Component renders without crashing
-1. **Props**: Required props, optional props, default values
+1. **Props**: Required props, optional props, default values that change observable behavior. Do not test pass-through styling props such as `className` unless they are an explicit, stable component API whose absence would break a real integration contract.
 1. **Edge Cases**: null, undefined, empty values, boundary conditions
 
 ### Conditional (When Present)

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

@@ -1,54 +1,59 @@
 ---
 name: how-to-write-component
-description: React/TypeScript component style guide. Use when writing, refactoring, or reviewing React components, especially around abstraction choices, props typing, state boundaries, shared local state with Jotai atoms, API types, query/mutation contracts, navigation, memoization, wrappers, and empty-state handling.
+description: Use when writing, refactoring, or reviewing React/TypeScript components in Dify web, especially decisions about component ownership, props/types, URL/query state, Jotai state, async state, generated API contracts, queries/mutations, overlays, effects, navigation, performance, and empty states.
 ---
 
 # How To Write A Component
 
-Use this as the decision guide for React/TypeScript component structure. Existing code is reference material, not automatic precedent; when it conflicts with these rules, adapt the approach instead of reproducing the violation.
+Use this as the component decision guide for Dify web. Existing code is reference material, not automatic precedent; if touched code violates these rules, adapt it and fix equivalent patterns in the same feature branch.
+
+## First Decisions
+
+| 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. |
+| 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. |
+| Should this query/mutation become an atom? | Use TanStack Query hooks at the lowest owner. | It reads atom state, feeds derived atoms, or participates in shared Jotai workflow orchestration. |
+| Should this be a helper/wrapper? | Prefer direct readable code at the use site. | The name captures a stable domain rule or the wrapper owns real behavior, validation, state, error handling, or semantics. |
+| Is an Effect needed? | No. Derive during render or handle the user action in the event handler. | It synchronizes with an external system such as browser APIs, subscriptions, timers, analytics, or imperative DOM/non-React widgets. |
 
 ## Core Defaults
 
-- Search before adding UI, hooks, helpers, or styling patterns. Reuse existing base components, feature components, hooks, utilities, and design styles when they fit.
-- Group code by feature 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.
-- Promote code to shared only when multiple verticals need the same stable primitive. Otherwise keep it local and compose shared primitives inside the owning feature.
-- Prefer local code and purpose-named helpers over catch-all utility modules; do not group workflow-specific defaults, validation, payload shaping, or metadata merging in a generic utils file just because they share a DTO.
-- Keep source/default selection, validation, and payload shaping close to the workflow that owns the behavior. Do not extract a shared helper just because two flows read the same DTO when their priority order, fallback behavior, or submit semantics differ.
-- Prefer direct, readable conditionals at the use site for small branch-specific decisions, especially form source selection and request payload assembly. Extract only when the helper name captures a stable domain rule and removes repeated complexity without hiding flow-specific behavior.
-- When fixing an invalid pattern, scan the touched feature or branch for equivalent patterns and fix them together.
-- 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 guidance.
+- 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: 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.
 
-## Feature Workflow Layout
+## Layout And Ownership
 
-- State-heavy wizards, drawers, modals, and secondary workflows work best as a small feature surface with route/entry files, a single feature-local state file, and feature-local UI.
-- Keep `ui/` shallow with owner files that map to the workflow's real composition boundaries and major visual regions.
-- Owner files contain the section components, field components, skeletons, and one-off helper components that belong to their visual region.
-- Folders represent groups of related files with a shared owner and a stable reason to change together.
-- The entry file handles route integration, provider wiring, close behavior, and feature surface mounting. The composition owner handles high-level workflow branching, and the closest visual owner handles section branching.
+- 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.
+- 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.
+- Avoid prop drilling. One pass-through layer is acceptable; repeated forwarding means ownership should move down or into feature-scoped Jotai UI state. Keep server/cache state in Query and API flow.
+- Do not replace prop drilling with one large view-model hook threaded through section props. Move each hook, query, derived value, and handler to the concrete section that consumes it.
+- Keep callbacks in a parent only for workflow coordination such as form submission, shared selection, batch behavior, or navigation. Otherwise let the child, menu, or row own the action.
 
-## Ownership
+## Feature-Scoped Jotai
 
-- Put local state, queries, mutations, handlers, and derived UI data in the lowest component that uses them. Extract a purpose-built owner component only when the logic has no natural home.
-- Repeated TanStack query calls in sibling components are acceptable when each component independently consumes the data. Do not hoist a query only because it is duplicated; TanStack Query handles deduplication and cache sharing.
-- Hoist state, queries, or callbacks to a parent only when the parent consumes the data, coordinates shared loading/error/empty UI, needs one consistent snapshot, or owns a workflow spanning children.
-- Pass stable domain identity across boundaries; avoid forwarding derived presentation state when the receiver can derive it from its own data source. A component that owns a visual surface should also own the data access, loading, empty, and error states for content rendered inside it unless a parent truly coordinates that state.
-- Loading states for visual surfaces should use skeleton placeholders scoped to the content that is actually loading, with shape, density, and dimensions close to the final UI. Avoid generic loading text or centered spinners for page sections, cards, lists, tables, forms, and drawers; reserve spinners for small inline busy indicators such as an in-progress status icon.
-- Avoid prop drilling. One pass-through layer is acceptable; repeated forwarding means ownership should move down or into feature-scoped Jotai UI state. Keep server/cache state in query and API data flow.
-- Do not replace prop drilling with one top-level hook that returns a large view model and then thread that object through section props. Move each hook, query, derived value, and handler to the concrete section that consumes it, or use feature-scoped Jotai atoms for simple shared form/UI state when siblings need the same source of truth.
-- When using feature-scoped Jotai state for a form, drawer, or other secondary surface, scope the store to that surface instance when stale cross-instance state is possible. Initialize stable config at the owning boundary, then let descendants read only the atoms or purpose-named hooks they actually need.
-- Keep callbacks in a parent only for workflow coordination such as form submission, shared selection, batch behavior, or navigation. Otherwise let the child or row own its action.
-- Prefer uncontrolled DOM state and CSS variables before adding controlled props.
-
-## Feature-Scoped Jotai State
-
-- A module's feature-local state lives in one state file for Jotai-backed features: primitive atoms, query atoms, derived atoms, write-only action atoms, mutation atoms, submission orchestration, provider exports, and optional scope configuration.
-- Atom order in the state file follows the dependency graph: types/constants, editable primitives, query atoms, query-data derived atoms, readiness/business derived atoms, write actions, mutation atoms, submission orchestration, provider exports.
-- Derived atom names read as business facts. Write atom names read as user or workflow commands.
-- UI components read and write the exact atom they use with `useAtomValue` or `useSetAtom`. Repeated workflow semantics live in named derived atoms or write atoms.
-- Non-query derived atoms return a narrow value with a clear domain name. Query atoms expose the TanStack Query result object so loading, error, fetch, and pagination state stay attached to the query contract.
-- Write-only atoms own state transitions that update multiple primitives, reset dependent state, guard stale async work, or advance the workflow.
-- `jotai-tanstack-query` atoms use the same QueryClient as the React Query provider. Query atoms belong in feature state when atoms are the feature's local state surface.
-- Jotai scope is an optional instance-isolation tool for secondary surfaces with independent local state. Query atoms keep shared cache behavior through the shared QueryClient.
+- A Jotai-backed feature has one feature-local state file for shared primitive atoms, query atoms, derived atoms, write-only actions, mutation atoms, submission orchestration, provider exports, and optional scope configuration.
+- Keep component-owned synchronous UI state local even inside Jotai features: dialog open flags, menus/popovers, confirmations, field drafts, and selected local options usually belong in component state.
+- 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.
+- 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.
+- For scoped primitives that are always hydrated by `ScopeProvider`, prefer `atomWithLazy<T>(() => { throw new Error(...) })` when consumers should see a non-null type.
+- Order state files by dependency graph: types/constants, primitives, query atoms, query-data derived atoms, business/readiness derived atoms, write actions, mutation atoms, submission orchestration, provider exports.
+- Name derived atoms as business facts and write atoms as user or workflow commands. Components should read or write the exact atom they need with `useAtomValue` or `useSetAtom`.
+- Menu/dialog `open` state usually stays local, but a scoped atom is acceptable when a composed menu plus secondary surface would otherwise pass confusing `open`/`onClose` props through unrelated layers. Scope that primitive with the surface instance so reset behavior stays local.
+- Keep independent dialog lifecycles separate. Avoid one discriminated "current action dialog" atom when dialogs have separate open state, loading guards, or reset behavior.
 
 ## Components, Props, And Types
 
@@ -57,72 +62,56 @@ Use this as the decision guide for React/TypeScript component structure. Existin
 - Prefer named exports. Use default exports only where the framework requires them, such as Next.js route 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.
-- Do not create type aliases that only rename another type. Use an alias only when it encodes a real UI concept, refinement, or reusable local contract.
-- Name values by their domain role and backend API contract, and keep that name stable across the call chain, especially persistent IDs and route params. Normalize framework or route params at the boundary.
-- Keep fallback and invariant checks at the lowest component that already handles that state; avoid defensive fallbacks that mask impossible states.
-- Do not extract fallback helpers whose only behavior is hiding missing display data. The component that renders the surface owns the empty, disabled, hidden, or placeholder state.
+- 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.
 
-## Generated API Contracts
+## Generated API And Nullable Data
 
 - Treat generated contracts as authoritative at API, query, mutation, cache, and service boundaries. For enterprise APIs, use `packages/contracts/generated/enterprise/*`.
-- Do not hand-write local request/response/reply/page/cache-data types that mirror generated DTOs. Import or infer the generated type.
-- Do not widen generated fields or enums for compatibility. Normalize legacy input at the boundary, then return the generated field type.
-- Do not repair generated or API-returned contract fields in components unless the API contract or product requirement says they need normalization. Treat enums, statuses, and presence flags as exact contract values.
-- Use generated enum objects and union types directly in props, comparisons, status logic, and i18n keys. Do not add local enum constants or parallel frontend enum/status layers unless they model real product state not represented by the API. Presentation-only tone maps should be keyed by the generated enum.
-- Normalize or coerce only at a real boundary, such as user-entered forms, search, URL/query params, file names, DOM IDs, or legacy adapters. Preserve user-entered values when whitespace or formatting can be meaningful.
-- Do not coerce nullable or optional API strings to `''` in query, derived model, or payload-building code. Keep `undefined` or `null` until the final boundary that requires a string.
-- Local UI models are fine for presentation, form state, select options, or guarded required-field refinements. Name them as UI concepts, not generated DTO mirrors.
-- Required-value refinements are allowed only after same-branch filtering or early return. Prefer nullable-tolerant props for render-only data.
-- When a component needs a stricter shape than a generated DTO, refine once at the API/query-to-UI boundary into a purpose-named UI type instead of hiding missing fields with generic fallback or coercion helpers.
-
-## Nullable API Data
-
-- Prefer nullable-tolerant call boundaries. Pass API-returned types through for render-only rows, and let the component render fallback, disabled state, or nothing.
-- Narrow only where a real value is required, such as mutation params, route hrefs, select values, or query input. Build that target model with `flatMap`, a local loop, or an early return so the required value is captured in the same branch.
-- If design says a field is the display value, use that field. Only the final component should decide whether a nullable display value renders a placeholder, hides content, or disables an action.
-- Do not wrap required arrays or fields in null-fallback helpers. Use empty collection fallbacks only for not-yet-loaded query data or genuinely nullable collections at the owning render boundary.
-- Do not drop rows only to satisfy props or React keys; use a stable fallback key when possible.
-- Use conditional spreads or explicit pushes for conditional array items instead of `undefined` placeholders followed by a narrowing filter.
-- Avoid truthiness type guards, `filter(Boolean)`, `filter(item => item.id)`, and `!` after those filters.
-- Use type guards only for meaningful domain or runtime validation, such as enum membership, object shape, or a reusable business invariant.
+- Do not hand-write DTO mirrors, widen generated fields/enums, or add parallel frontend enum/status layers unless they model product state not represented by the API.
+- Use generated enum objects and union types directly in props, comparisons, status logic, and i18n keys. Presentation-only tone maps should be keyed by generated enums.
+- Normalize or coerce only at real boundaries: user-entered forms, search, URL/query params, file names, DOM IDs, or legacy adapters.
+- Do not coerce nullable or optional API strings to `''` in query, derived model, or payload-building code. Keep `null` or `undefined` until the final boundary requiring a string.
+- Do not use `value || undefined` for mutation fields where `''` means "clear this value". Trim or normalize at the form boundary, then preserve intentional empty strings.
+- Prefer nullable-tolerant render props for API-returned rows. Narrow only where a real value is required, such as mutation params, route hrefs, select values, query input, or required React keys.
+- Build required values in the same branch that proves them, using `flatMap`, a local loop, or an early return. Avoid truthiness guards, `filter(Boolean)`, `filter(item => item.id)`, and `!` after filters.
+- Use conditional spreads or explicit pushes for conditional array items instead of `undefined` placeholders followed by narrowing filters.
+- Empty collection fallbacks are for not-yet-loaded query data or genuinely nullable collections at the owning render boundary, not for hiding required API fields.
 
 ## Queries And Mutations
 
-- Keep `web/contract/*` as the single source of truth for API shape; follow existing domain/router patterns and the `{ params, query?, body? }` input shape.
-- Consume queries directly with `useQuery(consoleQuery.xxx.queryOptions(...))` or `useQuery(marketplaceQuery.xxx.queryOptions(...))`.
-- Avoid pass-through hooks and thin `web/service/use-*` wrappers that only rename `queryOptions()` or `mutationOptions()`. Extract a small `queryOptions` helper only when repeated call-site options justify it.
-- Keep feature hooks for real orchestration, workflow state, or shared domain behavior.
-- For TanStack cache data, use generated or query-derived types; do not create local wrappers for `getQueryData` or `getQueriesData`.
-- For generated oRPC `queryOptions()` / `infiniteOptions()`, do not pass `skipToken` as `input`; keep a valid placeholder input shape and use `enabled` to gate missing required params because the OpenAPI codec encodes input eagerly.
-- Consume mutations directly with `useMutation(consoleQuery.xxx.mutationOptions(...))` or `useMutation(marketplaceQuery.xxx.mutationOptions(...))`; use oRPC clients as `mutationFn` only for custom flows.
-- Put shared cache behavior in `createTanstackQueryUtils(...experimental_defaults...)`; components may add UI feedback callbacks, but should not own shared invalidation rules.
+- 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(...))`.
+- 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(...)`.
+- 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.
 - Do not use deprecated `useInvalid` or `useReset`.
 - Prefer `mutate(...)`; use `mutateAsync(...)` only when Promise semantics are required, and wrap awaited calls in `try/catch`.
 
-## Component Boundaries
+## Boundaries And Overlays
 
-- Use the first level below a page or tab to organize independent page sections when it adds real 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. Do not expand a child component's responsibility just because its data is useful nearby; keep adjacent UI as a sibling owner or introduce a correctly named broader owner.
-- Split deeper components by the data and state each layer actually needs. Each component should access only necessary data, and ownership should stay at the lowest consumer.
+- 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 that owns the trigger, open state, and hidden content when it would obscure the parent flow.
-- Preserve composability by separating behavior ownership from layout ownership. A dropdown action may own its trigger, open state, and menu content; the caller owns placement such as slots, offsets, and alignment.
-- Avoid unnecessary DOM hierarchy. Do not add wrapper elements unless they provide layout, semantics, accessibility, state ownership, or integration with a library API; prefer fragments or styling an existing element when possible.
-- Avoid shallow wrappers, hook-to-props adapter components, layout-only render-prop wrappers, and prop renaming unless the wrapper adds validation, orchestration, error handling, state ownership, or a real semantic boundary. If a component only calls a hook and forwards every returned field to one child, move the hook into that child or make the wrapper own a real surface.
+- 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.
+- Preserve composability by separating behavior ownership from placement ownership: an action can own trigger/open/menu content while the caller owns slots, offsets, and alignment.
+- When a dialog, dropdown, or popover accepts controlled `open`, mount it unconditionally unless unmounting is required for performance or reset semantics. Use keyed scope or local state reset instead of `{open && <Surface />}` wrappers.
+- When opening a dialog from a menu item, keep the menu and dialog as sibling surfaces. Let the menu command open the dialog, and mount the dialog outside menu popup content.
+- For dialogs and alert dialogs, keep the root responsible for `open` wiring and put query/mutation hooks inside the content component when work should mount only after the overlay opens.
+- Prefer uncontrolled overlay roots when the library can own open state. Use `onOpenChange` for side effects and CSS/data selectors for open-state styling.
+- Avoid wrapper DOM unless it provides layout, semantics, accessibility, state ownership, or library integration. Avoid shallow wrappers, hook-to-props adapters, layout-only render props, children pass-through wrappers, and prop renaming unless they add real behavior or a real boundary.
 
-## You Might Not Need An Effect
-
-- Use Effects only to synchronize with external systems such as browser APIs, non-React widgets, subscriptions, timers, analytics that must run because the component was shown, or imperative DOM integration.
-- Do not use Effects to transform props or state for rendering. Calculate derived values during render, and use `useMemo` only when the calculation is actually expensive.
-- Do not use Effects to handle user actions. Put action-specific logic in the event handler where the cause is known.
-- Do not use Effects to copy one state value into another state value representing the same concept. Pick one source of truth and derive the rest during render.
-- Do not reset or adjust state from props with an Effect. Prefer a `key` reset, storing a stable ID and deriving the selected object, or guarded same-component render-time adjustment when truly necessary.
-- Prefer framework data APIs or TanStack Query for data fetching instead of writing request Effects in components.
-- If an Effect still seems necessary, first name the external system it synchronizes with. If there is no external system, remove the Effect and restructure the state or event flow.
-
-## Navigation And Performance
+## Effects, Navigation, And Performance
 
+- Use Effects only to synchronize with external systems. Do not use Effects to transform props/state for rendering, handle user actions, copy state, reset state from props, or fetch data.
+- For forms initialized from query data, prefer keyed remounts or surface-entry atom hydration over Effects that copy query data into form state.
+- Prefer framework data APIs or TanStack Query for data fetching.
 - Prefer `Link` for normal navigation. Use router APIs only for command-flow side effects such as mutation success, guarded redirects, or form submission.
-- Before reaching for `memo`, first try moving changing state down to the smallest component that actually uses it so unrelated sibling trees stay untouched.
-- If changing state must wrap other content, lift the unchanged content up and pass it as `children` so the stateful wrapper can update without React visiting that subtree.
+- Before using `memo`, move changing state down to the smallest component that uses it. If state must wrap stable content, lift the stable content up and pass it as `children`.
 - Avoid `memo`, `useMemo`, and `useCallback` unless there is a clear performance reason.

.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

@@ -8,8 +8,6 @@ on:
       - "build/**"
       - "release/e-*"
       - "hotfix/**"
-      - "feat/hitl-backend"
-      - "feat/rbac"
     tags:
       - "*"
 
@@ -23,6 +21,7 @@ env:
   DIFY_WEB_IMAGE_NAME: ${{ vars.DIFY_WEB_IMAGE_NAME || 'langgenius/dify-web' }}
   DIFY_API_IMAGE_NAME: ${{ vars.DIFY_API_IMAGE_NAME || 'langgenius/dify-api' }}
   DIFY_AGENT_IMAGE_NAME: ${{ vars.DIFY_AGENT_IMAGE_NAME || 'langgenius/dify-agent-backend' }}
+  DIFY_AGENT_LOCAL_SANDBOX_IMAGE_NAME: ${{ vars.DIFY_AGENT_LOCAL_SANDBOX_IMAGE_NAME || 'langgenius/dify-agent-local-sandbox' }}
 
 jobs:
   build:
@@ -76,6 +75,20 @@ jobs:
             file: "dify-agent/Dockerfile"
             platform: linux/arm64
             runs_on: depot-ubuntu-24.04-4
+          - service_name: "build-agent-local-sandbox-amd64"
+            image_name_env: "DIFY_AGENT_LOCAL_SANDBOX_IMAGE_NAME"
+            artifact_context: "local-sandbox"
+            build_context: "{{defaultContext}}:dify-agent"
+            file: "docker/local-sandbox/Dockerfile"
+            platform: linux/amd64
+            runs_on: depot-ubuntu-24.04-4
+          - service_name: "build-agent-local-sandbox-arm64"
+            image_name_env: "DIFY_AGENT_LOCAL_SANDBOX_IMAGE_NAME"
+            artifact_context: "local-sandbox"
+            build_context: "{{defaultContext}}:dify-agent"
+            file: "docker/local-sandbox/Dockerfile"
+            platform: linux/arm64
+            runs_on: depot-ubuntu-24.04-4
 
     steps:
       - name: Prepare
@@ -141,6 +154,9 @@ jobs:
           - service_name: "validate-agent-amd64"
             build_context: "{{defaultContext}}"
             file: "dify-agent/Dockerfile"
+          - service_name: "validate-agent-local-sandbox-amd64"
+            build_context: "{{defaultContext}}:dify-agent"
+            file: "docker/local-sandbox/Dockerfile"
     steps:
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@d7f5e7f509e45cec5c76c4d5afdd7de93d0b3df5 # v4.1.0
@@ -155,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:
@@ -169,6 +185,9 @@ jobs:
           - service_name: "merge-agent-images"
             image_name_env: "DIFY_AGENT_IMAGE_NAME"
             context: "agent"
+          - service_name: "merge-agent-local-sandbox-images"
+            image_name_env: "DIFY_AGENT_LOCAL_SANDBOX_IMAGE_NAME"
+            context: "local-sandbox"
     steps:
       - name: Download digests
         uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1

.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-hitl.yml

@@ -1,25 +0,0 @@
-name: Deploy HITL
-
-on:
-  workflow_run:
-    workflows: ["Build and Push API & Web"]
-    branches:
-      - "build/feat/hitl"
-    types:
-      - completed
-
-jobs:
-  deploy:
-    runs-on: depot-ubuntu-24.04
-    if: |
-      github.event.workflow_run.conclusion == 'success' &&
-      github.event.workflow_run.head_branch == 'build/feat/hitl'
-    steps:
-      - name: Deploy to server
-        uses: appleboy/ssh-action@0ff4204d59e8e51228ff73bce53f80d53301dee2 # v1.2.5
-        with:
-          host: ${{ secrets.HITL_SSH_HOST }}
-          username: ${{ secrets.SSH_USER }}
-          key: ${{ secrets.SSH_PRIVATE_KEY }}
-          script: |
-            ${{ vars.SSH_SCRIPT || secrets.SSH_SCRIPT }}

.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
@@ -61,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
@@ -105,6 +105,14 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         run: vp run knip
 
+      - name: Web dead code check production
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: vp run knip:production
+
+      - name: Web production unused declarations check
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: vp run knip:production-unused-check
+
   ts-common-style:
     name: TS Common
     runs-on: depot-ubuntu-24.04
@@ -167,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:

README.md

@@ -116,7 +116,7 @@ All of Dify's offerings come with corresponding APIs, so you could effortlessly
 ## Using Dify
 
 - **Cloud <br/>**
-  We host a [Dify Cloud](https://dify.ai) service for anyone to try with zero setup. It provides all the capabilities of the self-deployed version, and includes 200 free GPT-4 calls in the sandbox plan.
+  We host a [Dify Cloud](https://dify.ai) service for anyone to try with zero setup. It provides all the capabilities of the self-deployed version, and includes 200 free GPT-4 calls in the sandbox plan. If you run into issues with Dify Cloud, [contact our Cloud support team](mailto:cloud@dify.ai?subject=%5BGitHub%5DDify%20Cloud%20Support).
 
 - **Self-hosting Dify Community Edition<br/>**
   Quickly get Dify running in your environment with this [starter guide](#quick-start).

api/.env.example

@@ -36,6 +36,9 @@ FILES_ACCESS_TIMEOUT=300
 # Collaboration mode toggle
 ENABLE_COLLABORATION_MODE=true
 
+# Learn app feature toggle
+ENABLE_LEARN_APP=true
+
 # Access token expiration time in minutes
 ACCESS_TOKEN_EXPIRE_MINUTES=60
 
@@ -768,7 +771,6 @@ EVENT_BUS_REDIS_CHANNEL_TYPE=pubsub
 # Whether to use Redis cluster mode while use redis as event bus.
 #  It's highly recommended to enable this for large deployments.
 EVENT_BUS_REDIS_USE_CLUSTERS=false
-EVENT_BUS_LISTENER_JOIN_TIMEOUT_MS=2000
 
 # Whether to Enable human input timeout check task
 ENABLE_HUMAN_INPUT_TIMEOUT_TASK=true

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,6 +78,13 @@ def _filter_snapshot_to_specs(
     return CompositorSessionSnapshot(schema_version=snapshot.schema_version, layers=filtered_layers)
 
 
+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):
     """API-side model/plugin selection before it is converted to Dify Agent layers."""
 
@@ -263,6 +270,7 @@ class AgentBackendRunRequestBuilder:
                 RunLayerSpec(
                     name=DIFY_DRIVE_LAYER_ID,
                     type=DIFY_DRIVE_LAYER_TYPE_ID,
+                    deps={"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID},
                     metadata=run_input.metadata,
                     config=run_input.drive_config,
                 )
@@ -329,14 +337,15 @@ 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 (back proxy);
+            # 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={"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_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(),
                 )
@@ -460,6 +469,7 @@ class AgentBackendRunRequestBuilder:
                 RunLayerSpec(
                     name=DIFY_DRIVE_LAYER_ID,
                     type=DIFY_DRIVE_LAYER_TYPE_ID,
+                    deps={"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID},
                     metadata=run_input.metadata,
                     config=run_input.drive_config,
                 )
@@ -528,14 +538,15 @@ 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 (back proxy);
+            # 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={"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_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(),
                 )

api/commands/__init__.py

@@ -22,9 +22,10 @@ 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,
     clean_expired_messages,
     clean_workflow_runs,
     cleanup_orphaned_draft_variables,
@@ -51,6 +52,7 @@ from .vector import (
 __all__ = [
     "add_qdrant_index",
     "archive_workflow_runs",
+    "archive_workflow_runs_plan",
     "backfill_plugin_auto_upgrade",
     "clean_expired_messages",
     "clean_workflow_runs",
@@ -74,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/account.py

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

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

@@ -12,10 +12,160 @@ from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpi
 from services.retention.conversation.messages_clean_policy import create_message_clean_policy
 from services.retention.conversation.messages_clean_service import MessagesCleanService
 from services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs import WorkflowRunCleanup
+from services.retention.workflow_run.tenant_prefix import tenant_prefix_condition
 from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 
 logger = logging.getLogger(__name__)
 
+_HEX_PREFIXES = tuple("0123456789abcdef")
+
+
+class WorkflowRunArchivePlanRow(TypedDict):
+    tenant_prefix: str
+    total_tenants: int
+    workflow_runs: int
+    workflow_node_executions: int
+    paid_tenants: int
+    unpaid_tenants: int
+
+
+class WorkflowRunArchiveTenantPlan(TypedDict):
+    archive_tenant_ids: list[str] | None
+    paid_tenant_ids: list[str]
+    unpaid_tenant_ids: list[str]
+
+
+def _parse_tenant_prefixes(prefixes: str | None) -> list[str]:
+    if not prefixes:
+        return []
+
+    parsed = []
+    for raw_prefix in prefixes.split(","):
+        prefix = raw_prefix.strip().lower()
+        if not prefix:
+            continue
+        if len(prefix) != 1 or prefix not in _HEX_PREFIXES:
+            raise click.UsageError("--tenant-prefixes must be a comma-separated list of hex digits, e.g. 0,1,a,f.")
+        parsed.append(prefix)
+    return sorted(set(parsed))
+
+
+def _get_archive_candidate_tenant_ids_by_prefix(
+    prefix: str,
+    *,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime,
+) -> list[str]:
+    from graphon.enums import WorkflowExecutionStatus
+    from models.workflow import WorkflowRun
+    from services.retention.workflow_run.archive_paid_plan_workflow_run import WorkflowRunArchiver
+
+    conditions = [
+        WorkflowRun.created_at < end_before,
+        WorkflowRun.status.in_(WorkflowExecutionStatus.ended_values()),
+        WorkflowRun.type.in_(WorkflowRunArchiver.ARCHIVED_TYPE),
+        tenant_prefix_condition(WorkflowRun.tenant_id, prefix),
+    ]
+    if start_from is not None:
+        conditions.append(WorkflowRun.created_at >= start_from)
+
+    tenant_ids = db.session.scalars(
+        sa.select(WorkflowRun.tenant_id).where(*conditions).distinct().order_by(WorkflowRun.tenant_id)
+    ).all()
+    return list(tenant_ids)
+
+
+def _filter_paid_workflow_archive_tenant_ids(tenant_ids: list[str]) -> tuple[list[str], list[str]]:
+    from configs import dify_config
+    from enums.cloud_plan import CloudPlan
+    from services.billing_service import BillingService
+
+    tenant_ids = sorted(set(tenant_ids))
+    if not tenant_ids:
+        return [], []
+    if not dify_config.BILLING_ENABLED:
+        return tenant_ids, []
+
+    plans = BillingService.get_plan_bulk_with_cache(tenant_ids)
+    paid_tenant_ids = [
+        tenant_id
+        for tenant_id in tenant_ids
+        if plans.get(tenant_id) and plans[tenant_id].get("plan") in (CloudPlan.PROFESSIONAL, CloudPlan.TEAM)
+    ]
+    unpaid_tenant_ids = sorted(set(tenant_ids) - set(paid_tenant_ids))
+    return paid_tenant_ids, unpaid_tenant_ids
+
+
+def _resolve_archive_tenant_ids_from_plan(
+    *,
+    tenant_ids: str | None,
+    tenant_prefixes: list[str],
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime,
+) -> WorkflowRunArchiveTenantPlan:
+    """
+    Resolve the archive tenant scope once before scanning workflow_runs.
+
+    Prefix rollout should use the tenant list collected by the same planning path, then archive by
+    tenant_id IN (...). Scanning workflow_runs with a tenant prefix range in every archive run is too expensive on
+    the large production table this command is meant to shrink.
+    """
+    if tenant_ids:
+        requested_tenant_ids = [tid.strip() for tid in tenant_ids.split(",") if tid.strip()]
+    elif tenant_prefixes:
+        requested_tenant_ids = []
+        for prefix in tenant_prefixes:
+            requested_tenant_ids.extend(
+                _get_archive_candidate_tenant_ids_by_prefix(
+                    prefix,
+                    start_from=start_from,
+                    end_before=end_before,
+                )
+            )
+    else:
+        return WorkflowRunArchiveTenantPlan(
+            archive_tenant_ids=None,
+            paid_tenant_ids=[],
+            unpaid_tenant_ids=[],
+        )
+
+    paid_tenant_ids, unpaid_tenant_ids = _filter_paid_workflow_archive_tenant_ids(requested_tenant_ids)
+    return WorkflowRunArchiveTenantPlan(
+        archive_tenant_ids=paid_tenant_ids,
+        paid_tenant_ids=paid_tenant_ids,
+        unpaid_tenant_ids=unpaid_tenant_ids,
+    )
+
+
+def _resolve_archive_time_range(
+    *,
+    before_days: int,
+    from_days_ago: int | None,
+    to_days_ago: int | None,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime | None,
+) -> tuple[int, datetime.datetime | None, datetime.datetime | None]:
+    if (start_from is None) ^ (end_before is None):
+        raise click.UsageError("--start-from and --end-before must be provided together.")
+
+    if (from_days_ago is None) ^ (to_days_ago is None):
+        raise click.UsageError("--from-days-ago and --to-days-ago must be provided together.")
+
+    if from_days_ago is not None and to_days_ago is not None:
+        if start_from or end_before:
+            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()
+        start_from = now - datetime.timedelta(days=from_days_ago)
+        end_before = now - datetime.timedelta(days=to_days_ago)
+        before_days = 0
+
+    if start_from and end_before and start_from >= end_before:
+        raise click.UsageError("--start-from must be earlier than --end-before.")
+
+    return before_days, start_from, end_before
+
 
 @click.command("clear-free-plan-tenant-expired-logs", help="Clear free plan tenant expired logs.")
 @click.option("--days", prompt=True, help="The days to clear free plan tenant expired logs.", default=30)
@@ -139,11 +289,143 @@ def clean_workflow_runs(
     )
 
 
+@click.command(
+    "archive-workflow-runs-plan",
+    help="Plan workflow run archive rollout by tenant ID first hex digit.",
+)
+@click.option("--before-days", default=90, show_default=True, help="Plan runs older than N days.")
+@click.option(
+    "--from-days-ago",
+    default=None,
+    type=click.IntRange(min=0),
+    help="Lower bound in days ago (older). Must be paired with --to-days-ago.",
+)
+@click.option(
+    "--to-days-ago",
+    default=None,
+    type=click.IntRange(min=0),
+    help="Upper bound in days ago (newer). Must be paired with --from-days-ago.",
+)
+@click.option(
+    "--start-from",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    default=None,
+    help="Plan runs created at or after this timestamp (UTC if no timezone).",
+)
+@click.option(
+    "--end-before",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    default=None,
+    help="Plan runs created before this timestamp (UTC if no timezone).",
+)
+@click.option(
+    "--include-archived",
+    is_flag=True,
+    help="Compatibility no-op for V2 bundle archive; plan counts source rows in the requested window.",
+)
+def archive_workflow_runs_plan(
+    before_days: int,
+    from_days_ago: int | None,
+    to_days_ago: int | None,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime | None,
+    include_archived: bool,
+):
+    """
+    Print the 16 tenant-prefix rollout rows used to choose archive execution order.
+
+    Counts use the same workflow run eligibility as archive-workflow-runs: ended runs,
+    supported workflow types, and the requested created_at window. V2 bundle archive
+    does not maintain per-run archive logs, so this plan reports source-table volume.
+    """
+    from graphon.enums import WorkflowExecutionStatus
+    from models.workflow import WorkflowNodeExecutionModel, WorkflowRun
+    from services.retention.workflow_run.archive_paid_plan_workflow_run import WorkflowRunArchiver
+
+    before_days, start_from, end_before = _resolve_archive_time_range(
+        before_days=before_days,
+        from_days_ago=from_days_ago,
+        to_days_ago=to_days_ago,
+        start_from=start_from,
+        end_before=end_before,
+    )
+    plan_end_before = end_before or datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=before_days)
+    if include_archived:
+        click.echo(click.style("--include-archived is a no-op for V2 bundle archive plans.", fg="yellow"))
+
+    rows: list[WorkflowRunArchivePlanRow] = []
+    for prefix in _HEX_PREFIXES:
+        tenant_ids = _get_archive_candidate_tenant_ids_by_prefix(
+            prefix,
+            start_from=start_from,
+            end_before=plan_end_before,
+        )
+        total_tenants = len(tenant_ids)
+        paid_tenant_ids, unpaid_tenant_ids = _filter_paid_workflow_archive_tenant_ids(tenant_ids)
+
+        run_conditions = [
+            WorkflowRun.created_at < plan_end_before,
+            WorkflowRun.status.in_(WorkflowExecutionStatus.ended_values()),
+            WorkflowRun.type.in_(WorkflowRunArchiver.ARCHIVED_TYPE),
+            tenant_prefix_condition(WorkflowRun.tenant_id, prefix),
+        ]
+        if start_from is not None:
+            run_conditions.append(WorkflowRun.created_at >= start_from)
+        workflow_runs = (
+            db.session.scalar(sa.select(sa.func.count()).select_from(WorkflowRun).where(*run_conditions)) or 0
+        )
+        candidate_runs = sa.select(WorkflowRun.id).where(*run_conditions).subquery()
+        workflow_node_executions = (
+            db.session.scalar(
+                sa.select(sa.func.count())
+                .select_from(WorkflowNodeExecutionModel)
+                .join(candidate_runs, WorkflowNodeExecutionModel.workflow_run_id == candidate_runs.c.id)
+            )
+            or 0
+        )
+
+        rows.append(
+            WorkflowRunArchivePlanRow(
+                tenant_prefix=prefix,
+                total_tenants=total_tenants,
+                workflow_runs=workflow_runs,
+                workflow_node_executions=workflow_node_executions,
+                paid_tenants=len(paid_tenant_ids),
+                unpaid_tenants=len(unpaid_tenant_ids),
+            )
+        )
+
+    click.echo(
+        click.style(
+            f"Workflow archive plan for runs before {plan_end_before.isoformat()}"
+            f"{f' and at/after {start_from.isoformat()}' if start_from else ''}.",
+            fg="white",
+        )
+    )
+    click.echo("tenant_prefix,total_tenants,workflow_runs,workflow_node_executions,paid_tenants,unpaid_tenants")
+    for row in rows:
+        click.echo(
+            f"{row['tenant_prefix']},{row['total_tenants']},{row['workflow_runs']},"
+            f"{row['workflow_node_executions']},{row['paid_tenants']},{row['unpaid_tenants']}"
+        )
+
+    ordered_rows = sorted(
+        rows,
+        key=lambda row: (row["workflow_runs"] + row["workflow_node_executions"], row["tenant_prefix"]),
+    )
+    click.echo("suggested_execution_order=" + ",".join(row["tenant_prefix"] for row in ordered_rows))
+
+
 @click.command(
     "archive-workflow-runs",
     help="Archive workflow runs for paid plan tenants to S3-compatible storage.",
 )
 @click.option("--tenant-ids", default=None, help="Optional comma-separated tenant IDs for grayscale rollout.")
+@click.option(
+    "--tenant-prefixes",
+    default=None,
+    help="Optional comma-separated tenant ID first hex digits for rollout waves, e.g. 0,1,a,f.",
+)
 @click.option("--before-days", default=90, show_default=True, help="Archive runs older than N days.")
 @click.option(
     "--from-days-ago",
@@ -169,13 +451,36 @@ def clean_workflow_runs(
     default=None,
     help="Archive runs created before this timestamp (UTC if no timezone).",
 )
-@click.option("--batch-size", default=100, show_default=True, help="Batch size for processing.")
-@click.option("--workers", default=1, show_default=True, type=int, help="Concurrent workflow runs to archive.")
+@click.option("--batch-size", default=100, show_default=True, help="Maximum workflow runs per archive bundle.")
+@click.option(
+    "--workers",
+    default=1,
+    show_default=True,
+    type=int,
+    help="Reserved; bundle archive currently runs serially.",
+)
+@click.option(
+    "--run-shard-index",
+    default=None,
+    type=click.IntRange(min=0),
+    help="Zero-based workflow run shard index for parallel cron jobs. Must be paired with --run-shard-total.",
+)
+@click.option(
+    "--run-shard-total",
+    default=None,
+    type=click.IntRange(min=1, max=16),
+    help="Total workflow run shard count for parallel cron jobs. Must be paired with --run-shard-index.",
+)
 @click.option("--limit", default=None, type=int, help="Maximum number of runs to archive.")
 @click.option("--dry-run", is_flag=True, help="Preview without archiving.")
-@click.option("--delete-after-archive", is_flag=True, help="Delete runs and related data after archiving.")
+@click.option(
+    "--delete-after-archive",
+    is_flag=True,
+    help="Not supported by bundle archive; use a separate bundle delete workflow after validation.",
+)
 def archive_workflow_runs(
     tenant_ids: str | None,
+    tenant_prefixes: str | None,
     before_days: int,
     from_days_ago: int | None,
     to_days_ago: int | None,
@@ -183,6 +488,8 @@ def archive_workflow_runs(
     end_before: datetime.datetime | None,
     batch_size: int,
     workers: int,
+    run_shard_index: int | None,
+    run_shard_total: int | None,
     limit: int | None,
     dry_run: bool,
     delete_after_archive: bool,
@@ -190,14 +497,19 @@ def archive_workflow_runs(
     """
     Archive workflow runs for paid plan tenants older than the specified days.
 
-    This command archives the following tables to storage:
+    This command writes V2 tenant/month/shard archive bundles. Each bundle contains Parquet snapshots from:
+    - workflow_runs
+    - workflow_app_logs
     - workflow_node_executions
     - workflow_node_execution_offload
     - workflow_pauses
     - workflow_pause_reasons
     - workflow_trigger_logs
 
-    The workflow_runs and workflow_app_logs tables are preserved for UI listing.
+    Source database rows are always preserved by archive. Deletion must be handled by
+    a separate bundle-level delete workflow after manifest, checksum, row-count, and
+    restore-sampling validation. In --dry-run mode, no storage or database writes
+    happen; the command estimates per-table Parquet bytes and object size instead.
     """
     from services.retention.workflow_run.archive_paid_plan_workflow_run import WorkflowRunArchiver
 
@@ -209,32 +521,58 @@ def archive_workflow_runs(
         )
     )
 
-    if (start_from is None) ^ (end_before is None):
-        click.echo(click.style("start-from and end-before must be provided together.", fg="red"))
-        return
-
-    if (from_days_ago is None) ^ (to_days_ago is None):
-        click.echo(click.style("from-days-ago and to-days-ago must be provided together.", fg="red"))
-        return
-
-    if from_days_ago is not None and to_days_ago is not None:
-        if start_from or end_before:
-            click.echo(click.style("Choose either day offsets or explicit dates, not both.", fg="red"))
-            return
-        if from_days_ago <= to_days_ago:
-            click.echo(click.style("from-days-ago must be greater than to-days-ago.", fg="red"))
-            return
-        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 and end_before and start_from >= end_before:
-        click.echo(click.style("start-from must be earlier than end-before.", fg="red"))
+    try:
+        before_days, start_from, end_before = _resolve_archive_time_range(
+            before_days=before_days,
+            from_days_ago=from_days_ago,
+            to_days_ago=to_days_ago,
+            start_from=start_from,
+            end_before=end_before,
+        )
+        parsed_tenant_prefixes = _parse_tenant_prefixes(tenant_prefixes)
+    except click.UsageError as e:
+        click.echo(click.style(e.message, fg="red"))
         return
+    plan_end_before = end_before or datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=before_days)
     if workers < 1:
         click.echo(click.style("workers must be at least 1.", fg="red"))
         return
+    if (run_shard_index is None) ^ (run_shard_total is None):
+        click.echo(click.style("run-shard-index and run-shard-total must be provided together.", fg="red"))
+        return
+    if run_shard_index is not None and run_shard_total is not None and run_shard_index >= run_shard_total:
+        click.echo(click.style("run-shard-index must be less than run-shard-total.", fg="red"))
+        return
+    if delete_after_archive:
+        click.echo(click.style("delete-after-archive is not supported by bundle archive.", fg="red"))
+        return
+
+    try:
+        tenant_plan = _resolve_archive_tenant_ids_from_plan(
+            tenant_ids=tenant_ids,
+            tenant_prefixes=parsed_tenant_prefixes,
+            start_from=start_from,
+            end_before=plan_end_before,
+        )
+    except Exception:
+        logger.exception("Failed to resolve workflow archive tenant plan")
+        click.echo(click.style("Failed to resolve workflow archive tenant plan.", fg="red"))
+        return
+
+    planned_tenant_ids = tenant_plan["archive_tenant_ids"]
+    planned_paid_tenant_ids = tenant_plan["paid_tenant_ids"] if planned_tenant_ids is not None else None
+    paid_tenants = len(tenant_plan["paid_tenant_ids"])
+    unpaid_tenants = len(tenant_plan["unpaid_tenant_ids"])
+    if planned_tenant_ids is not None:
+        click.echo(
+            click.style(
+                f"Resolved archive tenant plan: paid_tenants={paid_tenants}, unpaid_tenants={unpaid_tenants}.",
+                fg="white",
+            )
+        )
+        if not planned_tenant_ids:
+            click.echo(click.style("No paid tenants matched the archive plan; nothing to archive.", fg="yellow"))
+            return
 
     archiver = WorkflowRunArchiver(
         days=before_days,
@@ -242,7 +580,11 @@ def archive_workflow_runs(
         start_from=start_from,
         end_before=end_before,
         workers=workers,
-        tenant_ids=[tid.strip() for tid in tenant_ids.split(",")] if tenant_ids else None,
+        tenant_ids=planned_tenant_ids,
+        tenant_prefixes=parsed_tenant_prefixes,
+        paid_tenant_ids=planned_paid_tenant_ids,
+        run_shard_index=run_shard_index,
+        run_shard_total=run_shard_total,
         limit=limit,
         dry_run=dry_run,
         delete_after_archive=delete_after_archive,
@@ -252,7 +594,9 @@ def archive_workflow_runs(
         click.style(
             f"Summary: processed={summary.total_runs_processed}, archived={summary.runs_archived}, "
             f"skipped={summary.runs_skipped}, failed={summary.runs_failed}, "
-            f"time={summary.total_elapsed_time:.2f}s",
+            f"bundles_archived={summary.bundles_archived}, bundles_skipped={summary.bundles_skipped}, "
+            f"bundles_failed={summary.bundles_failed}, "
+            f"object_size_bytes={summary.total_object_size_bytes}, time={summary.total_elapsed_time:.2f}s",
             fg="cyan",
         )
     )
@@ -268,6 +612,52 @@ def archive_workflow_runs(
     )
 
 
+def _echo_bundle_archive_operation_summary(summary) -> None:
+    status = "completed successfully" if summary.bundles_failed == 0 else "completed with failures"
+    fg = "green" if summary.bundles_failed == 0 else "red"
+    click.echo(
+        click.style(
+            f"{summary.operation} {status}. "
+            f"bundles_success={summary.bundles_succeeded} bundles_failed={summary.bundles_failed} "
+            f"runs={summary.runs_processed} rows={summary.rows_processed} "
+            f"archive_bytes={summary.archive_bytes} duration={summary.elapsed_time:.2f}s "
+            f"validation_time={summary.validation_time:.2f}s "
+            f"runs_per_second={summary.runs_per_second:.2f} rows_per_second={summary.rows_per_second:.2f} "
+            f"bytes_per_second={summary.bytes_per_second:.2f}",
+            fg=fg,
+        )
+    )
+    click.echo(click.style("table,row_count", fg="white"))
+    for table_name in [
+        "workflow_runs",
+        "workflow_app_logs",
+        "workflow_node_executions",
+        "workflow_node_execution_offload",
+        "workflow_pauses",
+        "workflow_pause_reasons",
+        "workflow_trigger_logs",
+    ]:
+        click.echo(f"{table_name},{summary.table_counts.get(table_name, 0)}")
+    for result in summary.results:
+        if result.success:
+            click.echo(
+                click.style(
+                    f"  bundle={result.bundle_id} tenant={result.tenant_id} runs={result.run_count} "
+                    f"rows={result.row_count} archive_bytes={result.archive_bytes} "
+                    f"time={result.elapsed_time:.2f}s validation={result.validation_time:.2f}s",
+                    fg="white",
+                )
+            )
+        else:
+            click.echo(
+                click.style(
+                    f"  failed bundle={result.bundle_id} tenant={result.tenant_id} "
+                    f"object_prefix={result.object_prefix} error={result.error}",
+                    fg="red",
+                )
+            )
+
+
 @click.command(
     "restore-workflow-runs",
     help="Restore archived workflow runs from S3-compatible storage.",
@@ -290,8 +680,8 @@ def archive_workflow_runs(
     default=None,
     help="Optional upper bound (exclusive) for created_at; must be paired with --start-from.",
 )
-@click.option("--workers", default=1, show_default=True, type=int, help="Concurrent workflow runs to restore.")
-@click.option("--limit", type=int, default=100, show_default=True, help="Maximum number of runs to restore.")
+@click.option("--workers", default=1, show_default=True, type=int, help="V1 --run-id compatibility only.")
+@click.option("--limit", type=int, default=100, show_default=True, help="Maximum number of V2 bundles to restore.")
 @click.option("--dry-run", is_flag=True, help="Preview without restoring.")
 def restore_workflow_runs(
     tenant_ids: str | None,
@@ -303,15 +693,18 @@ def restore_workflow_runs(
     dry_run: bool,
 ):
     """
-    Restore an archived workflow run from storage to the database.
+    Restore archived workflow runs from storage to the database.
 
-    This restores the following tables:
+    Batch restore uses V2 bundle metadata and validates archive objects before writing source rows. This restores:
+    - workflow_runs
+    - workflow_app_logs
     - workflow_node_executions
     - workflow_node_execution_offload
     - workflow_pauses
     - workflow_pause_reasons
     - workflow_trigger_logs
     """
+    from services.retention.workflow_run.bundle_archive_maintenance import WorkflowRunBundleArchiveMaintenance
     from services.retention.workflow_run.restore_archived_workflow_run import WorkflowRunRestore
 
     parsed_tenant_ids = None
@@ -335,39 +728,46 @@ def restore_workflow_runs(
         )
     )
 
-    restorer = WorkflowRunRestore(dry_run=dry_run, workers=workers)
     if run_id:
+        restorer = WorkflowRunRestore(dry_run=dry_run, workers=workers)
         results = [restorer.restore_by_run_id(run_id)]
-    else:
-        assert start_from is not None
-        assert end_before is not None
-        results = restorer.restore_batch(
-            parsed_tenant_ids,
-            start_date=start_from,
-            end_date=end_before,
-            limit=limit,
-        )
+        end_time = datetime.datetime.now(datetime.UTC)
+        elapsed = end_time - start_time
 
-    end_time = datetime.datetime.now(datetime.UTC)
-    elapsed = end_time - start_time
+        successes = sum(1 for result in results if result.success)
+        failures = len(results) - successes
 
-    successes = sum(1 for result in results if result.success)
-    failures = len(results) - successes
-
-    if failures == 0:
-        click.echo(
-            click.style(
-                f"Restore completed successfully. success={successes} duration={elapsed}",
-                fg="green",
+        if failures == 0:
+            click.echo(
+                click.style(
+                    f"Restore completed successfully. success={successes} duration={elapsed}",
+                    fg="green",
+                )
             )
-        )
-    else:
-        click.echo(
-            click.style(
-                f"Restore completed with failures. success={successes} failed={failures} duration={elapsed}",
-                fg="red",
+        else:
+            click.echo(
+                click.style(
+                    f"Restore completed with failures. success={successes} failed={failures} duration={elapsed}",
+                    fg="red",
+                )
             )
+        return
+
+    if workers != 1:
+        click.echo(
+            click.style("--workers is ignored for V2 bundle restore; bundles are processed serially.", fg="yellow")
         )
+    assert start_from is not None
+    assert end_before is not None
+    bundle_restorer = WorkflowRunBundleArchiveMaintenance(dry_run=dry_run, strict_content_validation=True)
+    summary = bundle_restorer.restore_batch(
+        tenant_ids=parsed_tenant_ids,
+        start_date=start_from,
+        end_date=end_before,
+        limit=limit,
+    )
+    _echo_bundle_archive_operation_summary(summary)
+    return
 
 
 @click.command(
@@ -392,8 +792,20 @@ def restore_workflow_runs(
     default=None,
     help="Optional upper bound (exclusive) for created_at; must be paired with --start-from.",
 )
-@click.option("--limit", type=int, default=100, show_default=True, help="Maximum number of runs to delete.")
+@click.option("--limit", type=int, default=100, show_default=True, help="Maximum number of V2 bundles to delete.")
 @click.option("--dry-run", is_flag=True, help="Preview without deleting.")
+@click.option(
+    "--skip-bad-archives",
+    is_flag=True,
+    help="Continue batch deletion when one archive object fails validation.",
+)
+@click.option(
+    "--restore-sample-interval",
+    type=int,
+    default=0,
+    show_default=True,
+    help="Run restore dry-run after every N successful deletes; 0 disables restore sampling.",
+)
 def delete_archived_workflow_runs(
     tenant_ids: str | None,
     run_id: str | None,
@@ -401,10 +813,16 @@ def delete_archived_workflow_runs(
     end_before: datetime.datetime | None,
     limit: int,
     dry_run: bool,
+    skip_bad_archives: bool,
+    restore_sample_interval: int,
 ):
     """
     Delete archived workflow runs from the database.
+
+    Batch delete uses V2 bundle metadata and validates object existence, manifest schema, object size, checksum, row
+    counts, and source/archive content checksums before deleting source rows. `--run-id` keeps the V1 per-run path.
     """
+    from services.retention.workflow_run.bundle_archive_maintenance import WorkflowRunBundleArchiveMaintenance
     from services.retention.workflow_run.delete_archived_workflow_run import ArchivedWorkflowRunDeletion
 
     parsed_tenant_ids = None
@@ -417,6 +835,8 @@ def delete_archived_workflow_runs(
         raise click.UsageError("--start-from and --end-before must be provided together.")
     if run_id is None and (start_from is None or end_before is None):
         raise click.UsageError("--start-from and --end-before are required for batch delete.")
+    if restore_sample_interval < 0:
+        raise click.BadParameter("restore-sample-interval must be >= 0")
 
     start_time = datetime.datetime.now(datetime.UTC)
     target_desc = f"workflow run {run_id}" if run_id else "workflow runs"
@@ -427,56 +847,85 @@ def delete_archived_workflow_runs(
         )
     )
 
-    deleter = ArchivedWorkflowRunDeletion(dry_run=dry_run)
     if run_id:
-        results = [deleter.delete_by_run_id(run_id)]
-    else:
-        assert start_from is not None
-        assert end_before is not None
-        results = deleter.delete_batch(
-            parsed_tenant_ids,
-            start_date=start_from,
-            end_date=end_before,
-            limit=limit,
+        deleter = ArchivedWorkflowRunDeletion(
+            dry_run=dry_run,
+            skip_bad_archives=skip_bad_archives,
+            restore_sample_interval=restore_sample_interval,
         )
+        results = [deleter.delete_by_run_id(run_id)]
+        for result in results:
+            if result.success:
+                click.echo(
+                    click.style(
+                        f"{'[DRY RUN] Would delete' if dry_run else 'Deleted'} "
+                        f"workflow run {result.run_id} (tenant={result.tenant_id}, "
+                        f"archive_key={result.archive_key}, counts={result.validated_counts})",
+                        fg="green",
+                    )
+                )
+                if result.restore_sampled:
+                    sample_status = "passed" if result.restore_sample_success else "failed"
+                    click.echo(
+                        click.style(
+                            f"  restore dry-run sample {sample_status} for workflow run {result.run_id}",
+                            fg="green" if result.restore_sample_success else "red",
+                        )
+                    )
+            else:
+                click.echo(
+                    click.style(
+                        f"Failed to delete workflow run {result.run_id}: {result.error}",
+                        fg="red",
+                    )
+                )
+                click.echo(
+                    click.style(
+                        "  runbook: pause this delete window, verify archive storage object and manifest/checksum, "
+                        "retry the same run after fixing storage or DB drift, or rerun with --skip-bad-archives "
+                        "to quarantine this run and continue the batch.",
+                        fg="yellow",
+                    )
+                )
 
-    for result in results:
-        if result.success:
+        end_time = datetime.datetime.now(datetime.UTC)
+        elapsed = end_time - start_time
+
+        successes = sum(1 for result in results if result.success)
+        failures = len(results) - successes
+
+        if failures == 0:
             click.echo(
                 click.style(
-                    f"{'[DRY RUN] Would delete' if dry_run else 'Deleted'} "
-                    f"workflow run {result.run_id} (tenant={result.tenant_id})",
+                    f"Delete completed successfully. success={successes} duration={elapsed}",
                     fg="green",
                 )
             )
         else:
             click.echo(
                 click.style(
-                    f"Failed to delete workflow run {result.run_id}: {result.error}",
+                    f"Delete completed with failures. success={successes} failed={failures} duration={elapsed}",
                     fg="red",
                 )
             )
+        return
 
-    end_time = datetime.datetime.now(datetime.UTC)
-    elapsed = end_time - start_time
-
-    successes = sum(1 for result in results if result.success)
-    failures = len(results) - successes
-
-    if failures == 0:
-        click.echo(
-            click.style(
-                f"Delete completed successfully. success={successes} duration={elapsed}",
-                fg="green",
-            )
-        )
-    else:
-        click.echo(
-            click.style(
-                f"Delete completed with failures. success={successes} failed={failures} duration={elapsed}",
-                fg="red",
-            )
-        )
+    if restore_sample_interval:
+        click.echo(click.style("--restore-sample-interval is ignored for V2 bundle delete.", fg="yellow"))
+    assert start_from is not None
+    assert end_before is not None
+    bundle_deleter = WorkflowRunBundleArchiveMaintenance(
+        dry_run=dry_run,
+        strict_content_validation=True,
+        stop_on_error=not skip_bad_archives,
+    )
+    summary = bundle_deleter.delete_batch(
+        tenant_ids=parsed_tenant_ids,
+        start_date=start_from,
+        end_date=end_before,
+        limit=limit,
+    )
+    _echo_bundle_archive_operation_summary(summary)
 
 
 def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:

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/feature/__init__.py

@@ -1073,6 +1073,12 @@ class MailConfig(BaseSettings):
         default=None,
     )
 
+
+class HomepageConfig(BaseSettings):
+    """
+    Configuration for homepage feature toggles exposed through system features.
+    """
+
     ENABLE_TRIAL_APP: bool = Field(
         description="Enable trial app",
         default=False,
@@ -1083,6 +1089,11 @@ class MailConfig(BaseSettings):
         default=False,
     )
 
+    ENABLE_LEARN_APP: bool = Field(
+        description="Enable Learn App",
+        default=True,
+    )
+
 
 class RagEtlConfig(BaseSettings):
     """
@@ -1489,6 +1500,7 @@ class FeatureConfig(
     EndpointConfig,
     FileAccessConfig,
     FileUploadConfig,
+    HomepageConfig,
     HttpConfig,
     InnerAPIConfig,
     IndexingConfig,

api/configs/middleware/cache/redis_pubsub_config.py

@@ -2,7 +2,6 @@ from typing import Literal, Protocol, cast
 from urllib.parse import quote_plus, urlunparse
 
 from pydantic import AliasChoices, Field
-from pydantic.types import NonNegativeInt
 from pydantic_settings import BaseSettings
 
 
@@ -71,24 +70,6 @@ class RedisPubSubConfig(BaseSettings):
         default=600,
     )
 
-    PUBSUB_LISTENER_JOIN_TIMEOUT_MS: NonNegativeInt = Field(
-        validation_alias=AliasChoices("EVENT_BUS_LISTENER_JOIN_TIMEOUT_MS", "PUBSUB_LISTENER_JOIN_TIMEOUT_MS"),
-        description=(
-            "Maximum time (milliseconds) that ``Subscription.close()`` waits for its listener thread to "
-            "finish before returning. Bounds the tail latency between a terminal event being delivered to "
-            "an SSE client and the response stream actually closing.\n\n"
-            "The listener thread blocks on a polling read (XREAD BLOCK for streams, get_message timeout "
-            "for pubsub/sharded) with a fixed 1s window, so close() naturally has to wait up to ~1s for "
-            "the thread to notice the subscription was closed. Setting this lower (e.g. 100) lets close() "
-            "return promptly while the daemon listener thread cleans itself up on the next poll "
-            "boundary - safe because the listener holds no critical state and exits within one poll "
-            "window. Setting it higher (e.g. 5000) gives the listener more grace before close() gives up "
-            "and logs a warning. Default 2000ms preserves the pre-change behaviour.\n\n"
-            "Also accepts ENV: EVENT_BUS_LISTENER_JOIN_TIMEOUT_MS."
-        ),
-        default=2000,
-    )
-
     def _build_default_pubsub_url(self) -> str:
         defaults = _redis_defaults(self)
         if not defaults.REDIS_HOST or not defaults.REDIS_PORT:

api/controllers/common/app_access.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from services.enterprise import rbac_service as enterprise_rbac_service
+
+if TYPE_CHECKING:
+    from services.app_service import AppListBaseParams
+    from services.enterprise.rbac_service import MyPermissionsResponse
+
+# Permission keys (dot-notation, from MyPermissionsResponse) that grant
+# list/preview access to an app. Keep this the single source of truth for both
+# the console and OpenAPI app-list endpoints.
+APP_LIST_PERMISSION_KEYS: frozenset[str] = frozenset({"app.preview", "app.acl.preview", "app.full_access"})
+
+# Workspace permission key that lets a caller see apps they maintain even when
+# those apps are not in their preview whitelist.
+_MANAGE_OWN_APPS_PERMISSION_KEY = "app.create_and_management"
+
+
+def has_app_list_permission(permission_keys: Sequence[str]) -> bool:
+    """Return True if any of ``permission_keys`` grants app list/preview access."""
+    return any(permission_key in APP_LIST_PERMISSION_KEYS for permission_key in permission_keys)
+
+
+@dataclass(frozen=True)
+class AppAccessFilter:
+    """Resolved RBAC visibility for app list/read endpoints.
+
+    ``accessible_app_ids`` of ``None`` means the caller can see every app in the
+    workspace (unrestricted). Otherwise it is the exact set of app ids the
+    caller may preview; combined with ``can_manage_own_apps`` it also covers
+    apps the caller maintains.
+    """
+
+    accessible_app_ids: set[str] | None
+    can_manage_own_apps: bool
+
+    @classmethod
+    def unrestricted(cls) -> AppAccessFilter:
+        """Filter that imposes no restriction (RBAC disabled / not applicable)."""
+        return cls(accessible_app_ids=None, can_manage_own_apps=False)
+
+    def is_app_accessible(self, app_id: str, maintainer: str | None, account_id: str) -> bool:
+        """Whether a single app is visible to the caller under this filter.
+
+        Mirrors the service-layer query gate: an app is visible when the filter
+        is unrestricted, the app id is whitelisted, or the caller maintains it
+        and holds ``app.create_and_management``.
+        """
+        if self.accessible_app_ids is None:
+            return True
+        if app_id in self.accessible_app_ids:
+            return True
+        return self.can_manage_own_apps and maintainer is not None and maintainer == account_id
+
+    def apply_to_params(self, params: AppListBaseParams) -> None:
+        if self.accessible_app_ids is None:
+            return
+        params.accessible_app_ids = sorted(self.accessible_app_ids)
+        params.include_own_apps = self.can_manage_own_apps
+
+
+def resolve_app_access_filter(
+    tenant_id: str,
+    account_id: str,
+    *,
+    permissions: MyPermissionsResponse | None = None,
+) -> AppAccessFilter:
+    """Compute the RBAC app-access filter for ``account_id`` in ``tenant_id``.
+
+    Pass ``permissions`` when the caller has already fetched the snapshot (the
+    console controller reuses it for per-app permission keys) to avoid a second
+    inner-API round trip; otherwise it is fetched here.
+    """
+    if permissions is None:
+        permissions = enterprise_rbac_service.RBACService.MyPermissions.get(tenant_id, account_id)
+    whitelist_scope = enterprise_rbac_service.RBACService.AppAccess.whitelist_resources(tenant_id, account_id)
+
+    can_manage_own_apps = _MANAGE_OWN_APPS_PERMISSION_KEY in permissions.workspace.permission_keys
+    has_default_preview = has_app_list_permission(permissions.app.default_permission_keys) or has_app_list_permission(
+        permissions.workspace.permission_keys
+    )
+
+    permission_app_ids: set[str] | None = None
+    if not has_default_preview:
+        # Collect apps the caller can preview via per-app permission overrides.
+        permission_app_ids = {
+            override.resource_id
+            for override in permissions.app.overrides
+            if has_app_list_permission(override.permission_keys)
+        }
+
+    accessible_app_ids: set[str] | None
+    if getattr(whitelist_scope, "unrestricted", False):
+        accessible_app_ids = permission_app_ids
+    else:
+        accessible_app_ids = set(whitelist_scope.resource_ids)
+        if permission_app_ids is not None:
+            accessible_app_ids |= permission_app_ids
+        elif has_default_preview:
+            # Default preview overrides the whitelist restriction.
+            accessible_app_ids = None
+
+    return AppAccessFilter(accessible_app_ids=accessible_app_ids, can_manage_own_apps=can_manage_own_apps)

api/controllers/common/wraps.py

@@ -1,23 +1,3 @@
-"""Shared decorator utilities for Dify controller layers.
-
-This module provides decorators that are not tied to any single API group (e.g.
-console, inner, service).  Currently it exposes the RBAC permission gate, which
-can be applied to any blueprint.
-
-Key exports
------------
-``rbac_permission_required`` – decorator that enforces enterprise RBAC access
-    control.  When ``RBAC_ENABLED`` is ``False`` it is a no-op.
-
-``RBACPermission``, ``RBACResourceScope`` – re-exported from ``core.rbac`` so
-    callers only need a single import site.
-
-Private helpers
----------------
-``_extract_resource_id``, ``_is_resource_owned_by_current_user`` – kept module-
-    private but accessible via the module namespace for unit-test patching.
-"""
-
 from collections.abc import Callable
 from functools import wraps
 
@@ -32,7 +12,57 @@ from models.dataset import Dataset
 from models.model import App
 from services.enterprise.rbac_service import RBACService
 
-__all__ = ["RBACPermission", "RBACResourceScope", "rbac_permission_required"]
+__all__ = ["RBACPermission", "RBACResourceScope", "enforce_rbac_access", "rbac_permission_required"]
+
+
+def enforce_rbac_access(
+    *,
+    tenant_id: str,
+    account_id: str,
+    resource_type: RBACResourceScope,
+    scene: RBACPermission,
+    resource_required: bool = True,
+    path_args: dict[str, object] | None = None,
+) -> None:
+    """Enforce enterprise RBAC for an explicit account/tenant pair.
+
+    This is the flask-login-independent core of the RBAC gate so it can run
+    inside request-handling layers that resolve the caller themselves (e.g. the
+    openapi auth pipeline, which has the account on ``AuthData`` before
+    flask-login is mounted).
+
+    No-op when ``RBAC_ENABLED`` is ``False``. For resource-scoped checks the
+    resource ID is taken from ``path_args`` merged with ``request.view_args``;
+    resource ownership short-circuits the check. Raises ``Forbidden`` when
+    access is denied. For workspace-level checks pass ``resource_required=False``
+    so the RBAC request omits ``resource_id``.
+
+    Args:
+        tenant_id: The tenant the access is evaluated against.
+        account_id: The account requesting access.
+        resource_type: The :class:`RBACResourceScope` member (app/dataset/workspace).
+        scene: The :class:`RBACPermission` permission point, e.g. ``RBACPermission.APP_DELETE``.
+        resource_required: Whether a concrete resource ID is required.
+        path_args: Extra path arguments to merge with ``request.view_args``.
+    """
+    if not dify_config.RBAC_ENABLED:
+        return
+
+    check_resource_type = None if resource_type == RBACResourceScope.WORKSPACE else resource_type
+    resource_id = None
+    if resource_required and check_resource_type:
+        resource_id = _extract_resource_id(resource_type, path_args)
+        if _is_resource_owned_by_current_user(tenant_id, account_id, resource_type, resource_id):
+            return
+    allowed = RBACService.CheckAccess.check(
+        tenant_id,
+        account_id,
+        scene=scene,
+        resource_type=check_resource_type,
+        resource_id=resource_id,
+    )
+    if not allowed:
+        raise Forbidden()
 
 
 def rbac_permission_required[**P, R](
@@ -41,14 +71,12 @@ def rbac_permission_required[**P, R](
     *,
     resource_required: bool = True,
 ) -> Callable[[Callable[P, R]], Callable[P, R]]:
-    """Check enterprise RBAC permissions for the current user.
+    """Check enterprise RBAC permissions for the current flask-login user.
 
     When ``RBAC_ENABLED`` is ``False`` the decorator is a no-op and the
-    request passes through unchanged. When enabled it extracts the resource ID
-    from ``request.view_args`` for resource-scoped checks, calls the RBAC
-    service ``check-access`` endpoint, and raises ``Forbidden`` if the access
-    is denied. For workspace-level checks, set ``resource_required=False`` so
-    the RBAC request omits ``resource_id``.
+    request passes through unchanged. When enabled it resolves the current
+    account/tenant and delegates to :func:`enforce_rbac_access`, raising
+    ``Forbidden`` if access is denied.
 
     Args:
         resource_type: The :class:`RBACResourceScope` member (app/dataset/workspace).
@@ -63,23 +91,14 @@ def rbac_permission_required[**P, R](
                 return view(*args, **kwargs)
 
             current_user, current_tenant_id = current_account_with_tenant()
-            check_resource_type = None if resource_type == RBACResourceScope.WORKSPACE else resource_type
-            resource_id = None
-            if resource_required and check_resource_type:
-                resource_id = _extract_resource_id(resource_type, kwargs)
-                if _is_resource_owned_by_current_user(current_tenant_id, current_user.id, resource_type, resource_id):
-                    return view(*args, **kwargs)
-            allowed = RBACService.CheckAccess.check(
-                current_tenant_id,
-                current_user.id,
+            enforce_rbac_access(
+                tenant_id=current_tenant_id,
+                account_id=current_user.id,
+                resource_type=resource_type,
                 scene=scene,
-                resource_type=check_resource_type,
-                resource_id=resource_id,
+                resource_required=resource_required,
+                path_args=kwargs,
             )
-
-            if not allowed:
-                raise Forbidden()
-
             return view(*args, **kwargs)
 
         return decorated

api/controllers/console/agent/composer.py

@@ -28,9 +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
+from services.entities.agent_entities import ComposerSavePayload, WorkflowComposerCopyFromRosterPayload
 
-register_schema_models(console_ns, ComposerSavePayload)
+register_schema_models(console_ns, ComposerSavePayload, WorkflowComposerCopyFromRosterPayload)
 register_response_schema_models(
     console_ns,
     AgentAppComposerResponse,
@@ -91,6 +91,38 @@ class WorkflowAgentComposerApi(Resource):
         )
 
 
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/agent-composer/copy-from-roster")
+class WorkflowAgentComposerCopyFromRosterApi(Resource):
+    @console_ns.expect(console_ns.models[WorkflowComposerCopyFromRosterPayload.__name__])
+    @console_ns.response(
+        200,
+        "Workflow roster agent copied to inline agent",
+        console_ns.models[WorkflowAgentComposerResponse.__name__],
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @get_app_model(mode=[AppMode.WORKFLOW, AppMode.ADVANCED_CHAT])
+    @with_current_user_id
+    @with_current_tenant_id
+    def post(self, tenant_id: str, account_id: str, app_model: App, node_id: str):
+        payload = WorkflowComposerCopyFromRosterPayload.model_validate(console_ns.payload or {})
+        return dump_response(
+            WorkflowAgentComposerResponse,
+            AgentComposerService.copy_workflow_composer_from_roster(
+                tenant_id=tenant_id,
+                app_id=app_model.id,
+                node_id=node_id,
+                account_id=account_id,
+                source_agent_id=payload.source_agent_id,
+                source_snapshot_id=payload.source_snapshot_id,
+                idempotency_key=payload.idempotency_key,
+            ),
+        )
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/agent-composer/validate")
 class WorkflowAgentComposerValidateApi(Resource):
     @console_ns.expect(console_ns.models[ComposerSavePayload.__name__])
@@ -104,7 +136,7 @@ class WorkflowAgentComposerValidateApi(Resource):
     @with_current_tenant_id
     def post(self, tenant_id: str, app_model: App, node_id: str):
         payload = ComposerSavePayload.model_validate(console_ns.payload or {})
-        ComposerConfigValidator.validate_save_payload(payload)
+        ComposerConfigValidator.validate_publish_payload(payload)
         findings = AgentComposerService.collect_validation_findings(
             tenant_id=tenant_id,
             payload=payload,
@@ -238,7 +270,7 @@ class AgentComposerValidateApi(Resource):
     def post(self, tenant_id: str, agent_id: UUID):
         _resolve_agent_app_id(tenant_id=tenant_id, agent_id=agent_id)
         payload = ComposerSavePayload.model_validate(console_ns.payload or {})
-        ComposerConfigValidator.validate_save_payload(payload)
+        ComposerConfigValidator.validate_publish_payload(payload)
         findings = AgentComposerService.collect_validation_findings(
             tenant_id=tenant_id,
             payload=payload,

api/controllers/console/agent/roster.py

@@ -3,16 +3,17 @@ from uuid import UUID
 from flask import abort, request
 from flask_restx import Resource
 from pydantic import AliasChoices, BaseModel, Field, field_validator
+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
+from controllers.console.apikey import ApiKeyItem, ApiKeyList, BaseApiKeyListResource, BaseApiKeyResource
 from controllers.console.app.app import (
     AppDetailWithSite as GenericAppDetailWithSite,
 )
 from controllers.console.app.app import (
     AppListQuery,
-    CopyAppPayload,
     _normalize_app_list_query_args,
 )
 from controllers.console.app.app import (
@@ -25,9 +26,13 @@ from controllers.console.app.app import (
     UpdateAppPayload as GenericUpdateAppPayload,
 )
 from controllers.console.wraps import (
+    RBACPermission,
+    RBACResourceScope,
     account_initialization_required,
     edit_permission_required,
     enterprise_license_required,
+    is_admin_or_owner_required,
+    rbac_permission_required,
     setup_required,
     with_current_tenant_id,
     with_current_user,
@@ -36,6 +41,7 @@ from extensions.ext_database import db
 from fields.agent_fields import (
     AgentConfigSnapshotDetailResponse,
     AgentConfigSnapshotListResponse,
+    AgentConfigSnapshotRestoreResponse,
     AgentInviteOptionsResponse,
     AgentLogListResponse,
     AgentLogMessageListResponse,
@@ -48,7 +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.model import IconType
+from models.enums import ApiTokenType
+from models.model import ApiToken, App, IconType
 from services.agent.errors import AgentNotFoundError
 from services.agent.observability_service import (
     AgentLogQueryParams,
@@ -102,6 +109,46 @@ class AgentAppUpdatePayload(GenericUpdateAppPayload):
         return role
 
 
+class AgentAppCopyPayload(BaseModel):
+    name: str | None = Field(default=None, description="Name for the copied agent")
+    description: str | None = Field(default=None, description="Description for the copied agent", max_length=400)
+    role: str | None = Field(default=None, description="Role for the copied agent", max_length=255)
+    icon_type: IconType | None = Field(default=None, description="Icon type")
+    icon: str | None = Field(default=None, description="Icon")
+    icon_background: str | None = Field(default=None, description="Icon background color")
+
+    @field_validator("role")
+    @classmethod
+    def validate_role(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        role = value.strip()
+        if not role:
+            raise ValueError("Agent role is required when provided.")
+        return role
+
+
+class AgentApiStatusPayload(BaseModel):
+    enable_api: bool = Field(..., description="Enable or disable Agent service API")
+
+
+class AgentApiAccessResponse(BaseModel):
+    enabled: bool
+    service_api_base_url: str
+    streaming_only: bool = True
+    chat_endpoint: str
+    stop_endpoint: str
+    conversations_endpoint: str
+    messages_endpoint: str
+    files_upload_endpoint: str
+    parameters_endpoint: str
+    info_endpoint: str
+    meta_endpoint: str
+    api_rpm: int
+    api_rph: int
+    api_key_count: int
+
+
 class AgentAppPublishedReferenceResponse(BaseModel):
     app_id: str
     app_name: str
@@ -185,6 +232,7 @@ class AgentStatisticsQuery(BaseModel):
 
 class AgentAppPartial(GenericAppPartial):
     app_id: str | None = None
+    debug_conversation_id: str | None = None
     role: str | None = None
     active_config_is_published: bool = False
     published_reference_count: int = 0
@@ -193,10 +241,15 @@ class AgentAppPartial(GenericAppPartial):
 
 class AgentAppDetailWithSite(GenericAppDetailWithSite):
     app_id: str | None = None
+    debug_conversation_id: str | None = None
     role: str | None = None
     active_config_is_published: bool = False
 
 
+class AgentDebugConversationRefreshResponse(BaseModel):
+    debug_conversation_id: str
+
+
 class AgentAppPagination(GenericAppPagination):
     data: list[AgentAppPartial] = Field(  # type: ignore[assignment]  # pyrefly: ignore[bad-override-mutable-attribute]
         validation_alias=AliasChoices("items", "data")
@@ -207,7 +260,8 @@ register_schema_models(
     console_ns,
     AgentAppCreatePayload,
     AgentAppUpdatePayload,
-    CopyAppPayload,
+    AgentAppCopyPayload,
+    AgentApiStatusPayload,
     AgentInviteOptionsQuery,
     AgentLogsQuery,
     AgentStatisticsQuery,
@@ -218,11 +272,14 @@ register_schema_models(
 register_response_schema_models(
     console_ns,
     AgentAppPagination,
+    AgentApiAccessResponse,
     AgentAppPublishedReferenceResponse,
     AgentAppDetailWithSite,
     AgentAppPartial,
+    AgentDebugConversationRefreshResponse,
     AgentConfigSnapshotDetailResponse,
     AgentConfigSnapshotListResponse,
+    AgentConfigSnapshotRestoreResponse,
     AgentInviteOptionsResponse,
     AgentLogListResponse,
     AgentLogMessageListResponse,
@@ -237,7 +294,7 @@ def _agent_roster_service() -> AgentRosterService:
     return AgentRosterService(db.session)
 
 
-def _serialize_agent_app_detail(app_model) -> 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`
@@ -260,6 +317,11 @@ def _serialize_agent_app_detail(app_model) -> dict:
     payload.pop("bound_agent_id", None)
     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,
+        agent_id=agent.id,
+        account_id=current_user.id,
+    )
     payload["role"] = agent.role or ""
     payload["active_config_is_published"] = roster_service.active_config_is_published(
         tenant_id=app_model.tenant_id,
@@ -268,7 +330,7 @@ def _serialize_agent_app_detail(app_model) -> dict:
     return payload
 
 
-def _serialize_agent_app_pagination(app_pagination, *, tenant_id: str) -> dict:
+def _serialize_agent_app_pagination(app_pagination, *, tenant_id: str, current_user: Account) -> dict:
     """Serialize Agent App lists with roster-shaped items.
 
     Each item starts from the shared App list shape, then drops
@@ -291,6 +353,11 @@ def _serialize_agent_app_pagination(app_pagination, *, tenant_id: str) -> dict:
         tenant_id=tenant_id,
         agent_ids=[agent.id for agent in agents_by_app_id.values()],
     )
+    debug_conversation_ids_by_agent_id = roster_service.load_or_create_agent_app_debug_conversation_ids_by_agent_id(
+        tenant_id=tenant_id,
+        agents=list(agents_by_app_id.values()),
+        account_id=current_user.id,
+    )
     payload = AgentAppPagination.model_validate(app_pagination, from_attributes=True).model_dump(mode="json")
     for item in payload["data"]:
         app_id = item["id"]
@@ -299,6 +366,7 @@ def _serialize_agent_app_pagination(app_pagination, *, tenant_id: str) -> dict:
         if agent:
             item["app_id"] = app_id
             item["id"] = agent.id
+            item["debug_conversation_id"] = debug_conversation_ids_by_agent_id.get(agent.id)
             item["role"] = agent.role or ""
             item["active_config_is_published"] = active_config_is_published_by_agent_id.get(agent.id, False)
             published_references = published_references_by_agent_id.get(agent.id, [])
@@ -323,6 +391,38 @@ def _resolve_agent_app_model(*, tenant_id: str, agent_id: UUID):
     return resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
 
 
+def _agent_api_key_count(app_id: str) -> int:
+    return (
+        db.session.scalar(
+            select(func.count(ApiToken.id)).where(
+                ApiToken.type == ApiTokenType.APP,
+                ApiToken.app_id == app_id,
+            )
+        )
+        or 0
+    )
+
+
+def _serialize_agent_api_access(app_model: App) -> dict:
+    base_url = app_model.api_base_url
+    response = AgentApiAccessResponse(
+        enabled=bool(app_model.enable_api),
+        service_api_base_url=base_url,
+        chat_endpoint=f"{base_url}/chat-messages",
+        stop_endpoint=f"{base_url}/chat-messages/{{task_id}}/stop",
+        conversations_endpoint=f"{base_url}/conversations",
+        messages_endpoint=f"{base_url}/messages",
+        files_upload_endpoint=f"{base_url}/files/upload",
+        parameters_endpoint=f"{base_url}/parameters",
+        info_endpoint=f"{base_url}/info",
+        meta_endpoint=f"{base_url}/meta",
+        api_rpm=app_model.api_rpm or 0,
+        api_rph=app_model.api_rph or 0,
+        api_key_count=_agent_api_key_count(str(app_model.id)),
+    )
+    return response.model_dump(mode="json")
+
+
 def _agent_observability_service() -> AgentObservabilityService:
     return AgentObservabilityService(db.session)
 
@@ -374,7 +474,11 @@ class AgentAppListApi(Resource):
             empty = AgentAppPagination(page=args.page, limit=args.limit, total=0, has_more=False, data=[])
             return empty.model_dump(mode="json")
 
-        return _serialize_agent_app_pagination(app_pagination, tenant_id=current_tenant_id)
+        return _serialize_agent_app_pagination(
+            app_pagination,
+            tenant_id=current_tenant_id,
+            current_user=current_user,
+        )
 
     @console_ns.expect(console_ns.models[AgentAppCreatePayload.__name__])
     @console_ns.response(201, "Agent app created successfully", console_ns.models[AgentAppDetailWithSite.__name__])
@@ -399,7 +503,7 @@ class AgentAppListApi(Resource):
         )
 
         app = AppService().create_app(current_tenant_id, params, current_user)
-        return _serialize_agent_app_detail(app), 201
+        return _serialize_agent_app_detail(app, current_user=current_user), 201
 
 
 @console_ns.route("/agent/<uuid:agent_id>")
@@ -409,10 +513,11 @@ class AgentAppApi(Resource):
     @login_required
     @account_initialization_required
     @enterprise_license_required
+    @with_current_user
     @with_current_tenant_id
-    def get(self, tenant_id: str, agent_id: UUID):
+    def get(self, tenant_id: str, current_user: Account, agent_id: UUID):
         app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
-        return _serialize_agent_app_detail(app_model)
+        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__])
@@ -422,8 +527,9 @@ class AgentAppApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @with_current_user
     @with_current_tenant_id
-    def put(self, tenant_id: str, agent_id: UUID):
+    def put(self, tenant_id: str, current_user: Account, agent_id: UUID):
         app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         args = AgentAppUpdatePayload.model_validate(console_ns.payload)
         args_dict: AppService.ArgsDict = {
@@ -437,7 +543,7 @@ class AgentAppApi(Resource):
             "role": args.role,
         }
         updated = AppService().update_app(app_model, args_dict)
-        return _serialize_agent_app_detail(updated)
+        return _serialize_agent_app_detail(updated, current_user=current_user)
 
     @console_ns.response(204, "Agent app deleted successfully")
     @console_ns.response(403, "Insufficient permissions")
@@ -452,9 +558,34 @@ class AgentAppApi(Resource):
         return "", 204
 
 
+@console_ns.route("/agent/<uuid:agent_id>/debug-conversation/refresh")
+class AgentDebugConversationRefreshApi(Resource):
+    @console_ns.response(
+        200,
+        "Agent debug conversation refreshed",
+        console_ns.models[AgentDebugConversationRefreshResponse.__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):
+        debug_conversation_id = _agent_roster_service().refresh_agent_app_debug_conversation_id(
+            tenant_id=tenant_id,
+            agent_id=str(agent_id),
+            account_id=current_user.id,
+        )
+        return AgentDebugConversationRefreshResponse(debug_conversation_id=debug_conversation_id).model_dump(
+            mode="json"
+        )
+
+
 @console_ns.route("/agent/<uuid:agent_id>/copy")
 class AgentAppCopyApi(Resource):
-    @console_ns.expect(console_ns.models[CopyAppPayload.__name__])
+    @console_ns.expect(console_ns.models[AgentAppCopyPayload.__name__])
     @console_ns.response(201, "Agent app copied successfully", console_ns.models[AgentAppDetailWithSite.__name__])
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.response(400, "Invalid request parameters")
@@ -465,18 +596,88 @@ class AgentAppCopyApi(Resource):
     @with_current_user
     @with_current_tenant_id
     def post(self, tenant_id: str, current_user: Account, agent_id: UUID):
-        args = CopyAppPayload.model_validate(console_ns.payload or {})
+        args = AgentAppCopyPayload.model_validate(console_ns.payload or {})
         copied_app = _agent_roster_service().duplicate_agent_app(
             tenant_id=tenant_id,
             agent_id=str(agent_id),
             account=current_user,
             name=args.name,
             description=args.description,
+            role=args.role,
             icon_type=args.icon_type,
             icon=args.icon,
             icon_background=args.icon_background,
         )
-        return _serialize_agent_app_detail(copied_app), 201
+        return _serialize_agent_app_detail(copied_app, current_user=current_user), 201
+
+
+@console_ns.route("/agent/<uuid:agent_id>/api-access")
+class AgentApiAccessApi(Resource):
+    @console_ns.response(200, "Agent service API access", console_ns.models[AgentApiAccessResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_tenant_id
+    def get(self, tenant_id: str, agent_id: UUID):
+        app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        return _serialize_agent_api_access(app_model)
+
+
+@console_ns.route("/agent/<uuid:agent_id>/api-enable")
+class AgentApiStatusApi(Resource):
+    @console_ns.expect(console_ns.models[AgentApiStatusPayload.__name__])
+    @console_ns.response(200, "Agent service API status updated", console_ns.models[AgentApiAccessResponse.__name__])
+    @console_ns.response(403, "Insufficient permissions")
+    @setup_required
+    @login_required
+    @is_admin_or_owner_required
+    @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @with_current_tenant_id
+    def post(self, tenant_id: str, agent_id: UUID):
+        app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        args = AgentApiStatusPayload.model_validate(console_ns.payload)
+        app_model = AppService().update_app_api_status(app_model, args.enable_api)
+        return _serialize_agent_api_access(app_model)
+
+
+@console_ns.route("/agent/<uuid:agent_id>/api-keys")
+class AgentApiKeyListApi(BaseApiKeyListResource):
+    resource_type = ApiTokenType.APP
+    resource_model = App
+    resource_id_field = "app_id"
+    token_prefix = "app-"
+
+    @console_ns.response(200, "Agent service API keys", console_ns.models[ApiKeyList.__name__])
+    @with_current_tenant_id
+    def get(self, tenant_id: str, agent_id: UUID) -> dict[str, object]:
+        app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        return dump_response(ApiKeyList, self._get_api_key_list(str(app_model.id), tenant_id))
+
+    @console_ns.response(201, "Agent service API key created", console_ns.models[ApiKeyItem.__name__])
+    @console_ns.response(400, "Maximum keys exceeded")
+    @with_current_tenant_id
+    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    def post(self, tenant_id: str, agent_id: UUID) -> tuple[dict[str, object], int]:
+        app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        return dump_response(ApiKeyItem, self._create_api_key(str(app_model.id), tenant_id)), 201
+
+
+@console_ns.route("/agent/<uuid:agent_id>/api-keys/<uuid:api_key_id>")
+class AgentApiKeyApi(BaseApiKeyResource):
+    resource_type = ApiTokenType.APP
+    resource_model = App
+    resource_id_field = "app_id"
+
+    @console_ns.response(204, "Agent service API key deleted")
+    @with_current_user
+    @with_current_tenant_id
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    def delete(self, tenant_id: str, current_user: Account, agent_id: UUID, api_key_id: UUID) -> tuple[str, int]:
+        app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        self._delete_api_key(str(app_model.id), str(api_key_id), tenant_id, current_user)
+        return "", 204
 
 
 @console_ns.route("/agent/invite-options")
@@ -649,3 +850,24 @@ class AgentRosterVersionDetailApi(Resource):
                 version_id=str(version_id),
             ),
         )
+
+
+@console_ns.route("/agent/<uuid:agent_id>/versions/<uuid:version_id>/restore")
+class AgentRosterVersionRestoreApi(Resource):
+    @console_ns.response(200, "Agent version restored", console_ns.models[AgentConfigSnapshotRestoreResponse.__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, version_id: UUID):
+        return dump_response(
+            AgentConfigSnapshotRestoreResponse,
+            _agent_roster_service().restore_agent_version(
+                tenant_id=tenant_id,
+                agent_id=str(agent_id),
+                version_id=str(version_id),
+                account_id=current_user.id,
+            ),
+        )

api/controllers/console/app/agent.py

@@ -1,4 +1,3 @@
-import logging
 from typing import Any
 from uuid import UUID
 
@@ -30,7 +29,6 @@ from fields.base import ResponseModel
 from libs.helper import uuid_value
 from libs.login import login_required
 from models import Account
-from models.agent_config_entities import AgentFileRefConfig, AgentSkillRefConfig
 from models.model import App, AppMode, UploadFile
 from services.agent.composer_service import AgentComposerService
 from services.agent.skill_package_service import SkillManifest, SkillPackageError
@@ -49,8 +47,6 @@ from services.agent_drive_service import (
 )
 from services.agent_service import AgentService
 
-logger = logging.getLogger(__name__)
-
 _WORKFLOW_AGENT_DRIVE_APP_MODES = [AppMode.WORKFLOW, AppMode.ADVANCED_CHAT]
 _AGENT_SKILL_UPLOAD_PARAMS = {
     "file": {
@@ -130,8 +126,16 @@ class AgentLogResponse(ResponseModel):
     files: list[Any] = Field(default_factory=list)
 
 
+class AgentUploadedSkillResponse(ResponseModel):
+    name: str
+    description: str
+    path: str
+    skill_md_key: str
+    archive_key: str | None = None
+
+
 class AgentSkillUploadResponse(ResponseModel):
-    skill: AgentSkillRefConfig
+    skill: AgentUploadedSkillResponse
     manifest: SkillManifest
 
 
@@ -145,13 +149,11 @@ class AgentDriveFileResponse(ResponseModel):
 
 class AgentDriveFileCommitResponse(ResponseModel):
     file: AgentDriveFileResponse
-    config_version_id: str | None = None
 
 
 class AgentDriveDeleteResponse(ResponseModel):
     result: str
     removed_keys: list[str] = Field(default_factory=list)
-    config_version_id: str | None = None
 
 
 register_schema_models(console_ns, AgentLogQuery, AgentDriveFilePayload, AgentDriveDeleteFileByAgentQuery)
@@ -161,6 +163,7 @@ register_response_schema_models(
     AgentDriveFileCommitResponse,
     AgentDriveFileResponse,
     AgentLogResponse,
+    AgentUploadedSkillResponse,
     AgentSkillUploadResponse,
     SkillToolInferenceResult,
 )
@@ -242,24 +245,6 @@ def _commit_drive_file_for_app(*, current_user: Account, app_model: App, allow_n
         return {"code": exc.code, "message": exc.message}, exc.status_code
 
     row = committed[0]
-    file_ref = AgentFileRefConfig.model_validate(
-        {
-            "id": row["key"],
-            "name": upload_file.name,
-            "file_id": upload_file.id,
-            "drive_key": row["key"],
-            "type": row.get("mime_type"),
-            "size": row.get("size"),
-        }
-    )
-    config_version_id = AgentComposerService.add_drive_file_ref(
-        tenant_id=app_model.tenant_id,
-        agent_id=agent_id,
-        account_id=current_user.id,
-        file_ref=file_ref,
-        app_id=app_model.id,
-        node_id=node_id,
-    )
     return {
         "file": {
             "name": upload_file.name,
@@ -268,7 +253,6 @@ def _commit_drive_file_for_app(*, current_user: Account, app_model: App, allow_n
             "size": row.get("size"),
             "mime_type": row.get("mime_type"),
         },
-        "config_version_id": config_version_id,
     }, 201
 
 
@@ -283,24 +267,17 @@ def _delete_drive_file_for_app(*, current_user: Account, app_model: App, allow_n
     except AgentDriveError as exc:
         return {"code": exc.code, "message": exc.message}, exc.status_code
 
-    config_version_id = AgentComposerService.remove_drive_refs(
-        tenant_id=app_model.tenant_id,
-        agent_id=agent_id,
-        account_id=current_user.id,
-        file_key=key,
-        app_id=app_model.id,
-        node_id=node_id,
-    )
-    removed_keys: list[str] = []
     try:
-        removed_keys = AgentDriveService().delete(tenant_id=app_model.tenant_id, agent_id=agent_id, key=key)
+        result = AgentDriveService().commit(
+            tenant_id=app_model.tenant_id,
+            user_id=current_user.id,
+            agent_id=agent_id,
+            items=[DriveCommitItem(key=key, file_ref=None)],
+        )
     except AgentDriveError as exc:
         return {"code": exc.code, "message": exc.message}, exc.status_code
-    except Exception:
-        # Soul-first ordering: the ref is already gone; orphan KV rows are
-        # harmless and an idempotent DELETE retry cleans them.
-        logger.exception("agent drive delete failed for key %s (soul already updated)", key)
-    return {"result": "success", "removed_keys": removed_keys, "config_version_id": config_version_id}
+    removed_keys = [item["key"] for item in result if item.get("removed")]
+    return {"result": "success", "removed_keys": removed_keys}
 
 
 def _delete_skill_for_app(*, current_user: Account, app_model: App, slug: str, allow_node_id: bool = True):
@@ -312,22 +289,20 @@ def _delete_skill_for_app(*, current_user: Account, app_model: App, slug: str, a
     if "/" in slug or not slug.strip():
         return {"code": "drive_key_invalid", "message": "skill slug must be a single path segment"}, 400
 
-    config_version_id = AgentComposerService.remove_drive_refs(
-        tenant_id=app_model.tenant_id,
-        agent_id=agent_id,
-        account_id=current_user.id,
-        skill_slug=slug,
-        app_id=app_model.id,
-        node_id=node_id,
-    )
-    removed_keys: list[str] = []
     try:
-        removed_keys = AgentDriveService().delete(tenant_id=app_model.tenant_id, agent_id=agent_id, prefix=f"{slug}/")
+        result = AgentDriveService().commit(
+            tenant_id=app_model.tenant_id,
+            user_id=current_user.id,
+            agent_id=agent_id,
+            items=[
+                DriveCommitItem(key=f"{slug}/SKILL.md", file_ref=None),
+                DriveCommitItem(key=f"{slug}/.DIFY-SKILL-FULL.zip", file_ref=None),
+            ],
+        )
     except AgentDriveError as exc:
         return {"code": exc.code, "message": exc.message}, exc.status_code
-    except Exception:
-        logger.exception("agent drive delete failed for skill %s (soul already updated)", slug)
-    return {"result": "success", "removed_keys": removed_keys, "config_version_id": config_version_id}
+    removed_keys = [item["key"] for item in result if item.get("removed")]
+    return {"result": "success", "removed_keys": removed_keys}
 
 
 def _infer_skill_tools_for_app(*, app_model: App, slug: str):
@@ -455,7 +430,7 @@ class AgentDriveFilesApi(Resource):
         return _commit_drive_file_for_app(current_user=current_user, app_model=app_model)
 
     @console_ns.doc("delete_agent_drive_file")
-    @console_ns.doc(description="Delete one drive file by key; soul ref first, then the KV row (ENG-625 D5)")
+    @console_ns.doc(description="Delete one drive file by key via drive commit-null semantics")
     @console_ns.doc(params={"app_id": "Application ID", **query_params_from_model(AgentDriveDeleteFileQuery)})
     @console_ns.response(200, "File removed", console_ns.models[AgentDriveDeleteResponse.__name__])
     @setup_required
@@ -486,9 +461,7 @@ class AgentSkillByAgentApi(Resource):
 @console_ns.route("/apps/<uuid:app_id>/agent/skills/<string:slug>")
 class AgentSkillApi(Resource):
     @console_ns.doc("delete_agent_skill")
-    @console_ns.doc(
-        description="Delete a standardized skill: soul ref first, then the <slug>/ drive prefix (ENG-625 D5)"
-    )
+    @console_ns.doc(description="Delete a standardized skill by removing its known drive keys via commit-null")
     @console_ns.doc(
         params={
             "app_id": "Application ID",

api/controllers/console/app/agent_app_feature.py

@@ -91,7 +91,10 @@ class AgentAppFeatureConfigResource(Resource):
         args = AgentAppFeaturesPayload.model_validate(console_ns.payload or {})
 
         new_app_model_config = AgentAppFeatureConfigService.update_features(
-            app_model=app_model, account=current_user, config=args.model_dump(exclude_none=True), session=db.session
+            app_model=app_model,
+            account=current_user,
+            config=args.model_dump(exclude_none=True),
+            session=db.session,
         )
 
         app_model_config_was_updated.send(app_model, app_model_config=new_app_model_config)

api/controllers/console/app/agent_drive_inspector.py

@@ -10,8 +10,12 @@ backend — drive data lives in the API's own DB/storage, served straight from
 
 from __future__ import annotations
 
+import json
+from collections.abc import Mapping
+from typing import Any
 from uuid import UUID
 
+from flask import Response
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 
@@ -49,6 +53,10 @@ class AgentDriveFileByAgentQuery(BaseModel):
     key: str = Field(min_length=1, description="Drive key, e.g. tender-analyzer/SKILL.md")
 
 
+class AgentDriveSkillInspectQuery(BaseModel):
+    node_id: str | None = Field(default=None, description="Workflow node ID (workflow composer variant)")
+
+
 class AgentDriveItemResponse(ResponseModel):
     key: str
     size: int | None = None
@@ -56,12 +64,63 @@ class AgentDriveItemResponse(ResponseModel):
     hash: str | None = None
     file_kind: str
     created_at: int | None = None
+    is_skill: bool | None = None
+    skill_metadata: str | None = None
 
 
 class AgentDriveListResponse(ResponseModel):
     items: list[AgentDriveItemResponse] = Field(default_factory=list)
 
 
+class AgentDriveSkillItemResponse(ResponseModel):
+    path: str
+    skill_md_key: str
+    archive_key: str | None = None
+    name: str
+    description: str
+    size: int | None = None
+    mime_type: str | None = None
+    hash: str | None = None
+    created_at: int | None = None
+
+
+class AgentDriveSkillListResponse(ResponseModel):
+    items: list[AgentDriveSkillItemResponse] = Field(default_factory=list)
+
+
+class AgentDriveSkillFileResponse(ResponseModel):
+    path: str
+    name: str
+    type: str
+    drive_key: str | None = None
+    available_in_drive: bool
+
+
+class AgentDriveSkillMarkdownResponse(ResponseModel):
+    key: str
+    size: int | None = None
+    truncated: bool
+    binary: bool
+    text: str | None = None
+
+
+class AgentDriveSkillInspectResponse(ResponseModel):
+    path: str
+    skill_md_key: str
+    archive_key: str | None = None
+    name: str
+    description: str
+    size: int | None = None
+    mime_type: str | None = None
+    hash: str | None = None
+    created_at: int | None = None
+    source: str
+    files: list[AgentDriveSkillFileResponse] = Field(default_factory=list)
+    file_tree: list[dict[str, Any]] = Field(default_factory=list)
+    skill_md: AgentDriveSkillMarkdownResponse
+    warnings: list[str] = Field(default_factory=list)
+
+
 class AgentDrivePreviewResponse(ResponseModel):
     key: str
     size: int | None = None
@@ -75,7 +134,12 @@ class AgentDriveDownloadResponse(ResponseModel):
 
 
 register_response_schema_models(
-    console_ns, AgentDriveListResponse, AgentDrivePreviewResponse, AgentDriveDownloadResponse
+    console_ns,
+    AgentDriveDownloadResponse,
+    AgentDriveListResponse,
+    AgentDrivePreviewResponse,
+    AgentDriveSkillInspectResponse,
+    AgentDriveSkillListResponse,
 )
 
 
@@ -96,6 +160,13 @@ def _handle(exc: AgentDriveError) -> tuple[dict[str, object], int]:
     return {"code": exc.code, "message": exc.message}, exc.status_code
 
 
+def _json_response(data: Mapping[str, Any]):
+    return Response(
+        response=json.dumps(data, ensure_ascii=False, separators=(",", ":")),
+        content_type="application/json; charset=utf-8",
+    )
+
+
 _WORKFLOW_APP_MODES = [AppMode.WORKFLOW, AppMode.ADVANCED_CHAT]
 
 
@@ -119,6 +190,49 @@ class AgentDriveListByAgentApi(Resource):
         return {"items": [{k: v for k, v in item.items() if k != "file_id"} for item in items]}
 
 
+@console_ns.route("/agent/<uuid:agent_id>/drive/skills")
+class AgentDriveSkillListByAgentApi(Resource):
+    @console_ns.doc("list_agent_drive_skills_by_agent")
+    @console_ns.doc(description="List drive-backed skills for an Agent App")
+    @console_ns.doc(params={"agent_id": "Agent ID"})
+    @console_ns.response(200, "Drive skills", console_ns.models[AgentDriveSkillListResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_tenant_id
+    def get(self, tenant_id: str, agent_id: UUID):
+        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:
+            return _handle(exc)
+        return {"items": items}
+
+
+@console_ns.route("/agent/<uuid:agent_id>/drive/skills/<path:skill_path>/inspect")
+class AgentDriveSkillInspectByAgentApi(Resource):
+    @console_ns.doc("inspect_agent_drive_skill_by_agent")
+    @console_ns.doc(description="Inspect one drive-backed skill for slash-menu hover/detail UI")
+    @console_ns.doc(params={"agent_id": "Agent ID", "skill_path": "Skill path/slug, e.g. tender-analyzer"})
+    @console_ns.response(200, "Drive skill inspect view", console_ns.models[AgentDriveSkillInspectResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_tenant_id
+    def get(self, tenant_id: str, agent_id: UUID, skill_path: str):
+        resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
+        try:
+            return _json_response(
+                AgentDriveService().inspect_skill(
+                    tenant_id=tenant_id,
+                    agent_id=str(agent_id),
+                    skill_path=skill_path,
+                )
+            )
+        except AgentDriveError as exc:
+            return _handle(exc)
+
+
 @console_ns.route("/agent/<uuid:agent_id>/drive/files/preview")
 class AgentDrivePreviewByAgentApi(Resource):
     @console_ns.doc("preview_agent_drive_file_by_agent")
@@ -182,6 +296,61 @@ class AgentDriveListApi(Resource):
         return {"items": [{k: v for k, v in item.items() if k != "file_id"} for item in items]}
 
 
+@console_ns.route("/apps/<uuid:app_id>/agent/drive/skills")
+class AgentDriveSkillListApi(Resource):
+    @console_ns.doc("list_agent_drive_skills")
+    @console_ns.doc(description="List drive-backed skills for the bound agent")
+    @console_ns.doc(params={"app_id": "Application ID", **query_params_from_model(AgentDriveListQuery)})
+    @console_ns.response(200, "Drive skills", console_ns.models[AgentDriveSkillListResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=_WORKFLOW_APP_MODES)
+    def get(self, app_model: App):
+        query = query_params_from_request(AgentDriveListQuery)
+        agent_id = _resolve_agent_id(app_model, query.node_id)
+        if not agent_id:
+            return _agent_not_bound()
+        try:
+            items = AgentDriveService().list_skills(tenant_id=app_model.tenant_id, agent_id=agent_id)
+        except AgentDriveError as exc:
+            return _handle(exc)
+        return {"items": items}
+
+
+@console_ns.route("/apps/<uuid:app_id>/agent/drive/skills/<path:skill_path>/inspect")
+class AgentDriveSkillInspectApi(Resource):
+    @console_ns.doc("inspect_agent_drive_skill")
+    @console_ns.doc(description="Inspect one drive-backed skill for slash-menu hover/detail UI")
+    @console_ns.doc(
+        params={
+            "app_id": "Application ID",
+            "skill_path": "Skill path/slug, e.g. tender-analyzer",
+            **query_params_from_model(AgentDriveSkillInspectQuery),
+        }
+    )
+    @console_ns.response(200, "Drive skill inspect view", console_ns.models[AgentDriveSkillInspectResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=_WORKFLOW_APP_MODES)
+    def get(self, app_model: App, skill_path: str):
+        query = query_params_from_request(AgentDriveSkillInspectQuery)
+        agent_id = _resolve_agent_id(app_model, query.node_id)
+        if not agent_id:
+            return _agent_not_bound()
+        try:
+            return _json_response(
+                AgentDriveService().inspect_skill(
+                    tenant_id=app_model.tenant_id,
+                    agent_id=agent_id,
+                    skill_path=skill_path,
+                )
+            )
+        except AgentDriveError as exc:
+            return _handle(exc)
+
+
 @console_ns.route("/apps/<uuid:app_id>/agent/drive/files/preview")
 class AgentDrivePreviewApi(Resource):
     @console_ns.doc("preview_agent_drive_file")
@@ -232,4 +401,8 @@ __all__ = [
     "AgentDriveListByAgentApi",
     "AgentDrivePreviewApi",
     "AgentDrivePreviewByAgentApi",
+    "AgentDriveSkillInspectApi",
+    "AgentDriveSkillInspectByAgentApi",
+    "AgentDriveSkillListApi",
+    "AgentDriveSkillListByAgentApi",
 ]

api/controllers/console/app/annotation.py

@@ -19,6 +19,7 @@ from controllers.console.wraps import (
     rbac_permission_required,
     setup_required,
 )
+from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.annotation_fields import (
     Annotation,
@@ -388,7 +389,9 @@ class AnnotationUpdateDeleteApi(Resource):
             update_args["answer"] = args.answer
         if args.question is not None:
             update_args["question"] = args.question
-        annotation = AppAnnotationService.update_app_annotation_directly(update_args, str(app_id), str(annotation_id))
+        annotation = AppAnnotationService.update_app_annotation_directly(
+            update_args, str(app_id), str(annotation_id), db.session
+        )
         return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")
 
     @setup_required
@@ -398,7 +401,7 @@ class AnnotationUpdateDeleteApi(Resource):
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @console_ns.response(204, "Annotation deleted successfully")
     def delete(self, app_id: UUID, annotation_id: UUID):
-        AppAnnotationService.delete_app_annotation(str(app_id), str(annotation_id))
+        AppAnnotationService.delete_app_annotation(str(app_id), str(annotation_id), db.session)
         return "", 204
 
 

api/controllers/console/app/app.py

@@ -14,6 +14,7 @@ from werkzeug.datastructures import MultiDict
 from werkzeug.exceptions import BadRequest, NotFound
 
 from configs import dify_config
+from controllers.common.app_access import resolve_app_access_filter
 from controllers.common.fields import RedirectUrlResponse, SimpleResultResponse
 from controllers.common.helpers import FileInfo
 from controllers.common.schema import (
@@ -42,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
@@ -68,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"]
 
@@ -78,7 +81,6 @@ _TAG_IDS_BRACKET_PATTERN = re.compile(r"^tag_ids\[(\d+)\]$")
 _CREATOR_IDS_BRACKET_PATTERN = re.compile(r"^creator_ids\[(\d+)\]$")
 AppListMode = Literal["completion", "chat", "advanced-chat", "workflow", "agent-chat", "agent", "channel", "all"]
 DEFAULT_APP_LIST_MODE: AppListMode = "all"
-APP_LIST_PERMISSION_KEYS = frozenset({"app.preview", "app.acl.preview", "app.full_access"})
 
 
 class AppListBaseQuery(BaseModel):
@@ -167,10 +169,6 @@ def _normalize_app_list_query_args(query_args: MultiDict[str, str]) -> dict[str,
     return normalized
 
 
-def _has_app_list_permission(permission_keys: Sequence[str]) -> bool:
-    return any(permission_key in APP_LIST_PERMISSION_KEYS for permission_key in permission_keys)
-
-
 class CreateAppPayload(BaseModel):
     name: str = Field(..., min_length=1, description="App name")
     description: str | None = Field(default=None, description="App description (max 400 chars)", max_length=400)
@@ -612,38 +610,12 @@ class AppListApi(Resource):
             current_user_id,
         )
         if dify_config.RBAC_ENABLED:
-            whitelist_scope = enterprise_rbac_service.RBACService.AppAccess.whitelist_resources(
+            access_filter = resolve_app_access_filter(
                 str(current_tenant_id),
                 current_user_id,
+                permissions=permissions,
             )
-            can_manage_own_apps = "app.create_and_management" in permissions.workspace.permission_keys
-            has_default_preview = _has_app_list_permission(
-                permissions.app.default_permission_keys
-            ) or _has_app_list_permission(permissions.workspace.permission_keys)
-            permission_app_ids: set[str] | None = None
-            if not has_default_preview:
-                permission_app_ids = {
-                    override.resource_id
-                    for override in permissions.app.overrides
-                    if _has_app_list_permission(override.permission_keys)
-                }
-
-            if getattr(whitelist_scope, "unrestricted", False):
-                accessible_app_ids = permission_app_ids
-            else:
-                accessible_app_ids = set(whitelist_scope.resource_ids)
-                if permission_app_ids is not None:
-                    accessible_app_ids |= permission_app_ids
-                elif has_default_preview:
-                    accessible_app_ids = None
-
-            if accessible_app_ids:
-                params.accessible_app_ids = sorted(accessible_app_ids)
-                params.include_own_apps = can_manage_own_apps
-            elif accessible_app_ids is not None and can_manage_own_apps:
-                params.is_created_by_me = True
-            elif accessible_app_ids is not None:
-                params.accessible_app_ids = []
+            access_filter.apply_to_params(params)
 
         # get app list
         app_service = AppService()
@@ -696,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,
@@ -1124,7 +1104,7 @@ class AppTraceApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_MONITOR)
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
     def post(self, app_model: App):
         # add app trace

api/controllers/console/app/audio.py

@@ -30,6 +30,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
+from extensions.ext_database import db
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs.login import login_required
 from models import App, AppMode
@@ -142,6 +143,7 @@ class ChatMessageTextApi(Resource):
 
             response = AudioService.transcript_tts(
                 app_model=app_model,
+                session=db.session,
                 text=payload.text,
                 voice=payload.voice,
                 message_id=payload.message_id,

api/controllers/console/app/completion.py

@@ -40,12 +40,15 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id
+from extensions.ext_database import db
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
 from libs.login import login_required
 from models import Account
 from models.model import App, AppMode
+from services.agent.errors import AgentNotFoundError
+from services.agent.roster_service import AgentRosterService
 from services.app_generate_service import AppGenerateService
 from services.app_task_service import AppTaskService
 from services.errors.llm import InvokeRateLimitError
@@ -191,10 +194,11 @@ class ChatMessageApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TEST_AND_RUN)
     @get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.AGENT])
-    def post(self, current_user: Account, app_model: App):
-        return _create_chat_message(current_user=current_user, app_model=app_model)
+    def post(self, current_tenant_id: str, current_user: Account, app_model: App):
+        return _create_chat_message(current_tenant_id=current_tenant_id, current_user=current_user, app_model=app_model)
 
 
 @console_ns.route("/agent/<uuid:agent_id>/chat-messages")
@@ -215,7 +219,12 @@ class AgentChatMessageApi(Resource):
     @with_current_tenant_id
     def post(self, current_tenant_id: str, current_user: Account, agent_id: UUID):
         app_model = resolve_agent_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
-        return _create_chat_message(current_user=current_user, app_model=app_model)
+        return _create_chat_message(
+            current_tenant_id=current_tenant_id,
+            current_user=current_user,
+            app_model=app_model,
+            agent_id=str(agent_id),
+        )
 
 
 @console_ns.route("/apps/<uuid:app_id>/chat-messages/<string:task_id>/stop")
@@ -249,11 +258,45 @@ class AgentChatMessageStopApi(Resource):
         return _stop_chat_message(current_user_id=current_user_id, app_model=app_model, task_id=task_id)
 
 
-def _create_chat_message(*, current_user: Account, app_model: App):
+def _resolve_current_user_agent_debug_conversation_id(
+    *, current_tenant_id: str, current_user: Account, app_model: App, agent_id: str | None
+) -> str:
+    roster_service = AgentRosterService(db.session)
+    if agent_id:
+        return roster_service.get_or_create_agent_app_debug_conversation_id(
+            tenant_id=current_tenant_id,
+            agent_id=agent_id,
+            account_id=current_user.id,
+        )
+
+    agent = roster_service.get_app_backing_agent(tenant_id=current_tenant_id, app_id=str(app_model.id))
+    if agent is None:
+        raise AgentNotFoundError()
+    return roster_service.get_or_create_agent_app_debug_conversation_id(
+        tenant_id=current_tenant_id,
+        agent_id=agent.id,
+        account_id=current_user.id,
+    )
+
+
+def _create_chat_message(
+    *, current_user: Account, app_model: App, current_tenant_id: str | None = None, agent_id: str | None = None
+):
     raw_payload = console_ns.payload or {}
     args_model = ChatMessagePayload.model_validate(raw_payload)
     args = args_model.model_dump(exclude_none=True, by_alias=True)
 
+    if AppMode.value_of(app_model.mode) == AppMode.AGENT:
+        debug_conversation_id = _resolve_current_user_agent_debug_conversation_id(
+            current_tenant_id=current_tenant_id or app_model.tenant_id,
+            current_user=current_user,
+            app_model=app_model,
+            agent_id=agent_id,
+        )
+        if args_model.conversation_id and args_model.conversation_id != debug_conversation_id:
+            raise NotFound("Conversation Not Exists.")
+        args["conversation_id"] = debug_conversation_id
+
     streaming = _resolve_debugger_chat_streaming(
         app_mode=AppMode.value_of(app_model.mode),
         response_mode=args_model.response_mode,

api/controllers/console/app/message.py

@@ -53,6 +53,7 @@ from libs.login import login_required
 from models.account import Account
 from models.enums import FeedbackFromSource, FeedbackRating
 from models.model import App, AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
+from services.conversation_service import ConversationService
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
 from services.message_service import MessageService, attach_message_extra_contents
@@ -186,10 +187,11 @@ class ChatMessageListApi(Resource):
     @account_initialization_required
     @setup_required
     @edit_permission_required
+    @with_current_user
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_VIEW_LAYOUT)
     @get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT, AppMode.AGENT])
-    def get(self, app_model: App):
-        return _list_chat_messages(app_model=app_model)
+    def get(self, current_user: Account, app_model: App):
+        return _list_chat_messages(app_model=app_model, current_user=current_user)
 
 
 @console_ns.route("/agent/<uuid:agent_id>/chat-messages")
@@ -205,10 +207,11 @@ class AgentChatMessageListApi(Resource):
     @setup_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_VIEW_LAYOUT)
+    @with_current_user
     @with_current_tenant_id
-    def get(self, current_tenant_id: str, agent_id: UUID):
+    def get(self, current_tenant_id: str, current_user: Account, agent_id: UUID):
         app_model = resolve_agent_app_model(tenant_id=current_tenant_id, agent_id=agent_id)
-        return _list_chat_messages(app_model=app_model)
+        return _list_chat_messages(app_model=app_model, current_user=current_user)
 
 
 @console_ns.route("/apps/<uuid:app_id>/feedbacks")
@@ -338,8 +341,8 @@ class MessageFeedbackExportApi(Resource):
 
         try:
             export_data = FeedbackService.export_feedbacks(
-                db.session(),
-                app_id=app_model.id,
+                app_model.id,
+                session=db.session(),
                 from_source=args.from_source,
                 rating=args.rating,
                 has_comment=args.has_comment,
@@ -390,14 +393,24 @@ class AgentMessageApi(Resource):
         return _get_message_detail(app_model=app_model, message_id=message_id)
 
 
-def _list_chat_messages(*, app_model: App):
+def _list_chat_messages(*, app_model: App, current_user: Account | None = None):
     args = ChatMessagesQuery.model_validate(request.args.to_dict())
 
-    conversation = db.session.scalar(
-        select(Conversation)
-        .where(Conversation.id == args.conversation_id, Conversation.app_id == app_model.id)
-        .limit(1)
-    )
+    if AppMode.value_of(app_model.mode) == AppMode.AGENT and current_user is not None:
+        try:
+            conversation = ConversationService.get_conversation(
+                app_model=app_model,
+                conversation_id=args.conversation_id,
+                user=current_user,
+            )
+        except ConversationNotExistsError:
+            raise NotFound("Conversation Not Exists.")
+    else:
+        conversation = db.session.scalar(
+            select(Conversation)
+            .where(Conversation.id == args.conversation_id, Conversation.app_id == app_model.id)
+            .limit(1)
+        )
 
     if not conversation:
         raise NotFound("Conversation Not Exists.")

api/controllers/console/app/ops_trace.py

@@ -13,6 +13,7 @@ from controllers.console.wraps import (
     RBACPermission,
     RBACResourceScope,
     account_initialization_required,
+    edit_permission_required,
     rbac_permission_required,
     setup_required,
 )
@@ -70,7 +71,7 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_MONITOR)
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
     def get(self, app_model: App):
         args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
@@ -95,9 +96,12 @@ class TraceAppConfigApi(Resource):
         console_ns.models[TraceAppConfigResponse.__name__],
     )
     @console_ns.response(400, "Invalid request parameters or configuration already exists")
+    @console_ns.response(403, "Insufficient permissions")
     @setup_required
     @login_required
     @account_initialization_required
+    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
     def post(self, app_model: App):
         """Create a new trace app configuration"""
@@ -125,9 +129,12 @@ class TraceAppConfigApi(Resource):
         console_ns.models[TraceAppConfigResponse.__name__],
     )
     @console_ns.response(400, "Invalid request parameters or configuration not found")
+    @console_ns.response(403, "Insufficient permissions")
     @setup_required
     @login_required
     @account_initialization_required
+    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
     def patch(self, app_model: App):
         """Update an existing trace app configuration"""
@@ -149,9 +156,12 @@ class TraceAppConfigApi(Resource):
     @console_ns.doc(params=query_params_from_model(TraceProviderQuery))
     @console_ns.response(204, "Tracing configuration deleted successfully")
     @console_ns.response(400, "Invalid request parameters or configuration not found")
+    @console_ns.response(403, "Insufficient permissions")
     @setup_required
     @login_required
     @account_initialization_required
+    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
     def delete(self, app_model: App):
         """Delete an existing trace app configuration"""

api/controllers/console/app/workflow_app_log.py

@@ -181,7 +181,7 @@ class WorkflowAppLogApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_MONITOR)
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_LOG_AND_ANNOTATION)
     @get_app_model(mode=[AppMode.WORKFLOW])
     def get(self, app_model: App):
         """
@@ -225,7 +225,7 @@ class WorkflowArchivedLogApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_MONITOR)
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_LOG_AND_ANNOTATION)
     @get_app_model(mode=[AppMode.WORKFLOW])
     def get(self, app_model: App):
         """

api/controllers/console/app/workflow_comment.py

@@ -17,6 +17,7 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user,
 )
+from extensions.ext_database import db
 from fields.base import ResponseModel
 from fields.member_fields import AccountWithRole
 from libs.helper import build_avatar_url, dump_response, to_timestamp
@@ -489,7 +490,7 @@ class WorkflowCommentMentionUsersApi(Resource):
         current_tenant = current_user.current_tenant  # need the tenant object here
         if current_tenant is None:
             raise ValueError("current tenant is required")
-        members = TenantService.get_tenant_members(current_tenant)
+        members = TenantService.get_tenant_members(current_tenant, session=db.session)
         users = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
         response = WorkflowCommentMentionUsersPayload(users=users)
         return response.model_dump(mode="json"), 200

api/controllers/console/app/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/activate.py

@@ -89,7 +89,9 @@ class ActivateCheckApi(Resource):
         workspaceId = args.workspace_id
         token = args.token
 
-        invitation = RegisterService.get_invitation_with_case_fallback(workspaceId, args.email, token)
+        invitation = RegisterService.get_invitation_with_case_fallback(
+            workspaceId, args.email, token, session=db.session
+        )
         if invitation:
             data = invitation.get("data", {})
             tenant = invitation.get("tenant", None)
@@ -137,7 +139,9 @@ class ActivateApi(Resource):
         args = ActivatePayload.model_validate(console_ns.payload)
 
         normalized_request_email = args.email.lower() if args.email else None
-        invitation = RegisterService.get_invitation_with_case_fallback(args.workspace_id, args.email, args.token)
+        invitation = RegisterService.get_invitation_with_case_fallback(
+            args.workspace_id, args.email, args.token, session=db.session
+        )
         if invitation is None:
             raise AlreadyActivateError()
 
@@ -184,6 +188,6 @@ class ActivateApi(Resource):
             account.status = AccountStatus.ACTIVE
             account.initialized_at = naive_utc_now()
 
-        TenantService.switch_tenant(account, tenant.id)
+        TenantService.switch_tenant(account, tenant.id, session=db.session)
 
         return {"result": "success"}

api/controllers/console/auth/data_source_bearer_auth.py

@@ -5,6 +5,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 extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.login import login_required
 from services.auth.api_key_auth_service import ApiKeyAuthService
@@ -58,7 +59,7 @@ class ApiKeyAuthDataSource(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def get(self, current_tenant_id: str):
-        data_source_api_key_bindings = ApiKeyAuthService.get_provider_auth_list(current_tenant_id)
+        data_source_api_key_bindings = ApiKeyAuthService.get_provider_auth_list(db.session(), current_tenant_id)
         if data_source_api_key_bindings:
             return {
                 "sources": [
@@ -83,7 +84,7 @@ class ApiKeyAuthDataSourceBinding(Resource):
     @login_required
     @account_initialization_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_CREATE, resource_required=False)
     @console_ns.expect(console_ns.models[ApiKeyAuthBindingPayload.__name__])
     @with_current_tenant_id
     def post(self, current_tenant_id: str):
@@ -92,7 +93,7 @@ class ApiKeyAuthDataSourceBinding(Resource):
         data = payload.model_dump()
         ApiKeyAuthService.validate_api_key_auth_args(data)
         try:
-            ApiKeyAuthService.create_provider_auth(current_tenant_id, data)
+            ApiKeyAuthService.create_provider_auth(db.session(), current_tenant_id, data)
         except Exception as e:
             raise ApiKeyAuthFailedError(str(e))
         return {"result": "success"}, 200
@@ -109,6 +110,6 @@ class ApiKeyAuthDataSourceBindingDelete(Resource):
     @with_current_tenant_id
     def delete(self, current_tenant_id: str, binding_id: UUID):
         # The role of the current user in the table must be admin or owner
-        ApiKeyAuthService.delete_provider_auth(current_tenant_id, str(binding_id))
+        ApiKeyAuthService.delete_provider_auth(db.session(), current_tenant_id, str(binding_id))
 
         return "", 204

api/controllers/console/auth/email_register.py

@@ -15,6 +15,7 @@ from controllers.console.auth.error import (
     InvalidTokenError,
     PasswordMismatchError,
 )
+from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.helper import EmailStr, extract_remote_ip
 from libs.helper import timezone as validate_timezone_string
@@ -22,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
 
 
@@ -100,7 +101,7 @@ class EmailRegisterSendEmailApi(Resource):
         if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(normalized_email):
             raise AccountInFreezeError()
 
-        account = AccountService.get_account_by_email_with_case_fallback(args.email)
+        account = AccountService.get_account_by_email_with_case_fallback(db.session, args.email)
         token = AccountService.send_email_register_email(email=normalized_email, account=account, language=language)
         return {"result": "success", "data": token}
 
@@ -175,7 +176,7 @@ class EmailRegisterResetApi(Resource):
         email = register_data.get("email", "")
         normalized_email = email.lower()
 
-        account = AccountService.get_account_by_email_with_case_fallback(email)
+        account = AccountService.get_account_by_email_with_case_fallback(db.session, email)
 
         if account:
             raise EmailAlreadyInUseError()
@@ -186,7 +187,7 @@ class EmailRegisterResetApi(Resource):
             timezone=args.timezone,
             language=args.language,
         )
-        token_pair = AccountService.login(account=account, ip_address=extract_remote_ip(request))
+        token_pair = AccountService.login(account=account, session=db.session, ip_address=extract_remote_ip(request))
         AccountService.reset_login_error_rate_limit(normalized_email)
 
         return {"result": "success", "data": token_pair.model_dump()}
@@ -205,6 +206,9 @@ class EmailRegisterResetApi(Resource):
                 password=password,
                 interface_language=get_valid_language(language),
                 timezone=timezone,
+                session=db.session,
             )
+        except SeatsLimitExceededError:
+            raise SeatsLimitExceeded()
         except AccountRegisterError:
             raise AccountInFreezeError()

api/controllers/console/auth/forgot_password.py

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

api/controllers/console/auth/login.py

@@ -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
 
@@ -119,7 +120,9 @@ class LoginApi(Resource):
         invite_token = args.invite_token
         invitation_data: InvitationDetailDict | None = None
         if invite_token:
-            invitation_data = RegisterService.get_invitation_with_case_fallback(None, request_email, invite_token)
+            invitation_data = RegisterService.get_invitation_with_case_fallback(
+                None, request_email, invite_token, session=db.session
+            )
             if invitation_data is None:
                 invite_token = None
 
@@ -145,7 +148,7 @@ class LoginApi(Resource):
             _log_console_login_failure(email=normalized_email, reason=LoginFailureReason.INVALID_CREDENTIALS)
             raise AuthenticationFailedError() from exc
         # SELF_HOSTED only have one workspace
-        tenants = TenantService.get_join_tenants(account)
+        tenants = TenantService.get_join_tenants(account, session=db.session)
         if len(tenants) == 0:
             system_features = FeatureService.get_system_features()
 
@@ -157,7 +160,7 @@ class LoginApi(Resource):
                     "data": "workspace not found, please contact system admin to invite you to join in a workspace",
                 }
 
-        token_pair = AccountService.login(account=account, ip_address=extract_remote_ip(request))
+        token_pair = AccountService.login(account=account, session=db.session, ip_address=extract_remote_ip(request))
         AccountService.reset_login_error_rate_limit(normalized_email)
 
         # Create response with cookies instead of returning tokens in body
@@ -291,7 +294,7 @@ class EmailCodeLoginApi(Resource):
             _log_console_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_IN_FREEZE)
             raise AccountInFreezeError()
         if account:
-            tenants = TenantService.get_join_tenants(account)
+            tenants = TenantService.get_join_tenants(account, session=db.session)
             if not tenants:
                 workspaces = FeatureService.get_system_features().license.workspaces
                 if not workspaces.is_available():
@@ -299,7 +302,7 @@ class EmailCodeLoginApi(Resource):
                 if not FeatureService.get_system_features().is_allow_create_workspace:
                     raise NotAllowedCreateWorkspace()
                 else:
-                    new_tenant = TenantService.create_tenant(f"{account.name}'s Workspace")
+                    new_tenant = TenantService.create_tenant(f"{account.name}'s Workspace", session=db.session)
                     TenantService.create_tenant_member(new_tenant, account, db.session, role="owner")
                     account.current_tenant = new_tenant
                     tenant_was_created.send(new_tenant)
@@ -311,15 +314,18 @@ class EmailCodeLoginApi(Resource):
                     name=user_email,
                     interface_language=get_valid_language(language),
                     timezone=args.timezone,
+                    session=db.session,
                 )
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()
+            except SeatsLimitExceededError:
+                raise SeatsLimitExceeded()
             except AccountRegisterError:
                 _log_console_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_IN_FREEZE)
                 raise AccountInFreezeError()
             except WorkspacesLimitExceededError:
                 raise WorkspacesLimitExceeded()
-        token_pair = AccountService.login(account, ip_address=extract_remote_ip(request))
+        token_pair = AccountService.login(account, session=db.session, ip_address=extract_remote_ip(request))
         AccountService.reset_login_error_rate_limit(user_email)
 
         # Create response with cookies instead of returning tokens in body
@@ -343,7 +349,7 @@ class RefreshTokenApi(Resource):
             return {"result": "fail", "message": "No refresh token provided"}, 401
 
         try:
-            new_token_pair = AccountService.refresh_token(refresh_token)
+            new_token_pair = AccountService.refresh_token(refresh_token, session=db.session)
 
             # Create response with new cookies
             response = make_response({"result": "success"})
@@ -358,22 +364,22 @@ class RefreshTokenApi(Resource):
 
 
 def _get_account_with_case_fallback(email: str):
-    account = AccountService.get_user_through_email(email)
+    account = AccountService.get_user_through_email(email, session=db.session)
     if account or email == email.lower():
         return account
 
-    return AccountService.get_user_through_email(email.lower())
+    return AccountService.get_user_through_email(email.lower(), session=db.session)
 
 
 def _authenticate_account_with_case_fallback(
     original_email: str, normalized_email: str, password: str, invite_token: str | None
 ):
     try:
-        return AccountService.authenticate(original_email, password, invite_token)
+        return AccountService.authenticate(original_email, password, invite_token, session=db.session)
     except services.errors.account.AccountPasswordError:
         if original_email == normalized_email:
             raise
-        return AccountService.authenticate(normalized_email, password, invite_token)
+        return AccountService.authenticate(normalized_email, password, invite_token, session=db.session)
 
 
 def _log_console_login_failure(*, email: str, reason: LoginFailureReason) -> None:

api/controllers/console/auth/oauth.py

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

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,

api/controllers/console/datasets/external.py

@@ -26,6 +26,7 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user,
 )
+from extensions.ext_database import db
 from fields.base import ResponseModel
 from fields.dataset_fields import (
     dataset_detail_fields,
@@ -390,6 +391,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
 
         try:
             response = HitTestingService.external_retrieve(
+                session=db.session,
                 dataset=dataset,
                 query=payload.query,
                 account=current_user,

api/controllers/console/datasets/hit_testing_base.py

@@ -18,6 +18,7 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
+from extensions.ext_database import db
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs.login import resolve_account_fallback
 from models.account import Account
@@ -115,6 +116,7 @@ class DatasetsHitTestingBase:
         try:
             current_user, _ = resolve_account_fallback(current_user, current_tenant_id)
             response = HitTestingService.retrieve(
+                session=db.session,
                 dataset=dataset,
                 query=cast(str, args.get("query")),
                 account=current_user,

api/controllers/console/datasets/metadata.py

@@ -17,6 +17,7 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user,
 )
+from extensions.ext_database import db
 from fields.dataset_fields import (
     DatasetMetadataBuiltInFieldsResponse,
     DatasetMetadataListResponse,
@@ -65,7 +66,9 @@ class DatasetMetadataCreateApi(Resource):
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user)
 
-        metadata = MetadataService.create_metadata(dataset_id_str, metadata_args, current_user, current_tenant_id)
+        metadata = MetadataService.create_metadata(
+            db.session(), dataset_id_str, metadata_args, current_user, current_tenant_id
+        )
         return dump_response(DatasetMetadataResponse, metadata), 201
 
     @setup_required
@@ -81,7 +84,7 @@ class DatasetMetadataCreateApi(Resource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        metadata = MetadataService.get_dataset_metadatas(dataset)
+        metadata = MetadataService.get_dataset_metadatas(db.session(), dataset)
         return dump_response(DatasetMetadataListResponse, metadata), 200
 
 
@@ -108,7 +111,7 @@ class DatasetMetadataApi(Resource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata = MetadataService.update_metadata_name(
-            dataset_id_str, metadata_id_str, name, current_user, current_tenant_id
+            db.session(), dataset_id_str, metadata_id_str, name, current_user, current_tenant_id
         )
         return dump_response(DatasetMetadataResponse, metadata), 200
 
@@ -127,7 +130,7 @@ class DatasetMetadataApi(Resource):
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user)
 
-        MetadataService.delete_metadata(dataset_id_str, metadata_id_str)
+        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.
         return "", 204
 
@@ -166,9 +169,9 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
 
         match action:
             case "enable":
-                MetadataService.enable_built_in_field(dataset)
+                MetadataService.enable_built_in_field(db.session(), dataset)
             case "disable":
-                MetadataService.disable_built_in_field(dataset)
+                MetadataService.disable_built_in_field(db.session(), dataset)
         # Frontend callers only await success and invalidate metadata caches; no response body is consumed.
         return "", 204
 
@@ -195,7 +198,7 @@ class DocumentMetadataEditApi(Resource):
 
         metadata_args = MetadataOperationData.model_validate(console_ns.payload or {})
 
-        MetadataService.update_documents_metadata(dataset, metadata_args, current_user)
+        MetadataService.update_documents_metadata(db.session(), dataset, metadata_args, current_user)
 
         # Frontend callers only await success and invalidate caches; no response body is consumed.
         return "", 204

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

@@ -222,7 +222,7 @@ class DatasourceAuth(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.CREDENTIAL_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.CREDENTIAL_CREATE, resource_required=False)
     @with_current_tenant_id
     def post(self, current_tenant_id: str, provider_id: str):
         payload = DatasourceCredentialPayload.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/audio.py

@@ -20,6 +20,7 @@ from controllers.console.app.error import (
 )
 from controllers.console.explore.wraps import InstalledAppResource
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
+from extensions.ext_database import db
 from graphon.model_runtime.errors.invoke import InvokeError
 from models.model import InstalledApp
 from services.audio_service import AudioService
@@ -99,7 +100,13 @@ class ChatTextApi(InstalledAppResource):
             text = payload.text
             voice = payload.voice
 
-            response = AudioService.transcript_tts(app_model=app_model, text=text, voice=voice, message_id=message_id)
+            response = AudioService.transcript_tts(
+                app_model=app_model,
+                session=db.session,
+                text=text,
+                voice=voice,
+                message_id=message_id,
+            )
             return response
         except services.errors.app_model_config.AppModelConfigBrokenError:
             logger.exception("App model config broken.")

api/controllers/console/explore/installed_app.py

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

api/controllers/console/explore/message.py

@@ -27,7 +27,11 @@ from controllers.console.wraps import with_current_user
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from fields.conversation_fields import ResultResponse
-from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem, SuggestedQuestionsResponse
+from fields.message_fields import (
+    ExploreMessageInfiniteScrollPagination,
+    ExploreMessageListItem,
+    SuggestedQuestionsResponse,
+)
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from models import Account
@@ -56,7 +60,7 @@ register_schema_models(console_ns, MessageListQuery, MessageFeedbackPayload, Mor
 register_response_schema_models(
     console_ns,
     GeneratedAppResponse,
-    MessageInfiniteScrollPagination,
+    ExploreMessageInfiniteScrollPagination,
     ResultResponse,
     SuggestedQuestionsResponse,
 )
@@ -68,7 +72,7 @@ register_response_schema_models(
 )
 class MessageListApi(InstalledAppResource):
     @console_ns.doc(params=query_params_from_model(MessageListQuery))
-    @console_ns.response(200, "Success", console_ns.models[MessageInfiniteScrollPagination.__name__])
+    @console_ns.response(200, "Success", console_ns.models[ExploreMessageInfiniteScrollPagination.__name__])
     @with_current_user
     def get(self, current_user: Account, installed_app: InstalledApp):
         app_model = installed_app.app
@@ -88,9 +92,9 @@ class MessageListApi(InstalledAppResource):
                 str(args.first_id) if args.first_id else None,
                 args.limit,
             )
-            adapter = TypeAdapter(MessageListItem)
+            adapter = TypeAdapter(ExploreMessageListItem)
             items = [adapter.validate_python(message, from_attributes=True) for message in pagination.data]
-            return MessageInfiniteScrollPagination(
+            return ExploreMessageInfiniteScrollPagination(
                 limit=pagination.limit,
                 has_more=pagination.has_more,
                 data=items,

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__])
@@ -419,7 +452,13 @@ class TrialChatTextApi(TrialAppResource):
             app_id = app_model.id
             user_id = current_user.id
 
-            response = AudioService.transcript_tts(app_model=app_model, text=text, voice=voice, message_id=message_id)
+            response = AudioService.transcript_tts(
+                app_model=app_model,
+                session=db.session,
+                text=text,
+                voice=voice,
+                message_id=message_id,
+            )
             RecommendedAppService.add_trial_app_record(db.session, app_id, user_id)
             return response
         except services.errors.app_model_config.AppModelConfigBrokenError:
@@ -599,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/init_validate.py

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

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/setup.py

@@ -79,7 +79,7 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
     if get_setup_status():
         raise AlreadySetupError()
 
-    tenant_count = TenantService.get_tenant_count()
+    tenant_count = TenantService.get_tenant_count(session=db.session)
     if tenant_count > 0:
         raise AlreadySetupError()
 
@@ -94,6 +94,7 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
         password=payload.password,
         ip_address=extract_remote_ip(request),
         language=payload.language,
+        session=db.session,
     )
 
     return SetupResponse(result="success")

api/controllers/console/snippets/snippet_workflow.py

@@ -80,6 +80,13 @@ class SnippetDraftConfigResponse(BaseModel):
     parallel_depth_limit: int
 
 
+class SnippetWorkflowPaginationResponse(BaseModel):
+    items: list[SnippetWorkflowResponse]
+    page: int
+    limit: int
+    has_more: bool
+
+
 register_schema_models(
     console_ns,
     SnippetDraftSyncPayload,
@@ -98,6 +105,7 @@ register_response_schema_models(
     SimpleResultResponse,
     SnippetDraftConfigResponse,
     SnippetWorkflowResponse,
+    SnippetWorkflowPaginationResponse,
     WorkflowPublishResponse,
     WorkflowPaginationResponse,
     WorkflowRestoreResponse,
@@ -329,7 +337,7 @@ class SnippetPublishedAllWorkflowApi(Resource):
     @console_ns.response(
         200,
         "Published workflows retrieved successfully",
-        console_ns.models[WorkflowPaginationResponse.__name__],
+        console_ns.models[SnippetWorkflowPaginationResponse.__name__],
     )
     @setup_required
     @login_required
@@ -350,7 +358,7 @@ class SnippetPublishedAllWorkflowApi(Resource):
                 limit=args.limit,
             )
 
-        return WorkflowPaginationResponse.model_validate(
+        response = SnippetWorkflowPaginationResponse.model_validate(
             {
                 "items": workflows,
                 "page": args.page,
@@ -359,6 +367,9 @@ class SnippetPublishedAllWorkflowApi(Resource):
             },
             from_attributes=True,
         ).model_dump(mode="json")
+        for item in response["items"]:
+            item["input_fields"] = snippet.input_fields_list
+        return response
 
 
 @console_ns.route("/snippets/<uuid:snippet_id>/workflows/<string:workflow_id>/restore")

api/controllers/console/socketio/workflow.py

@@ -4,6 +4,7 @@ from typing import cast
 
 from flask import Request as FlaskRequest
 
+from extensions.ext_database import db
 from extensions.ext_socketio import sio
 from libs.passport import PassportService
 from libs.token import extract_access_token
@@ -43,7 +44,7 @@ def socket_connect(sid, environ, auth):
             return False
 
         with sio.app.app_context():
-            user = AccountService.load_logged_in_account(account_id=user_id)
+            user = AccountService.load_logged_in_account(account_id=user_id, session=db.session)
             if not user:
                 logging.warning("Socket connect rejected: user not found (user_id=%s, sid=%s)", user_id, sid)
                 return False

api/controllers/console/workspace/__init__.py

@@ -5,6 +5,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden
 
+from configs import dify_config
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models.account import TenantPluginPermission
@@ -17,6 +18,9 @@ def plugin_permission_required(
     def interceptor[**P, R](view: Callable[P, R]) -> Callable[P, R]:
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+            if dify_config.RBAC_ENABLED:
+                return view(*args, **kwargs)
+
             current_user, current_tenant_id = current_account_with_tenant()
             user = current_user
             tenant_id = current_tenant_id

api/controllers/console/workspace/account.py

@@ -328,7 +328,7 @@ class AccountNameApi(Resource):
     def post(self, current_user: Account):
         payload = console_ns.payload or {}
         args = AccountNamePayload.model_validate(payload)
-        updated_account = AccountService.update_account(current_user, name=args.name)
+        updated_account = AccountService.update_account(current_user, session=db.session, name=args.name)
 
         return _serialize_account(updated_account)
 
@@ -374,7 +374,7 @@ class AccountAvatarApi(Resource):
         payload = console_ns.payload or {}
         args = AccountAvatarPayload.model_validate(payload)
 
-        updated_account = AccountService.update_account(current_user, avatar=args.avatar)
+        updated_account = AccountService.update_account(current_user, session=db.session, avatar=args.avatar)
 
         return _serialize_account(updated_account)
 
@@ -391,7 +391,9 @@ class AccountInterfaceLanguageApi(Resource):
         payload = console_ns.payload or {}
         args = AccountInterfaceLanguagePayload.model_validate(payload)
 
-        updated_account = AccountService.update_account(current_user, interface_language=args.interface_language)
+        updated_account = AccountService.update_account(
+            current_user, session=db.session, interface_language=args.interface_language
+        )
 
         return _serialize_account(updated_account)
 
@@ -408,7 +410,9 @@ class AccountInterfaceThemeApi(Resource):
         payload = console_ns.payload or {}
         args = AccountInterfaceThemePayload.model_validate(payload)
 
-        updated_account = AccountService.update_account(current_user, interface_theme=args.interface_theme)
+        updated_account = AccountService.update_account(
+            current_user, session=db.session, interface_theme=args.interface_theme
+        )
 
         return _serialize_account(updated_account)
 
@@ -425,7 +429,7 @@ class AccountTimezoneApi(Resource):
         payload = console_ns.payload or {}
         args = AccountTimezonePayload.model_validate(payload)
 
-        updated_account = AccountService.update_account(current_user, timezone=args.timezone)
+        updated_account = AccountService.update_account(current_user, session=db.session, timezone=args.timezone)
 
         return _serialize_account(updated_account)
 
@@ -443,7 +447,8 @@ class AccountPasswordApi(Resource):
         args = AccountPasswordPayload.model_validate(payload)
 
         try:
-            AccountService.update_account_password(current_user, args.password, args.new_password)
+            assert args.password is not None
+            AccountService.update_account_password(current_user, args.password, args.new_password, session=db.session)
         except ServiceCurrentPasswordIncorrectError:
             raise CurrentPasswordIncorrectError()
 
@@ -731,7 +736,7 @@ class ChangeEmailResetApi(Resource):
         if AccountService.is_account_in_freeze(normalized_new_email):
             raise AccountInFreezeError()
 
-        if not AccountService.check_email_unique(normalized_new_email):
+        if not AccountService.check_email_unique(normalized_new_email, session=db.session):
             raise EmailAlreadyInUseError()
 
         reset_data = AccountService.get_change_email_data(args.token)
@@ -755,7 +760,9 @@ class ChangeEmailResetApi(Resource):
         # legitimately verified token.
         AccountService.revoke_change_email_token(args.token)
 
-        updated_account = AccountService.update_account_email(current_user, email=normalized_new_email)
+        updated_account = AccountService.update_account_email(
+            current_user, email=normalized_new_email, session=db.session
+        )
 
         AccountService.send_change_email_completed_notify_email(
             email=normalized_new_email,
@@ -775,6 +782,6 @@ class CheckEmailUnique(Resource):
         normalized_email = args.email.lower()
         if AccountService.is_account_in_freeze(normalized_email):
             raise AccountInFreezeError()
-        if not AccountService.check_email_unique(normalized_email):
+        if not AccountService.check_email_unique(normalized_email, session=db.session):
             raise EmailAlreadyInUseError()
         return {"result": "success"}

api/controllers/console/workspace/endpoint.py

@@ -169,7 +169,7 @@ class EndpointCollectionApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @account_initialization_required
     @with_current_user_id
     @with_current_tenant_id
@@ -198,7 +198,7 @@ class DeprecatedEndpointCreateApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @account_initialization_required
     @with_current_user_id
     @with_current_tenant_id
@@ -290,7 +290,7 @@ class EndpointItemApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @account_initialization_required
     @with_current_user_id
     @with_current_tenant_id
@@ -310,7 +310,7 @@ class EndpointItemApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @account_initialization_required
     @with_current_user_id
     @with_current_tenant_id
@@ -340,7 +340,7 @@ class DeprecatedEndpointDeleteApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @account_initialization_required
     @with_current_user_id
     @with_current_tenant_id
@@ -371,7 +371,7 @@ class DeprecatedEndpointUpdateApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @account_initialization_required
     @with_current_user_id
     @with_current_tenant_id
@@ -394,7 +394,7 @@ class EndpointEnableApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @account_initialization_required
     @with_current_user_id
     @with_current_tenant_id
@@ -422,7 +422,7 @@ class EndpointDisableApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @account_initialization_required
     @with_current_user_id
     @with_current_tenant_id

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
 
 
@@ -131,7 +131,7 @@ def _normalize_invitee_emails(emails: list[str]) -> list[str]:
 def _count_new_member_invites(tenant_id: str, emails: list[str]) -> int:
     new_member_count = 0
     for email in emails:
-        account = AccountService.get_account_by_email_with_case_fallback(email)
+        account = AccountService.get_account_by_email_with_case_fallback(db.session, email)
         if not account:
             new_member_count += 1
             continue
@@ -186,7 +186,7 @@ class MemberListApi(Resource):
             current_user, _ = current_account_with_tenant()
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
-        members = TenantService.get_tenant_members(current_user.current_tenant)
+        members = TenantService.get_tenant_members(current_user.current_tenant, session=db.session)
         if dify_config.RBAC_ENABLED:
             member_ids = [member.id for member in members]
             member_roles = enterprise_rbac_service.RBACService.MemberRoles.batch_get(
@@ -273,6 +273,7 @@ class MemberInviteEmailApi(Resource):
                         language=interface_language,
                         role=invitee_role,
                         inviter=inviter,
+                        session=db.session,
                     )
                     encoded_invitee_email = parse.quote(invitee_email)
                     invitation_results.append(
@@ -290,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)})
 
@@ -317,7 +326,9 @@ class MemberCancelInviteApi(Resource):
             abort(404)
         else:
             try:
-                TenantService.remove_member_from_tenant(current_user.current_tenant, member, current_user)
+                TenantService.remove_member_from_tenant(
+                    current_user.current_tenant, member, current_user, session=db.session
+                )
             except services.errors.account.CannotOperateSelfError as e:
                 return {"code": "cannot-operate-self", "message": str(e)}, 400
             except services.errors.account.NoPermissionError as e:
@@ -360,7 +371,9 @@ class MemberUpdateRoleApi(Resource):
 
         try:
             assert member is not None, "Member not found"
-            TenantService.update_member_role(current_user.current_tenant, member, new_role, current_user)
+            TenantService.update_member_role(
+                current_user.current_tenant, member, new_role, current_user, session=db.session
+            )
         except services.errors.account.CannotOperateSelfError as e:
             return {"code": "cannot-operate-self", "message": str(e)}, 400
         except services.errors.account.NoPermissionError as e:
@@ -387,7 +400,7 @@ class DatasetOperatorMemberListApi(Resource):
     def get(self, current_user: Account):
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
-        members = TenantService.get_dataset_operator_members(current_user.current_tenant)
+        members = TenantService.get_dataset_operator_members(current_user.current_tenant, session=db.session)
         member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
         response = AccountWithRoleList(accounts=member_models)
         return response.model_dump(mode="json"), 200
@@ -413,7 +426,7 @@ class SendOwnerTransferEmailApi(Resource):
         # check if the current user is the owner of the workspace
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
-        if not TenantService.is_owner(current_user, current_user.current_tenant):
+        if not TenantService.is_owner(current_user, current_user.current_tenant, session=db.session):
             raise NotOwnerError()
 
         if args.language is not None and args.language == "zh-Hans":
@@ -448,7 +461,7 @@ class OwnerTransferCheckApi(Resource):
         # check if the current user is the owner of the workspace
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
-        if not TenantService.is_owner(current_user, current_user.current_tenant):
+        if not TenantService.is_owner(current_user, current_user.current_tenant, session=db.session):
             raise NotOwnerError()
 
         user_email = current_user.email
@@ -494,7 +507,7 @@ class OwnerTransfer(Resource):
         # check if the current user is the owner of the workspace
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
-        if not TenantService.is_owner(current_user, current_user.current_tenant):
+        if not TenantService.is_owner(current_user, current_user.current_tenant, session=db.session):
             raise NotOwnerError()
 
         if current_user.id == str(member_id):
@@ -516,12 +529,14 @@ class OwnerTransfer(Resource):
 
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
-        if not TenantService.is_member(member, current_user.current_tenant):
+        if not TenantService.is_member(member, current_user.current_tenant, session=db.session):
             raise MemberNotInTenantError()
 
         try:
             assert member is not None, "Member not found"
-            TenantService.update_member_role(current_user.current_tenant, member, "owner", current_user)
+            TenantService.update_member_role(
+                current_user.current_tenant, member, "owner", current_user, session=db.session
+            )
 
             AccountService.send_new_owner_transfer_notify_email(
                 account=member,

api/controllers/console/workspace/model_providers.py

@@ -169,7 +169,7 @@ class ModelProviderCredentialApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_CREATE, resource_required=False)
     @account_initialization_required
     @with_current_tenant_id
     def post(self, current_tenant_id: str, provider: str):
@@ -244,7 +244,7 @@ class ModelProviderCredentialSwitchApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_USE, resource_required=False)
     @account_initialization_required
     @with_current_tenant_id
     def post(self, current_tenant_id: str, provider: str):
@@ -326,7 +326,7 @@ class PreferredProviderTypeUpdateApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_USE, resource_required=False)
     @account_initialization_required
     @with_current_tenant_id
     def post(self, tenant_id: str, provider: str):

api/controllers/console/workspace/models.py

@@ -395,7 +395,7 @@ class ModelProviderModelCredentialApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_CREATE, resource_required=False)
     @account_initialization_required
     @with_current_tenant_id
     def post(self, tenant_id: str, provider: str):
@@ -481,7 +481,7 @@ class ModelProviderModelCredentialSwitchApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_USE, resource_required=False)
     @account_initialization_required
     @with_current_tenant_id
     def post(self, current_tenant_id: str, provider: str):

api/controllers/console/workspace/plugin.py

@@ -469,6 +469,7 @@ class PluginDebuggingKeyApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_DEBUG, resource_required=False)
     @plugin_permission_required(debug_required=True)
     @with_current_tenant_id
     def get(self, tenant_id: str):
@@ -614,6 +615,7 @@ class PluginUploadFromPkgApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -634,6 +636,7 @@ class PluginUploadFromGithubApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -653,6 +656,7 @@ class PluginUploadFromBundleApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -673,6 +677,7 @@ class PluginInstallFromPkgApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -693,6 +698,7 @@ class PluginInstallFromGithubApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -719,6 +725,7 @@ class PluginInstallFromMarketplaceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -739,6 +746,7 @@ class PluginFetchMarketplacePkgApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def get(self, tenant_id: str):
@@ -764,6 +772,7 @@ class PluginFetchManifestApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def get(self, tenant_id: str):
@@ -862,6 +871,7 @@ class PluginUpgradeFromMarketplaceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -884,6 +894,7 @@ class PluginUpgradeFromGithubApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -911,6 +922,7 @@ class PluginUninstallApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_DELETE, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -978,7 +990,7 @@ class PluginFetchDynamicSelectOptionsApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @account_initialization_required
     @with_current_user
     @with_current_tenant_id
@@ -1041,10 +1053,11 @@ class PluginChangeAutoUpgradeApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_PREFERENCES, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     def post(self, tenant_id: str, user: Account):
-        if not user.is_admin_or_owner:
+        if not dify_config.RBAC_ENABLED and not user.is_admin_or_owner:
             raise Forbidden()
 
         args = ParserAutoUpgradeChange.model_validate(console_ns.payload)
@@ -1097,6 +1110,7 @@ class PluginAutoUpgradeExcludePluginApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_PREFERENCES, resource_required=False)
     @with_current_tenant_id
     def post(self, tenant_id: str):
         # exclude one single plugin

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]):
@@ -201,21 +202,23 @@ def _legacy_workspace_roles(
     This keeps the new `/rbac/roles` endpoint compatible with the original
     Dify role model when enterprise RBAC is disabled.
     """
-
-    legacy_roles = [
-        svc.RBACRole(
-            id=role_name,
-            tenant_id="",
-            type=svc.RBACRoleType.WORKSPACE.value,
-            category="global_system_default",
-            name=role_name,
-            description="",
-            is_builtin=True,
-            permission_keys=list(_LEGACY_ROLE_PERMISSION_KEYS[role_name]),
-            role_tag="owner" if role_name == "owner" else "",
+    legacy_roles = []
+    for role_name in ("owner", "admin", "editor", "normal", "dataset_operator"):
+        if not dify_config.DATASET_OPERATOR_ENABLED and role_name == "dataset_operator":
+            continue
+        legacy_roles.append(
+            svc.RBACRole(
+                id=role_name,
+                tenant_id="",
+                type=svc.RBACRoleType.WORKSPACE.value,
+                category="global_system_default",
+                name=role_name,
+                description="",
+                is_builtin=True,
+                permission_keys=list(dict.fromkeys(_LEGACY_ROLE_PERMISSION_KEYS[role_name])),
+                role_tag="owner" if role_name == "owner" else "",
+            )
         )
-        for role_name in ("owner", "admin", "editor", "normal", "dataset_operator")
-    ]
 
     if not include_owner:
         legacy_roles = [r for r in legacy_roles if r.name != "owner"]
@@ -244,11 +247,6 @@ def _legacy_workspace_roles(
     )
 
 
-# ---------------------------------------------------------------------------
-# Permission catalogs.
-# ---------------------------------------------------------------------------
-
-
 @console_ns.route("/workspaces/current/rbac/role-permissions/catalog")
 class RBACWorkspaceCatalogApi(Resource):
     @login_required
@@ -375,30 +373,6 @@ class RBACRoleCopyApi(Resource):
         return _dump(role), 201
 
 
-@console_ns.route("/workspaces/current/rbac/roles/<uuid:role_id>/members")
-class RBACRoleMembersApi(Resource):
-    @login_required
-    @rbac_permission_required(
-        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
-    )
-    @console_ns.response(200, "Success", console_ns.models[_RBACRoleAccountList.__name__])
-    def get(self, role_id):
-        tenant_id, account_id = _current_ids()
-        return _dump(
-            svc.RBACService.Roles.members(
-                tenant_id,
-                account_id,
-                str(role_id),
-                options=_pagination_options(),
-            )
-        )
-
-
-# ---------------------------------------------------------------------------
-# Access policies (tenant-level permission sets).
-# ---------------------------------------------------------------------------
-
-
 class _AccessPolicyCreateRequest(BaseModel):
     name: str
     resource_type: svc.RBACResourceType
@@ -538,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):
@@ -611,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")
@@ -643,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,
             )
         )
@@ -788,11 +757,6 @@ class RBACDatasetMemberBindingsApi(Resource):
         return {"result": "success"}
 
 
-# ---------------------------------------------------------------------------
-# Workspace-level access (Settings > Access Rules).
-# ---------------------------------------------------------------------------
-
-
 @console_ns.route("/workspaces/current/rbac/workspace/apps/access-policy")
 class RBACWorkspaceAppMatrixApi(Resource):
     @login_required

api/controllers/console/workspace/tool_providers.py

@@ -971,7 +971,7 @@ class ToolBuiltinProviderSetDefaultApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_USE, resource_required=False)
     @account_initialization_required
     @with_current_tenant_id
     def post(self, current_tenant_id: str, provider: str):
@@ -1070,6 +1070,7 @@ class ToolProviderMCPApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.MCP_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     def post(self, tenant_id: str, user: Account):
@@ -1125,6 +1126,7 @@ class ToolProviderMCPApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.MCP_MANAGE, resource_required=False)
     @with_current_tenant_id
     def put(self, current_tenant_id: str):
         payload = MCPProviderUpdatePayload.model_validate(console_ns.payload or {})
@@ -1178,6 +1180,7 @@ class ToolProviderMCPApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.MCP_MANAGE, resource_required=False)
     @with_current_tenant_id
     def delete(self, current_tenant_id: str):
         payload = MCPProviderDeletePayload.model_validate(console_ns.payload or {})
@@ -1196,6 +1199,7 @@ class ToolMCPAuthApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.MCP_MANAGE, resource_required=False)
     @with_current_tenant_id
     def post(self, tenant_id: str):
         payload = MCPAuthPayload.model_validate(console_ns.payload or {})
@@ -1300,6 +1304,7 @@ class ToolMCPUpdateApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.MCP_MANAGE, resource_required=False)
     @with_current_tenant_id
     def get(self, tenant_id: str, provider_id: str):
         with sessionmaker(db.engine).begin() as session:

api/controllers/console/workspace/workspace.py

@@ -325,10 +325,10 @@ class TenantApi(Resource):
             raise ValueError("No current tenant")
 
         if tenant.status == TenantStatus.ARCHIVE:
-            tenants = TenantService.get_join_tenants(current_user)
+            tenants = TenantService.get_join_tenants(current_user, session=db.session)
             # if there is any tenant, switch to the first one
             if len(tenants) > 0:
-                TenantService.switch_tenant(current_user, tenants[0].id)
+                TenantService.switch_tenant(current_user, tenants[0].id, session=db.session)
                 tenant = tenants[0]
             # else, raise Unauthorized
             else:
@@ -351,7 +351,7 @@ class SwitchWorkspaceApi(Resource):
 
         # check if tenant_id is valid, 403 if not
         try:
-            TenantService.switch_tenant(current_user, args.tenant_id)
+            TenantService.switch_tenant(current_user, args.tenant_id, session=db.session)
         except Exception:
             raise AccountNotLinkTenantError("Account not link tenant")
 

api/controllers/inner_api/knowledge/retrieval.py

@@ -1,9 +1,10 @@
-"""Inner API endpoint for tenant-scoped knowledge retrieval.
+"""Plugin inner API endpoint for tenant-scoped knowledge retrieval.
 
 This controller is a thin HTTP wrapper around
 ``services.knowledge_retrieval_inner_service.InnerKnowledgeRetrievalService``.
-It intentionally keeps authorization simple: shared inner API key plus
-tenant-scoped app/dataset validation in the service layer.
+It uses the plugin inner API key because dify-agent calls this endpoint through
+the same trusted Dify API bridge as other agent/plugin inner calls; tenant-scoped
+app/dataset validation remains in the service layer.
 """
 
 from flask_restx import Resource
@@ -11,8 +12,9 @@ from pydantic import ValidationError
 
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.inner_api import inner_api_ns
-from controllers.inner_api.wraps import inner_api_only
+from controllers.inner_api.wraps import plugin_inner_api_only
 from core.workflow.nodes.knowledge_retrieval import exc as retrieval_exc
+from extensions.ext_database import db
 from libs.exception import BaseHTTPException
 from services.entities.knowledge_retrieval_inner import InnerKnowledgeRetrieveRequest, InnerKnowledgeRetrieveResponse
 from services.errors.knowledge_retrieval import ExternalKnowledgeRetrievalError, InnerKnowledgeRetrievalServiceError
@@ -48,7 +50,7 @@ register_response_schema_models(inner_api_ns, InnerKnowledgeRetrieveResponse)
 class InnerKnowledgeRetrieveApi(Resource):
     """Retrieve knowledge from one or more datasets within the caller tenant."""
 
-    @inner_api_only
+    @plugin_inner_api_only
     @inner_api_ns.doc("inner_knowledge_retrieve")
     @inner_api_ns.doc(description="Retrieve knowledge for trusted internal callers")
     @inner_api_ns.expect(inner_api_ns.models[InnerKnowledgeRetrieveRequest.__name__])
@@ -60,9 +62,8 @@ class InnerKnowledgeRetrieveApi(Resource):
     @inner_api_ns.doc(
         responses={
             400: "Invalid request body",
-            401: "Unauthorized - invalid inner API key",
             403: "Caller tenant does not own the requested resource",
-            404: "App or dataset not found",
+            404: "Invalid plugin inner API key, app not found, or dataset not found",
             422: "Invalid retrieval configuration",
             429: "Knowledge retrieval rate limited",
             502: "External knowledge retrieval failed",
@@ -81,7 +82,7 @@ class InnerKnowledgeRetrieveApi(Resource):
             ) from exc
 
         try:
-            response = InnerKnowledgeRetrievalService().retrieve(payload)
+            response = InnerKnowledgeRetrievalService().retrieve(payload, session=db.session)
         except InnerKnowledgeRetrievalServiceError as exc:
             raise InnerKnowledgeRetrievalHttpError(
                 error_code=exc.error_code,

api/controllers/inner_api/plugin/agent_drive.py

@@ -1,10 +1,12 @@
-"""Inner API for the agent drive (agent 网盘) control plane — ENG-591.
+"""Inner API for the agent drive (agent 网盘) control plane.
 
-Two endpoints, called by the dify-agent server (not the sandbox) with the inner
-API key. The drive ref is the URL segment ``agent-<agent_id>``; the path-like
-file key travels in the query/body, never as a URL path segment (so its ``/``
-characters do not collide with routing). Drive-owned semantics: tenant scoped,
-no user-level FileAccessScope.
+These endpoints are called by the dify-agent server (not the sandbox) with the
+inner API key. The drive ref is the URL segment ``agent-<agent_id>``; the
+path-like file key travels in the query/body, never as a URL path segment (so
+its ``/`` characters do not collide with routing). Drive-owned semantics:
+tenant scoped, no user-level FileAccessScope. Commit still canonicalizes the
+trusted execution-context user through the same EndUser lookup as plugin file
+upload before validating ToolFile ownership.
 """
 
 from flask import request
@@ -13,6 +15,7 @@ from pydantic import BaseModel, ValidationError
 
 from controllers.console.wraps import setup_required
 from controllers.inner_api import inner_api_ns
+from controllers.inner_api.plugin.wraps import get_user
 from controllers.inner_api.wraps import plugin_inner_api_only
 from services.agent_drive_service import (
     AgentDriveError,
@@ -56,6 +59,24 @@ class AgentDriveManifestApi(Resource):
         return {"items": items}
 
 
+@inner_api_ns.route("/drive/<string:drive_ref>/skills")
+class AgentDriveSkillsApi(Resource):
+    @setup_required
+    @plugin_inner_api_only
+    @inner_api_ns.doc("agent_drive_skills")
+    @inner_api_ns.doc(description="List the skill catalog of an agent drive")
+    def get(self, drive_ref: str):
+        try:
+            agent_id = parse_agent_drive_ref(drive_ref)
+            tenant_id = (request.args.get("tenant_id") or "").strip()
+            if not tenant_id:
+                raise AgentDriveError("missing_tenant_id", "tenant_id is required", status_code=400)
+            items = AgentDriveService().list_skills(tenant_id=tenant_id, agent_id=agent_id)
+        except AgentDriveError as exc:
+            return _error_response(exc)
+        return {"items": items}
+
+
 @inner_api_ns.route("/drive/<string:drive_ref>/commit")
 class AgentDriveCommitApi(Resource):
     @setup_required
@@ -69,9 +90,10 @@ class AgentDriveCommitApi(Resource):
                 body = _CommitRequest.model_validate(request.get_json(silent=True) or {})
             except ValidationError as exc:
                 raise AgentDriveError("invalid_request", str(exc), status_code=400) from exc
+            user = get_user(body.tenant_id, body.user_id)
             items = AgentDriveService().commit(
                 tenant_id=body.tenant_id,
-                user_id=body.user_id,
+                user_id=user.id,
                 agent_id=agent_id,
                 items=body.items,
             )

api/controllers/inner_api/workspace/workspace.py

@@ -47,7 +47,7 @@ class EnterpriseWorkspace(Resource):
         if account is None:
             return {"message": "owner account not found."}, 404
 
-        tenant = TenantService.create_tenant(args.name, is_from_dashboard=True)
+        tenant = TenantService.create_tenant(args.name, is_from_dashboard=True, session=db.session)
         TenantService.create_tenant_member(tenant, account, db.session, role="owner")
 
         tenant_was_created.send(tenant)
@@ -84,7 +84,7 @@ class EnterpriseWorkspaceNoOwnerEmail(Resource):
     def post(self):
         args = WorkspaceOwnerlessPayload.model_validate(inner_api_ns.payload or {})
 
-        tenant = TenantService.create_tenant(args.name, is_from_dashboard=True)
+        tenant = TenantService.create_tenant(args.name, is_from_dashboard=True, session=db.session)
 
         tenant_was_created.send(tenant)
 

api/controllers/openapi/__init__.py

@@ -31,7 +31,7 @@ from controllers.openapi._models import (
     AppDslExportQuery,
     AppDslExportResponse,
     AppDslImportPayload,
-    AppInfoResponse,
+    AppInfo,
     AppListQuery,
     AppListResponse,
     AppListRow,
@@ -62,7 +62,6 @@ from controllers.openapi._models import (
     SessionListQuery,
     SessionListResponse,
     SessionRow,
-    TagItem,
     TaskStopResponse,
     UsageInfo,
     WorkflowRunData,
@@ -96,12 +95,11 @@ register_response_schema_models(
     openapi_ns,
     ErrorBody,
     EventStreamResponse,
-    TagItem,
     UsageInfo,
     MessageMetadata,
     AppListRow,
     AppListResponse,
-    AppInfoResponse,
+    AppInfo,
     AppDescribeInfo,
     AppDescribeResponse,
     AppDslExportResponse,

api/controllers/openapi/_errors.py

@@ -63,6 +63,8 @@ class OpenApiErrorCode(StrEnum):
     FILE_EXTENSION_BLOCKED = "file_extension_blocked"
     MEMBER_LIMIT_EXCEEDED = "member_limit_exceeded"
     MEMBER_LICENSE_EXCEEDED = "member_license_exceeded"
+    HUMAN_INPUT_FORM_NOT_FOUND = "form_not_found"
+    RECIPIENT_SURFACE_MISMATCH = "recipient_surface_mismatch"
 
 
 class ErrorDetail(BaseModel):
@@ -239,3 +241,16 @@ class MemberLicenseExceeded(OpenApiError):  # noqa: N818
     error_code = OpenApiErrorCode.MEMBER_LICENSE_EXCEEDED
     description = "Workspace member license capacity reached."
     hint = "Contact your workspace administrator to expand the license seat count."
+
+
+class HumanInputFormNotFound(OpenApiError):  # noqa: N818
+    code = 404
+    error_code = OpenApiErrorCode.HUMAN_INPUT_FORM_NOT_FOUND
+    description = "No human-input form matches this token. It may be wrong, expired, or already submitted."
+
+
+class RecipientSurfaceMismatch(OpenApiError):  # noqa: N818
+    code = 403
+    error_code = OpenApiErrorCode.RECIPIENT_SURFACE_MISMATCH
+    description = "This form's recipient can't be submitted via the OpenAPI surface."
+    hint = "Action it through its channel (web app or console)."

api/controllers/openapi/_models.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from typing import Any, Literal
+from enum import StrEnum
+from typing import Any, Final, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
@@ -13,6 +14,30 @@ from models.model import AppMode
 MAX_PAGE_LIMIT = 200
 
 
+class SupportedAppType(StrEnum):
+    """App types the ``app`` usage face (``get app``) lists and filters.
+
+    A curated subset of :class:`AppMode`: the real, user-facing app categories.
+    Excludes runtime-only mode tags that are not standalone apps
+    (``rag-pipeline`` is a knowledge ``Pipeline``; ``channel`` is unused) and the
+    roster-owned ``agent`` type (surfaced through the roster, not this list).
+
+    Members reference ``AppMode.*.value`` so the subset relationship is
+    type-checked: dropping a member from ``AppMode`` breaks this at import.
+    This is the single source for the listable set — params, filters, and the
+    generated CLI whitelist all derive from it.
+    """
+
+    COMPLETION = AppMode.COMPLETION.value
+    CHAT = AppMode.CHAT.value
+    ADVANCED_CHAT = AppMode.ADVANCED_CHAT.value
+    WORKFLOW = AppMode.WORKFLOW.value
+    AGENT_CHAT = AppMode.AGENT_CHAT.value
+
+
+SUPPORTED_APP_TYPES: Final[tuple[AppMode, ...]] = tuple(AppMode(t.value) for t in SupportedAppType)
+
+
 class UsageInfo(BaseModel):
     prompt_tokens: int = 0
     completion_tokens: int = 0
@@ -38,18 +63,12 @@ class PaginationEnvelope[T](BaseModel):
         return cls(page=page, limit=limit, total=total, has_more=page * limit < total, data=items)
 
 
-class TagItem(BaseModel):
-    name: str
-
-
 class AppListRow(BaseModel):
     id: str
     name: str
     description: str | None = None
     mode: AppMode
-    tags: list[TagItem] = []
     updated_at: str | None = None
-    created_by_name: str | None = None
     workspace_id: str | None = None
     workspace_name: str | None = None
 
@@ -70,16 +89,14 @@ class PermittedExternalAppsListResponse(BaseModel):
     data: list[AppListRow]
 
 
-class AppInfoResponse(BaseModel):
+class AppInfo(BaseModel):
     id: str
     name: str
     description: str | None = None
     mode: str
-    author: str | None = None
-    tags: list[TagItem] = []
 
 
-class AppDescribeInfo(AppInfoResponse):
+class AppDescribeInfo(AppInfo):
     updated_at: str | None = None
     service_api_enabled: bool
     is_agent: bool = False
@@ -287,14 +304,13 @@ class AppDescribeQuery(BaseModel):
 
 
 class AppListQuery(BaseModel):
-    """mode is a closed enum."""
+    """mode is a closed enum of listable app types."""
 
     workspace_id: UUIDStr
     page: int = Field(1, ge=1)
     limit: int = Field(20, ge=1, le=MAX_PAGE_LIMIT)
-    mode: AppMode | None = None
+    mode: SupportedAppType | None = None
     name: str | None = Field(None, max_length=200)
-    tag: str | None = Field(None, max_length=100)
 
 
 class AppRunRequest(BaseModel):
@@ -344,7 +360,7 @@ class PermittedExternalAppsListQuery(BaseModel):
 
     page: int = Field(1, ge=1)
     limit: int = Field(20, ge=1, le=MAX_PAGE_LIMIT)
-    mode: AppMode | None = None
+    mode: SupportedAppType | None = None
     name: str | None = Field(None, max_length=200)
 
 

api/controllers/openapi/app_dsl.py

@@ -5,11 +5,12 @@ from typing import cast
 from flask_restx import Resource
 from sqlalchemy.orm import Session
 
+from controllers.common.wraps import RBACPermission, RBACResourceScope
 from controllers.openapi import openapi_ns
 from controllers.openapi._contract import accepts, returns
 from controllers.openapi._models import AppDslExportQuery, AppDslExportResponse, AppDslImportPayload
 from controllers.openapi.auth.composition import auth_router
-from controllers.openapi.auth.data import AuthData
+from controllers.openapi.auth.data import AuthData, RBACRequirement
 from extensions.ext_database import db
 from libs.oauth_bearer import Scope, TokenType
 from models import Account, App
@@ -37,6 +38,11 @@ class AppDslImportApi(Resource):
         scope=Scope.WORKSPACE_WRITE,
         allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}),
         allowed_roles=frozenset({TenantAccountRole.EDITOR, TenantAccountRole.ADMIN, TenantAccountRole.OWNER}),
+        rbac=RBACRequirement(
+            resource_type=RBACResourceScope.APP,
+            scene=RBACPermission.APP_IMPORT_EXPORT_DSL,
+            resource_required=False,
+        ),
     )
     @returns(200, Import, "Import completed")
     @returns(202, Import, "Import pending confirmation")
@@ -89,6 +95,11 @@ class AppDslImportConfirmApi(Resource):
         scope=Scope.WORKSPACE_WRITE,
         allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}),
         allowed_roles=frozenset({TenantAccountRole.EDITOR, TenantAccountRole.ADMIN, TenantAccountRole.OWNER}),
+        rbac=RBACRequirement(
+            resource_type=RBACResourceScope.APP,
+            scene=RBACPermission.APP_IMPORT_EXPORT_DSL,
+            resource_required=False,
+        ),
     )
     @returns(200, Import, "Import confirmed")
     @returns(400, Import, "Import failed")
@@ -125,6 +136,7 @@ class AppDslExportApi(Resource):
         scope=Scope.APPS_READ,
         allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}),
         allowed_roles=frozenset({TenantAccountRole.EDITOR, TenantAccountRole.ADMIN, TenantAccountRole.OWNER}),
+        rbac=RBACRequirement(resource_type=RBACResourceScope.APP, scene=RBACPermission.APP_IMPORT_EXPORT_DSL),
     )
     @accepts(query=AppDslExportQuery)
     @returns(200, AppDslExportResponse, "Export successful")
@@ -155,6 +167,7 @@ class AppDslCheckDependenciesApi(Resource):
         scope=Scope.APPS_READ,
         allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}),
         allowed_roles=frozenset({TenantAccountRole.EDITOR, TenantAccountRole.ADMIN, TenantAccountRole.OWNER}),
+        rbac=RBACRequirement(resource_type=RBACResourceScope.APP, scene=RBACPermission.APP_IMPORT_EXPORT_DSL),
     )
     @returns(200, CheckDependenciesResult, "Dependencies checked")
     def get(self, app_id: str, *, auth_data: AuthData):

api/controllers/openapi/app_run.py

@@ -19,12 +19,13 @@ from werkzeug.exceptions import (
 
 import services
 from controllers.common.fields import EventStreamResponse
+from controllers.common.wraps import RBACPermission, RBACResourceScope
 from controllers.openapi import openapi_ns
 from controllers.openapi._audit import emit_app_run
 from controllers.openapi._contract import accepts, returns
 from controllers.openapi._models import AppRunRequest, TaskStopResponse
 from controllers.openapi.auth.composition import auth_router
-from controllers.openapi.auth.data import AuthData
+from controllers.openapi.auth.data import AuthData, RBACRequirement
 from controllers.service_api.app.error import (
     AppUnavailableError,
     CompletionRequestError,
@@ -136,7 +137,10 @@ _DISPATCH: dict[AppMode, Callable[[App, Any, AppRunRequest], Any]] = {
 
 @openapi_ns.route("/apps/<string:app_id>/run")
 class AppRunApi(Resource):
-    @auth_router.guard(scope=Scope.APPS_RUN)
+    @auth_router.guard(
+        scope=Scope.APPS_RUN,
+        rbac=RBACRequirement(resource_type=RBACResourceScope.APP, scene=RBACPermission.APP_TEST_AND_RUN),
+    )
     @openapi_ns.response(200, "Run result (SSE stream)", openapi_ns.models[EventStreamResponse.__name__])
     @accepts(body=AppRunRequest)
     def post(self, app_id: str, *, auth_data: AuthData, body: AppRunRequest):
@@ -167,7 +171,10 @@ class AppRunApi(Resource):
 
 @openapi_ns.route("/apps/<string:app_id>/tasks/<string:task_id>/stop")
 class AppRunTaskStopApi(Resource):
-    @auth_router.guard(scope=Scope.APPS_RUN)
+    @auth_router.guard(
+        scope=Scope.APPS_RUN,
+        rbac=RBACRequirement(resource_type=RBACResourceScope.APP, scene=RBACPermission.APP_TEST_AND_RUN),
+    )
     @returns(200, TaskStopResponse, description="Task stopped")
     def post(self, app_id: str, task_id: str, *, auth_data: AuthData):
         app_model, caller, caller_kind = auth_data.require_app_context()

api/controllers/openapi/apps.py

@@ -8,33 +8,41 @@ from typing import Any, cast
 from flask_restx import Resource
 from werkzeug.exceptions import Conflict, NotFound, UnprocessableEntity
 
+from configs import dify_config
+from controllers.common.app_access import AppAccessFilter, resolve_app_access_filter
 from controllers.common.fields import Parameters
+from controllers.common.wraps import RBACPermission, RBACResourceScope
 from controllers.openapi import openapi_ns
 from controllers.openapi._contract import accepts, returns
 from controllers.openapi._input_schema import EMPTY_INPUT_SCHEMA, build_input_schema, resolve_app_config
 from controllers.openapi._models import (
+    SUPPORTED_APP_TYPES,
     AppDescribeInfo,
     AppDescribeQuery,
     AppDescribeResponse,
     AppListQuery,
     AppListResponse,
     AppListRow,
-    TagItem,
 )
 from controllers.openapi.auth.composition import auth_router
-from controllers.openapi.auth.data import AuthData
+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.model import AppMode
 from services.account_service import TenantService
 from services.app_service import AppListParams, AppService
-from services.tag_service import TagService
 
 _ALLOWED_DESCRIBE_FIELDS: frozenset[str] = frozenset({"info", "parameters", "input_schema"})
 
 
+def _is_listable(app: App) -> bool:
+    """Whether the openapi app face exposes this app (curated, listable types only)."""
+    return app.mode in SUPPORTED_APP_TYPES
+
+
 _EMPTY_PARAMETERS: dict[str, Any] = {
     "opening_statement": None,
     "suggested_questions": [],
@@ -84,54 +92,55 @@ def parameters_payload(app: App) -> dict:
     return Parameters.model_validate(parameters).model_dump(mode="json")
 
 
+def build_app_describe_response(app: App, fields: set[str] | None) -> AppDescribeResponse:
+    """Public projection of an app (name / params / input schema) — never internal config."""
+    want_info = fields is None or "info" in fields
+    want_params = fields is None or "parameters" in fields
+    want_schema = fields is None or "input_schema" in fields
+
+    info = (
+        AppDescribeInfo(
+            id=str(app.id),
+            name=app.name,
+            mode=app.mode,
+            description=app.description,
+            updated_at=app.updated_at.isoformat() if app.updated_at else None,
+            service_api_enabled=bool(app.enable_api),
+            is_agent=app.mode in (AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT),
+        )
+        if want_info
+        else None
+    )
+
+    parameters: dict[str, Any] | None = None
+    input_schema: dict[str, Any] | None = None
+    if want_params:
+        try:
+            parameters = parameters_payload(app)
+        except AppUnavailableError:
+            parameters = dict(_EMPTY_PARAMETERS)
+    if want_schema:
+        try:
+            input_schema = build_input_schema(app)
+        except AppUnavailableError:
+            input_schema = dict(EMPTY_INPUT_SCHEMA)
+
+    return AppDescribeResponse(info=info, parameters=parameters, input_schema=input_schema)
+
+
 @openapi_ns.route("/apps/<string:app_id>/describe")
 class AppDescribeApi(AppReadResource):
-    @auth_router.guard(scope=Scope.APPS_READ, allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}))
+    @auth_router.guard(
+        scope=Scope.APPS_READ,
+        allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}),
+        rbac=RBACRequirement(resource_type=RBACResourceScope.APP, scene=RBACPermission.APP_VIEW_LAYOUT),
+    )
     @returns(200, AppDescribeResponse, description="App description")
     @accepts(query=AppDescribeQuery)
     def get(self, app_id: str, *, auth_data: AuthData, query: AppDescribeQuery):
         # describe is UUID-only (workspace_id query param dropped in #37212).
         app = self._load(app_id)
-
-        requested = query.fields
-        want_info = requested is None or "info" in requested
-        want_params = requested is None or "parameters" in requested
-        want_schema = requested is None or "input_schema" in requested
-
-        info = (
-            AppDescribeInfo(
-                id=str(app.id),
-                name=app.name,
-                mode=app.mode,
-                description=app.description,
-                tags=[TagItem(name=t.name) for t in app.tags],
-                author=app.author_name,
-                updated_at=app.updated_at.isoformat() if app.updated_at else None,
-                service_api_enabled=bool(app.enable_api),
-                is_agent=app.mode in ("agent-chat", "advanced-chat"),
-            )
-            if want_info
-            else None
-        )
-
-        parameters: dict[str, Any] | None = None
-        input_schema: dict[str, Any] | None = None
-        if want_params:
-            try:
-                parameters = parameters_payload(app)
-            except AppUnavailableError:
-                parameters = dict(_EMPTY_PARAMETERS)
-        if want_schema:
-            try:
-                input_schema = build_input_schema(app)
-            except AppUnavailableError:
-                input_schema = dict(EMPTY_INPUT_SCHEMA)
-
-        return AppDescribeResponse(
-            info=info,
-            parameters=parameters,
-            input_schema=input_schema,
-        )
+        return build_app_describe_response(app, query.fields)
 
 
 @openapi_ns.route("/apps")
@@ -152,45 +161,57 @@ class AppListApi(Resource):
         else:
             parsed_uuid = None
 
+        # Compute RBAC-accessible app IDs when RBAC is enabled and the caller is an account.
+        # ``None`` means unrestricted (caller can see all apps in the workspace);
+        # 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 != "end_user" and auth_data.account_id is not None
+        )
+        access_filter = AppAccessFilter.unrestricted()
+        if apply_rbac_filter:
+            access_filter = resolve_app_access_filter(workspace_id, str(auth_data.account_id))
+
         tenant_name: str | None = None
         if parsed_uuid is not None:
             app: App | None = AppService.get_visible_app_by_id(db.session, str(parsed_uuid))
             if app is None or str(app.tenant_id) != workspace_id:
                 return empty
+            if not _is_listable(app):
+                return empty
+            # Apply RBAC visibility to the UUID fast-path the same way the service
+            # layer does for paginated queries (id in accessible set OR own app).
+            if apply_rbac_filter and not access_filter.is_app_accessible(
+                str(app.id), str(app.maintainer) if app.maintainer else None, str(auth_data.account_id)
+            ):
+                return empty
             tenant_name = TenantService.get_tenant_name(db.session, workspace_id)
             item = AppListRow(
                 id=str(app.id),
                 name=app.name,
                 description=app.description,
                 mode=app.mode,
-                tags=[TagItem(name=t.name) for t in app.tags],
                 updated_at=app.updated_at.isoformat() if app.updated_at else None,
-                created_by_name=getattr(app, "author_name", None),
                 workspace_id=str(workspace_id),
                 workspace_name=tenant_name,
             )
             env = AppListResponse(page=1, limit=1, total=1, has_more=False, data=[item])
             return env
 
-        tag_ids: list[str] | None = None
-        if query.tag:
-            tags = TagService.get_tag_by_tag_name("app", workspace_id, query.tag, db.session)
-            if not tags:
-                return empty
-            tag_ids = [tag.id for tag in tags]
-
         params = AppListParams(
             page=query.page,
             limit=query.limit,
             mode=query.mode.value if query.mode else "all",  # type:ignore
             name=query.name,
-            tag_ids=tag_ids,
             status="normal",
             # Visibility gate pushed into the query — pagination.total stays
             # consistent across pages because invisible rows never count.
             openapi_visible=True,
         )
 
+        if apply_rbac_filter:
+            access_filter.apply_to_params(params)
+
         pagination = AppService().get_paginate_apps(str(auth_data.account_id), workspace_id, params, db.session)
         if pagination is None:
             return empty
@@ -205,13 +226,12 @@ class AppListApi(Resource):
                 name=r.name,
                 description=r.description,
                 mode=r.mode,
-                tags=[TagItem(name=t.name) for t in r.tags],
                 updated_at=r.updated_at.isoformat() if r.updated_at else None,
-                created_by_name=getattr(r, "author_name", None),
                 workspace_id=str(workspace_id),
                 workspace_name=tenant_name,
             )
             for r in pagination.items
+            if _is_listable(r)
         ]
 
         env = AppListResponse(

api/controllers/openapi/apps_permitted_external.py

@@ -8,14 +8,18 @@ EE blueprint chain so this module is unreachable there.
 from __future__ import annotations
 
 from flask_restx import Resource
+from werkzeug.exceptions import NotFound
 
 from controllers.openapi import openapi_ns
 from controllers.openapi._contract import accepts, returns
 from controllers.openapi._models import (
+    AppDescribeQuery,
+    AppDescribeResponse,
     AppListRow,
     PermittedExternalAppsListQuery,
     PermittedExternalAppsListResponse,
 )
+from controllers.openapi.apps import build_app_describe_response
 from controllers.openapi.auth.composition import auth_router
 from controllers.openapi.auth.data import AuthData, Edition
 from extensions.ext_database import db
@@ -67,9 +71,7 @@ class PermittedExternalAppsListApi(Resource):
                     name=app.name,
                     description=app.description,
                     mode=app.mode,
-                    tags=[],  # tenant-scoped; not surfaced cross-tenant
                     updated_at=app.updated_at.isoformat() if app.updated_at else None,
-                    created_by_name=None,  # cross-tenant author leak prevention
                     workspace_id=str(app.tenant_id),
                     workspace_name=tenant.name if tenant else None,
                 )
@@ -82,3 +84,20 @@ class PermittedExternalAppsListApi(Resource):
             data=items,
         )
         return env
+
+
+@openapi_ns.route("/permitted-external-apps/<string:app_id>/describe")
+class PermittedExternalAppDescribeApi(Resource):
+    @auth_router.guard(
+        scope=Scope.APPS_READ_PERMITTED_EXTERNAL,
+        allowed_token_types=frozenset({TokenType.OAUTH_EXTERNAL_SSO}),
+        edition=frozenset({Edition.EE}),
+    )
+    @returns(200, AppDescribeResponse, description="Permitted external app description")
+    @accepts(query=AppDescribeQuery)
+    def get(self, app_id: str, *, auth_data: AuthData, query: AppDescribeQuery):
+        # App already loaded and ACL-checked by the external_sso pipeline; project it.
+        app = auth_data.app
+        if app is None:
+            raise NotFound("app not found")
+        return build_app_describe_response(app, query.fields)

api/controllers/openapi/auth/composition.py

@@ -3,9 +3,11 @@ from __future__ import annotations
 from controllers.openapi.auth.conditions import (
     EDITION_EE,
     HAS_ALLOWED_ROLES,
+    HAS_RBAC,
     LOADED_APP_IS_PRIVATE,
     PATH_HAS_APP_ID,
     WEBAPP_AUTH_ENABLED,
+    WEBAPP_RUN_SCOPED,
     WORKSPACE_MEMBERSHIP_REQUIRED,
     WORKSPACE_SCOPED,
 )
@@ -25,6 +27,7 @@ from controllers.openapi.auth.verify import (
     check_acl,
     check_app_api_enabled,
     check_private_app_permission,
+    check_rbac_permission,
     check_scope,
     check_workspace_member,
     check_workspace_mismatch,
@@ -47,8 +50,9 @@ account_pipeline = AuthPipeline(
         When(WORKSPACE_SCOPED, then=check_workspace_member),
         When(PATH_HAS_APP_ID, then=check_workspace_mismatch),
         When(HAS_ALLOWED_ROLES, then=check_workspace_role),
-        When(PATH_HAS_APP_ID & EDITION_EE & WEBAPP_AUTH_ENABLED, then=check_acl),
-        When(EDITION_EE & LOADED_APP_IS_PRIVATE, then=check_private_app_permission),
+        When(HAS_RBAC, then=check_rbac_permission),
+        When(PATH_HAS_APP_ID & EDITION_EE & WEBAPP_AUTH_ENABLED & WEBAPP_RUN_SCOPED, then=check_acl),
+        When(EDITION_EE & LOADED_APP_IS_PRIVATE & WEBAPP_RUN_SCOPED, then=check_private_app_permission),
     ],
 )
 

api/controllers/openapi/auth/conditions.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from collections.abc import Callable
 
 from controllers.openapi.auth.data import AuthData, Edition, RequestContext, current_edition
-from libs.oauth_bearer import TokenType
+from libs.oauth_bearer import Scope, TokenType
 from services.enterprise.enterprise_service import WebAppAccessMode
 from services.feature_service import FeatureService
 
@@ -50,8 +50,11 @@ EDITION_SAAS = config_cond(lambda: current_edition() == Edition.SAAS)
 
 WEBAPP_AUTH_ENABLED = config_cond(lambda: FeatureService.get_system_features().webapp_auth.enabled)
 
+WEBAPP_RUN_SCOPED = request_cond(lambda ctx: ctx.scope == Scope.APPS_RUN)
+
 WORKSPACE_MEMBERSHIP_REQUIRED = request_cond(lambda ctx: ctx.workspace_membership)
 HAS_ALLOWED_ROLES = request_cond(lambda ctx: ctx.allowed_roles is not None)
+HAS_RBAC = request_cond(lambda ctx: ctx.rbac is not None)
 
 # Caller must belong to the resolved tenant: either an app-scoped path (tenant
 # from the app) or an explicit workspace-membership path (tenant from request).

api/controllers/openapi/auth/data.py

@@ -8,6 +8,7 @@ from pydantic import BaseModel, ConfigDict, Field
 from werkzeug.exceptions import InternalServerError
 
 from configs import dify_config
+from core.rbac import RBACPermission, RBACResourceScope
 from libs.oauth_bearer import Scope, TokenType
 from models.account import Account, Tenant, TenantAccountRole
 from models.model import App, EndUser
@@ -35,6 +36,14 @@ class ExternalIdentity(BaseModel):
     issuer: str | None = None
 
 
+class RBACRequirement(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    resource_type: RBACResourceScope
+    scene: RBACPermission
+    resource_required: bool = True
+
+
 class RequestContext(BaseModel):
     model_config = ConfigDict(frozen=True)
 
@@ -43,6 +52,7 @@ class RequestContext(BaseModel):
     path_params: dict[str, str]
     workspace_membership: bool = False
     allowed_roles: frozenset[TenantAccountRole] | None = None
+    rbac: RBACRequirement | None = None
 
 
 class AuthData(BaseModel):
@@ -59,6 +69,7 @@ class AuthData(BaseModel):
     path_params: dict[str, str] = Field(default_factory=dict)
 
     allowed_roles: frozenset[TenantAccountRole] | None = None
+    rbac: RBACRequirement | None = None
 
     app: App | None = None
     tenant: Tenant | None = None