feat(openapi): distinguish expired OAuth bearer from invalid token

Previously an expired OAuth bearer and an unknown/invalid one both
surfaced as an indistinguishable generic 401 (and an invalid token
actually leaked a 500), so a client could not tell "session expired,
re-authenticate" apart from "never authenticated."

The resolver now raises a distinct TokenExpiredError for expired DB
rows and records a separate `expired` negative-cache marker, so a
retry within the negative-cache TTL still reports expiry instead of
collapsing into a generic miss. The auth pipeline maps the two domain
errors to unified OpenApiError responses: SessionExpired (code
`token_expired`) and InvalidBearer (code `unauthorized`), both 401.
This also fixes the latent 500 on invalid bearers.

The `token_expired` code is synced through the contract codegen into
the generated types/zod, and the difyctl error mapper branches the
401 on it. The CLI `expired_token` taxonomy member (RFC 8628
device-flow code expiry) is merged into `token_expired`; the RFC 8628
wire value is unchanged.

Closes WTA-1062

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

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

.github/workflows/api-tests.yml

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

.github/workflows/build-push.yml

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

.github/workflows/cli-release.yml

@@ -23,7 +23,7 @@ concurrency:
 jobs:
   validate:
     name: validate manifest + resolve target Dify release
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
 
     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-4
+    runs-on: depot-ubuntu-24.04
 
     steps:
       - name: Checkout code

.github/workflows/deploy-agent.yml

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

.github/workflows/deploy-saas.yml

@@ -13,7 +13,7 @@ on:
 
 jobs:
   deploy:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     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-4
+    runs-on: depot-ubuntu-24.04
     permissions:
       actions: read
       contents: read

.github/workflows/pyrefly-diff.yml

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

api/app_factory.py

@@ -2,14 +2,14 @@ import logging
 import time
 
 import socketio
-from flask import g, request
+from flask import 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 clear_request_context, init_request_context
+from core.logging.context import init_request_context
 from dify_app import DifyApp
 from extensions.ext_socketio import sio
 from services.enterprise.enterprise_service import EnterpriseService
@@ -60,7 +60,6 @@ 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)
@@ -118,26 +117,9 @@ 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/commands/__init__.py

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

api/commands/data_migrate.py

@@ -7,7 +7,6 @@ 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 (
@@ -178,4 +177,3 @@ 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,55 +1,11 @@
 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 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
+from models import TenantAccountJoin, TenantAccountRole
+from services.enterprise.rbac_service import ListOption, RBACService
 
 
 def _resolve_builtin_role_id(tenant_id: str, operator_account_id: str, legacy_role: str) -> str:
@@ -59,86 +15,26 @@ 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.
     """
-    if legacy_role not in _LEGACY_ROLE_TO_BUILTIN_TAG:
+    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:
         raise ValueError(f"Unsupported legacy workspace role: {legacy_role}")
 
-    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(
+    roles = RBACService.Roles.list(
         tenant_id=tenant_id,
         account_id=operator_account_id,
-        member_account_id=member_account_id,
-        role_ids=[role_id],
-    )
-    return member_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}")
 
 
 @click.command(
@@ -146,16 +42,7 @@ def _replace_member_role(
 )
 @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.")
-@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:
+def migrate_member_roles_to_rbac(tenant_id: str | None, dry_run: bool) -> None:
     """Backfill RBAC member-role bindings from legacy `TenantAccountJoin.role` data.
 
     This is an offline migration command for workspaces that already have
@@ -163,322 +50,63 @@ def migrate_member_roles_to_rbac(
     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")
 
-    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] = {}
+    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)
 
-    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}")
+        joins = list(session.scalars(stmt).all())
 
-        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:
+    if not joins:
         click.echo(click.style("No workspace members found for migration.", fg="yellow"))
         return
 
-    if dry_run:
-        click.echo(
-            click.style(
-                f"Dry run completed. Scanned {scanned_count} members across {tenant_count} tenants. "
-                "No RBAC bindings were written.",
-                fg="yellow",
+    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,
             )
-        )
-    else:
+            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]
         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",
-            )
+            f"tenant={workspace_id} member={member_account_id} "
+            f"legacy_role={legacy_role} -> rbac_role_id={resolved_role_id}"
         )
 
+        if dry_run:
+            continue
 
-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
+        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
 
     if dry_run:
-        click.echo(
-            click.style(
-                f"Dry run completed. Scanned {scanned_count} datasets; "
-                f"{partial_dataset_count} partial-member datasets would be migrated.",
-                fg="yellow",
-            )
-        )
+        click.echo(click.style("Dry run completed. No RBAC bindings were written.", fg="yellow"))
     else:
-        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",
-            )
-        )
+        click.echo(click.style(f"RBAC member-role migration completed. Migrated {migrated_count} members.", fg="green"))

api/configs/enterprise/__init__.py

@@ -34,12 +34,6 @@ 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/controllers/console/app/app.py

@@ -43,7 +43,6 @@ 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
@@ -70,7 +69,6 @@ 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"]
 
@@ -668,14 +666,6 @@ class AppListApi(Resource):
 
         app_service = AppService()
         app = app_service.create_app(current_tenant_id, params, current_user)
-        if dify_config.RBAC_ENABLED:
-            enterprise_rbac_service.RBACService.AppAccess.replace_whitelist(
-                tenant_id=str(current_tenant_id),
-                account_id=current_user.id,
-                app_id=str(app.id),
-                payload=enterprise_rbac_service.ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
-            )
-            initialize_created_app_rbac_access_task.delay(current_tenant_id, current_user.id, app_id=app.id)
         permission_keys_map = enterprise_rbac_service.RBACService.AppPermissions.batch_get(
             str(current_tenant_id),
             current_user.id,

api/controllers/console/app/message.py

@@ -167,12 +167,16 @@ register_schema_models(
     ChatMessagesQuery,
     MessageFeedbackPayload,
     FeedbackExportQuery,
+)
+register_response_schema_models(
+    console_ns,
     AnnotationCountResponse,
     SuggestedQuestionsResponse,
     MessageDetailResponse,
     MessageInfiniteScrollPaginationResponse,
+    SimpleResultResponse,
+    TextFileResponse,
 )
-register_response_schema_models(console_ns, SimpleResultResponse, TextFileResponse)
 
 
 @console_ns.route("/apps/<uuid:app_id>/chat-messages")

api/controllers/console/app/wraps.py

@@ -19,8 +19,7 @@ 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, TrialApp
-from services.recommended_app_service import RecommendedAppService
+from models import App, AppMode
 
 
 def _load_app_model(session: Session, app_id: str) -> App | None:
@@ -42,10 +41,7 @@ def _load_app_model_from_scoped_session(app_id: str) -> App | None:
 
 
 def _load_app_model_with_trial(app_id: str) -> App | None:
-    """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)
-    )
+    app_model = db.session.scalar(select(App).where(App.id == app_id, App.status == "normal").limit(1))
     return app_model
 
 
@@ -199,8 +195,6 @@ 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:
@@ -213,8 +207,6 @@ def get_app_model_with_trial[**P, R](
             del kwargs["app_id"]
 
             app_model = _load_app_model_with_trial(app_id)
-            if app_model is None:
-                app_model = RecommendedAppService.get_app(app_id, session=db.session())
 
             if not app_model:
                 raise AppNotFoundError()

api/controllers/console/auth/email_register.py

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

api/controllers/console/auth/login.py

@@ -25,7 +25,6 @@ from controllers.console.error import (
     AccountNotFound,
     EmailSendIpLimitError,
     NotAllowedCreateWorkspace,
-    SeatsLimitExceeded,
     WorkspacesLimitExceeded,
 )
 from controllers.console.wraps import (
@@ -52,7 +51,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, SeatsLimitExceededError
+from services.errors.account import AccountRegisterError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService
 
@@ -318,8 +317,6 @@ class EmailCodeLoginApi(Resource):
                 )
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()
-            except SeatsLimitExceededError:
-                raise SeatsLimitExceeded()
             except AccountRegisterError:
                 _log_console_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_IN_FREEZE)
                 raise AccountInFreezeError()

api/controllers/console/auth/oauth.py

@@ -25,7 +25,7 @@ from libs.token import (
 from models import Account, AccountStatus
 from services.account_service import AccountService, RegisterService, TenantService
 from services.billing_service import BillingService
-from services.errors.account import AccountNotFoundError, AccountRegisterError, SeatsLimitExceededError
+from services.errors.account import AccountNotFoundError, AccountRegisterError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkSpaceNotFoundError
 from services.feature_service import FeatureService
 
@@ -38,7 +38,6 @@ 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):
@@ -88,36 +87,6 @@ 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
@@ -140,7 +109,6 @@ 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)
@@ -151,7 +119,6 @@ class OAuthLogin(Resource):
             invite_token=invite_token,
             timezone=timezone,
             language=language,
-            redirect_url=redirect_url,
         )
         return redirect(auth_url)
 
@@ -177,7 +144,6 @@ 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
@@ -216,8 +182,6 @@ 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}")
 
@@ -246,9 +210,9 @@ class OAuthCallback(Resource):
             ip_address=extract_remote_ip(request),
         )
 
-        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()}"
+        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()}"
         response = redirect(target_url)
 
         set_access_token_to_cookie(request, response, token_pair.access_token)

api/controllers/console/datasets/datasets.py

@@ -50,8 +50,6 @@ 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)
 
@@ -567,16 +565,6 @@ 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/error.py

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

api/controllers/console/explore/installed_app.py

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

api/controllers/console/explore/recommended_app.py

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

api/controllers/console/explore/trial.py

@@ -44,9 +44,7 @@ from controllers.console.explore.error import (
     NotWorkflowAppError,
 )
 from controllers.console.explore.wraps import TrialAppResource, trial_feature_enable
-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.console.wraps import 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
@@ -66,7 +64,6 @@ 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 (
@@ -78,8 +75,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 dump_response, uuid_value
-from models import Account, App
+from libs.helper import uuid_value
+from models import Account
 from models.account import TenantStatus
 from models.model import AppMode, Site
 from models.workflow import Workflow
@@ -202,36 +199,6 @@ 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__])
@@ -638,18 +605,6 @@ 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, UploadFile
+from models import Account
 from services.file_service import FileService
 
 from . import console_ns
@@ -38,59 +38,6 @@ 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):
@@ -117,11 +64,41 @@ 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):
-        upload_file = upload_file_from_request(current_user=current_user)
+        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)
 
         response = FileResponse.model_validate(upload_file, from_attributes=True)
         return response.model_dump(mode="json"), 201

api/controllers/console/remote_files.py

@@ -46,61 +46,6 @@ 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__])
@@ -108,8 +53,53 @@ class RemoteFileUpload(Resource):
     @login_required
     @with_current_user
     def post(self, current_user: Account):
-        remote_file = upload_remote_file_from_request(current_user=current_user)
+        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
         return (
-            remote_file.model_dump(mode="json"),
+            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"),
             201,
         )

api/controllers/console/workspace/members.py

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

api/controllers/console/workspace/rbac.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from enum import StrEnum
 from typing import Any
 
 from flask import request
@@ -13,11 +14,9 @@ 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]):
@@ -512,8 +511,14 @@ class RBACAccessPolicyBindingUnlockApi(Resource):
 # ---------------------------------------------------------------------------
 
 
+class _AccessScope(StrEnum):
+    ALL = "all"
+    SPECIFIC = "specific"
+    ONLY_ME = "only_me"
+
+
 class _ResourceAccessScopeRequest(BaseModel):
-    scope: RBACResourceWhitelistScope
+    scope: _AccessScope
 
 
 class _ReplaceBindingsRequest(BaseModel):
@@ -579,15 +584,14 @@ class RBACAppWhitelistApi(Resource):
     def put(self, app_id):
         tenant_id, account_id = _current_ids()
         request = _payload(_ResourceAccessScopeRequest)
-        result = svc.RBACService.AppAccess.replace_whitelist(
-            tenant_id,
-            account_id,
-            str(app_id),
-            svc.ReplaceMemberBindings(scope=request.scope.value),
+        return _dump(
+            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")
@@ -612,8 +616,8 @@ class RBACAppUserAccessPolicyAssignmentApi(Resource):
             svc.RBACService.AppAccess.replace_user_access_policies(
                 tenant_id,
                 account_id,
-                app_id,
-                target_account_id,
+                str(app_id),
+                str(target_account_id),
                 payload,
             )
         )

api/controllers/openapi/__init__.py

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

api/controllers/openapi/_errors.py

@@ -34,6 +34,7 @@ class OpenApiErrorCode(StrEnum):
     # transport-generic (resolved from HTTP status for plain werkzeug raises)
     BAD_REQUEST = "bad_request"
     UNAUTHORIZED = "unauthorized"
+    TOKEN_EXPIRED = "token_expired"
     FORBIDDEN = "forbidden"
     NOT_FOUND = "not_found"
     METHOD_NOT_ALLOWED = "method_not_allowed"
@@ -223,6 +224,19 @@ class OpenApiErrorFormatter:
         return isinstance(part, (str, int)) and not isinstance(part, bool)
 
 
+class InvalidBearer(OpenApiError):  # noqa: N818
+    code = 401
+    error_code = OpenApiErrorCode.UNAUTHORIZED
+    description = "Invalid or unknown bearer token."
+
+
+class SessionExpired(OpenApiError):  # noqa: N818
+    code = 401
+    error_code = OpenApiErrorCode.TOKEN_EXPIRED
+    description = "Your session has expired."
+    hint = "Re-authenticate to continue (e.g. re-run your login command)."
+
+
 class FilenameNotExists(OpenApiError):  # noqa: N818
     code = 400
     error_code = OpenApiErrorCode.FILENAME_NOT_EXISTS

api/controllers/openapi/app_run.py

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

api/controllers/openapi/apps.py

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

api/controllers/openapi/apps_permitted_external.py

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

api/controllers/openapi/auth/data.py

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

api/controllers/openapi/auth/pipeline.py

@@ -17,6 +17,7 @@ from flask_login import user_logged_in
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
 from controllers.openapi._audit import emit_wrong_surface
+from controllers.openapi._errors import InvalidBearer, SessionExpired
 from controllers.openapi.auth.data import (
     AuthData,
     Edition,
@@ -28,7 +29,9 @@ from controllers.openapi.auth.data import (
 from controllers.openapi.auth.flow import When
 from libs.oauth_bearer import (
     AuthContext,
+    InvalidBearerError,
     Scope,
+    TokenExpiredError,
     TokenType,
     extract_bearer,
     get_authenticator,
@@ -217,7 +220,12 @@ class PipelineRouter:
         if not token:
             raise Unauthorized("bearer required")
 
-        identity = get_authenticator().authenticate(token)
+        try:
+            identity = get_authenticator().authenticate(token)
+        except TokenExpiredError:
+            raise SessionExpired()
+        except InvalidBearerError:
+            raise InvalidBearer()
 
         if allowed_token_types is not None and identity.token_type not in allowed_token_types:
             emit_wrong_surface(

api/controllers/openapi/auth/prepare.py

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

api/controllers/openapi/auth/verify.py

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

api/controllers/openapi/human_input_form.py

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

api/controllers/openapi/workflow_events.py

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

api/controllers/openapi/workspaces.py

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

api/controllers/service_api/dataset/dataset.py

@@ -43,7 +43,6 @@ from models.dataset import DatasetPermissionEnum
 from models.enums import TagType
 from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
-from services.enterprise.rbac_service import RBACResourceWhitelistScope, RBACService, ReplaceMemberBindings
 from services.entities.knowledge_entities.knowledge_entities import (
     ExternalRetrievalModel,
     KnowledgeProvider,
@@ -59,7 +58,6 @@ from services.tag_service import (
 from services.tag_service import (
     UpdateTagPayload as UpdateTagServicePayload,
 )
-from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
 
 register_enum_models(service_api_ns, DatasetPermissionEnum)
 
@@ -525,15 +523,6 @@ class DatasetListApi(DatasetApiResource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
-        if payload.permission == DatasetPermissionEnum.ALL_TEAM and dify_config.RBAC_ENABLED:
-            RBACService.DatasetAccess.replace_whitelist(
-                tenant_id,
-                current_user.id,
-                dataset.id,
-                ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
-            )
-            initialize_created_app_rbac_access_task.delay(tenant_id, current_user.id, dataset_id=dataset.id)
-
         return _dump_service_dataset_detail(dataset), 200
 
 

api/controllers/service_api/wraps.py

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

api/controllers/trigger/webhook.py

@@ -7,26 +7,10 @@ from werkzeug.exceptions import NotFound, RequestEntityTooLarge
 from controllers.trigger import bp
 from core.trigger.debug.event_bus import TriggerDebugEventBus
 from core.trigger.debug.events import WebhookDebugEvent, build_webhook_pool_key
-from enums.quota_type import QuotaType
-from services.errors.app import QuotaExceededError
 from services.trigger.webhook_service import RawWebhookDataDict, WebhookService
 
 logger = logging.getLogger(__name__)
 
-_QUOTA_EXCEEDED_MESSAGES = {
-    QuotaType.TRIGGER: "Trigger event quota exceeded. Please upgrade your plan.",
-    QuotaType.WORKFLOW: "Workflow execution quota exceeded. Please upgrade your plan.",
-}
-_DEFAULT_QUOTA_EXCEEDED_MESSAGE = "Quota exceeded. Please upgrade your plan."
-
-
-def _get_quota_exceeded_message(feature: str) -> str:
-    try:
-        quota_type = QuotaType(feature)
-    except ValueError:
-        return _DEFAULT_QUOTA_EXCEEDED_MESSAGE
-    return _QUOTA_EXCEEDED_MESSAGES.get(quota_type, _DEFAULT_QUOTA_EXCEEDED_MESSAGE)
-
 
 def _prepare_webhook_execution(webhook_id: str, is_debug: bool = False):
     """Fetch trigger context, extract request data, and validate payload using unified processing.
@@ -76,15 +60,8 @@ def handle_webhook(webhook_id: str):
         response_data, status_code = WebhookService.generate_webhook_response(node_config)
         return jsonify(response_data), status_code
 
-    except QuotaExceededError as error:
-        return jsonify(
-            {
-                "error": "Too Many Requests",
-                "message": _get_quota_exceeded_message(error.feature),
-            }
-        ), 429
-    except ValueError as error:
-        raise NotFound(str(error))
+    except ValueError as e:
+        raise NotFound(str(e))
     except RequestEntityTooLarge:
         raise
     except Exception as e:

api/controllers/web/login.py

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

api/controllers/web/wraps.py

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

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

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

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

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

api/core/app/apps/base_app_generator.py

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

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

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

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

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

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

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

api/core/app/layers/pause_state_persist_layer.py

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

api/core/callback_handler/workflow_tool_callback_handler.py

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

api/core/helper/code_executor/code_executor.py

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

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

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

api/core/helper/code_executor/template_transformer.py

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

api/core/helper/creators.py

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

api/core/helper/credential_utils.py

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

api/core/helper/download.py

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

api/core/helper/encrypter.py

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

api/core/helper/marketplace.py

@@ -1,6 +1,5 @@
 import logging
 from collections.abc import Sequence
-from typing import Any
 
 import httpx
 from yarl import URL
@@ -20,7 +19,7 @@ def get_plugin_pkg_url(plugin_unique_identifier: str) -> str:
     return str((marketplace_api_url / "api/v1/plugins/download").with_query(unique_identifier=plugin_unique_identifier))
 
 
-def download_plugin_pkg(plugin_unique_identifier: str) -> bytes:
+def download_plugin_pkg(plugin_unique_identifier: str):
     return download_with_size_limit(get_plugin_pkg_url(plugin_unique_identifier), dify_config.PLUGIN_MAX_PACKAGE_SIZE)
 
 
@@ -40,7 +39,7 @@ def batch_fetch_plugin_manifests(plugin_ids: list[str]) -> Sequence[MarketplaceP
     return [MarketplacePluginDeclaration.model_validate(plugin) for plugin in response.json()["data"]["plugins"]]
 
 
-def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict[str, Any]]:
+def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict]:
     if not plugin_ids:
         return []
 
@@ -54,19 +53,10 @@ def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict[str, Any]]:
     response.raise_for_status()
 
     data = response.json()
-    plugins = data.get("data", {}).get("plugins", [])
-    if not isinstance(plugins, list):
-        raise ValueError("Marketplace did not return a valid plugins list")
-
-    result: list[dict[str, Any]] = []
-    for plugin in plugins:
-        if not isinstance(plugin, dict) or not all(isinstance(key, str) for key in plugin):
-            raise ValueError("Marketplace did not return a valid plugins list")
-        result.append(plugin)
-    return result
+    return data.get("data", {}).get("plugins", [])
 
 
-def record_install_plugin_event(plugin_unique_identifier: str) -> None:
+def record_install_plugin_event(plugin_unique_identifier: str):
     url = str(marketplace_api_url / "api/v1/stats/plugins/install_count")
     response = httpx.post(url, json={"unique_identifier": plugin_unique_identifier}, timeout=MARKETPLACE_TIMEOUT)
     response.raise_for_status()

api/core/helper/model_provider_cache.py

@@ -34,7 +34,7 @@ class ProviderCredentialsCache:
         else:
             return None
 
-    def set(self, credentials: dict[str, Any]) -> None:
+    def set(self, credentials: dict[str, Any]):
         """
         Cache model provider credentials.
 
@@ -43,7 +43,7 @@ class ProviderCredentialsCache:
         """
         redis_client.setex(self.cache_key, 86400, json.dumps(credentials))
 
-    def delete(self) -> None:
+    def delete(self):
         """
         Delete cached model provider credentials.
 

api/core/helper/module_import_helper.py

@@ -20,18 +20,17 @@ def import_module_from_source[T: (str, bytes)](
                 raise Exception(f"Failed to load module {module_name} from {py_file_path!r}")
         else:
             # Refer to: https://docs.python.org/3/library/importlib.html#importing-a-source-file-directly
-            new_spec = importlib.util.spec_from_file_location(module_name, py_file_path)
-            if not new_spec or not new_spec.loader:
+            # FIXME: mypy does not support the type of spec.loader
+            spec = importlib.util.spec_from_file_location(module_name, py_file_path)  # type: ignore[assignment]
+            if not spec or not spec.loader:
                 raise Exception(f"Failed to load module {module_name} from {py_file_path!r}")
             if use_lazy_loader:
                 # Refer to: https://docs.python.org/3/library/importlib.html#implementing-lazy-imports
-                new_spec.loader = importlib.util.LazyLoader(new_spec.loader)
-            spec = new_spec
+                spec.loader = importlib.util.LazyLoader(spec.loader)
         module = importlib.util.module_from_spec(spec)
         if not existed_spec:
             sys.modules[module_name] = module
-        if spec.loader is not None:
-            spec.loader.exec_module(module)
+        spec.loader.exec_module(module)
         return module
     except Exception as e:
         logger.exception("Failed to load module %s from script file '%s'", module_name, repr(py_file_path))

api/core/helper/provider_cache.py

@@ -9,11 +9,11 @@ from extensions.ext_redis import redis_client
 class ProviderCredentialsCache(ABC):
     """Base class for provider credentials cache"""
 
-    def __init__(self, **kwargs: Any) -> None:
+    def __init__(self, **kwargs):
         self.cache_key = self._generate_cache_key(**kwargs)
 
     @abstractmethod
-    def _generate_cache_key(self, **kwargs: Any) -> str:
+    def _generate_cache_key(self, **kwargs) -> str:
         """Generate cache key based on subclass implementation"""
         pass
 
@@ -28,11 +28,11 @@ class ProviderCredentialsCache(ABC):
                 return None
         return None
 
-    def set(self, config: dict[str, Any]) -> None:
+    def set(self, config: dict[str, Any]):
         """Cache provider credentials"""
         redis_client.setex(self.cache_key, 86400, json.dumps(config))
 
-    def delete(self) -> None:
+    def delete(self):
         """Delete cached provider credentials"""
         redis_client.delete(self.cache_key)
 
@@ -48,7 +48,7 @@ class SingletonProviderCredentialsCache(ProviderCredentialsCache):
         )
 
     @override
-    def _generate_cache_key(self, **kwargs: Any) -> str:
+    def _generate_cache_key(self, **kwargs) -> str:
         tenant_id = kwargs["tenant_id"]
         provider_type = kwargs["provider_type"]
         identity_name = kwargs["provider_identity"]
@@ -63,7 +63,7 @@ class ToolProviderCredentialsCache(ProviderCredentialsCache):
         super().__init__(tenant_id=tenant_id, provider=provider, credential_id=credential_id)
 
     @override
-    def _generate_cache_key(self, **kwargs: Any) -> str:
+    def _generate_cache_key(self, **kwargs) -> str:
         tenant_id = kwargs["tenant_id"]
         provider = kwargs["provider"]
         credential_id = kwargs["credential_id"]
@@ -77,10 +77,10 @@ class NoOpProviderCredentialCache:
         """Get cached provider credentials"""
         return None
 
-    def set(self, config: dict[str, Any]) -> None:
+    def set(self, config: dict[str, Any]):
         """Cache provider credentials"""
         pass
 
-    def delete(self) -> None:
+    def delete(self):
         """Delete cached provider credentials"""
         pass

api/core/helper/provider_encryption.py

@@ -125,7 +125,5 @@ class ProviderConfigEncrypter:
         return data
 
 
-def create_provider_encrypter(
-    tenant_id: str, config: list[BasicProviderConfig], cache: ProviderConfigCache
-) -> tuple[ProviderConfigEncrypter, ProviderConfigCache]:
+def create_provider_encrypter(tenant_id: str, config: list[BasicProviderConfig], cache: ProviderConfigCache):
     return ProviderConfigEncrypter(tenant_id=tenant_id, config=config, provider_config_cache=cache), cache

api/core/helper/tool_parameter_cache.py

@@ -37,11 +37,11 @@ class ToolParameterCache:
         else:
             return None
 
-    def set(self, parameters: dict[str, Any]) -> None:
+    def set(self, parameters: dict[str, Any]):
         """Cache model provider credentials."""
         redis_client.setex(self.cache_key, 86400, json.dumps(parameters))
 
-    def delete(self) -> None:
+    def delete(self):
         """
         Delete cached model provider credentials.
 

api/core/helper/trace_id_helper.py

@@ -61,7 +61,7 @@ def get_external_trace_id(request: Any) -> str | None:
     return None
 
 
-def extract_external_trace_id_from_args(args: Mapping[str, Any]) -> dict[str, Any]:
+def extract_external_trace_id_from_args(args: Mapping[str, Any]):
     """
     Extract 'external_trace_id' from args.
 

api/core/logging/__init__.py

@@ -2,13 +2,9 @@
 
 from core.logging.context import (
     clear_request_context,
-    get_identity_context,
     get_request_id,
     get_trace_id,
     init_request_context,
-    request_logging_context,
-    reset_request_context,
-    set_identity_context,
 )
 from core.logging.filters import IdentityContextFilter, TraceContextFilter
 from core.logging.structured_formatter import StructuredJSONFormatter
@@ -18,11 +14,7 @@ __all__ = [
     "StructuredJSONFormatter",
     "TraceContextFilter",
     "clear_request_context",
-    "get_identity_context",
     "get_request_id",
     "get_trace_id",
     "init_request_context",
-    "request_logging_context",
-    "reset_request_context",
-    "set_identity_context",
 ]

api/core/logging/context.py

@@ -5,16 +5,10 @@ using Python's contextvars for thread-safe and async-safe storage.
 """
 
 import uuid
-from collections.abc import Generator
-from contextlib import contextmanager
-from contextvars import ContextVar, Token
-
-type IdentityContext = tuple[str, str, str]
-type LoggingContextTokens = tuple[Token[str], Token[str], Token[IdentityContext]]
+from contextvars import ContextVar
 
 _request_id: ContextVar[str] = ContextVar("log_request_id", default="")
 _trace_id: ContextVar[str] = ContextVar("log_trace_id", default="")
-_identity: ContextVar[IdentityContext] = ContextVar("log_identity", default=("", "", ""))
 
 
 def get_request_id() -> str:
@@ -27,49 +21,15 @@ def get_trace_id() -> str:
     return _trace_id.get()
 
 
-def get_identity_context() -> IdentityContext:
-    """Get the immutable tenant, user, and user-type snapshot for logging."""
-    return _identity.get()
-
-
-def set_identity_context(
-    *, tenant_id: str | None = None, user_id: str | None = None, user_type: str | None = None
-) -> None:
-    """Set primitive identity values already resolved by an authentication boundary."""
-    _identity.set((tenant_id or "", user_id or "", user_type or ""))
-
-
-def init_request_context() -> LoggingContextTokens:
-    """Initialize request context and discard identity left by earlier work."""
+def init_request_context() -> None:
+    """Initialize request context. Call at start of each request."""
     req_id = uuid.uuid4().hex[:10]
     trace_id = uuid.uuid5(uuid.NAMESPACE_DNS, req_id).hex
-    return (
-        _request_id.set(req_id),
-        _trace_id.set(trace_id),
-        _identity.set(("", "", "")),
-    )
-
-
-def reset_request_context(tokens: LoggingContextTokens) -> None:
-    """Restore the logging context that preceded ``init_request_context``."""
-    request_id_token, trace_id_token, identity_token = tokens
-    _identity.reset(identity_token)
-    _trace_id.reset(trace_id_token)
-    _request_id.reset(request_id_token)
-
-
-@contextmanager
-def request_logging_context() -> Generator[None, None, None]:
-    """Run work in an isolated logging context and restore its caller afterward."""
-    tokens = init_request_context()
-    try:
-        yield
-    finally:
-        reset_request_context(tokens)
+    _request_id.set(req_id)
+    _trace_id.set(trace_id)
 
 
 def clear_request_context() -> None:
-    """Clear request context at a request or task lifecycle boundary."""
+    """Clear request context. Call at end of request (optional)."""
     _request_id.set("")
     _trace_id.set("")
-    _identity.set(("", "", ""))

api/core/logging/filters.py

@@ -4,7 +4,10 @@ import contextlib
 import logging
 from typing import override
 
-from core.logging.context import get_identity_context, get_request_id, get_trace_id
+import flask
+
+from core.logging.context import get_request_id, get_trace_id
+from core.logging.structured_formatter import IdentityDict
 
 
 class TraceContextFilter(logging.Filter):
@@ -48,13 +51,49 @@ class TraceContextFilter(logging.Filter):
 
 
 class IdentityContextFilter(logging.Filter):
-    """Add an identity snapshot without invoking authentication or database work.
-
-    Logging can run while other libraries hold internal locks, so this filter must
-    only read primitive ContextVar values populated by authentication boundaries.
+    """
+    Filter that adds user identity context to log records.
+    Extracts tenant_id, user_id, and user_type from Flask-Login current_user.
     """
 
     @override
     def filter(self, record: logging.LogRecord) -> bool:
-        record.tenant_id, record.user_id, record.user_type = get_identity_context()
+        identity = self._extract_identity()
+        record.tenant_id = identity.get("tenant_id", "")
+        record.user_id = identity.get("user_id", "")
+        record.user_type = identity.get("user_type", "")
         return True
+
+    def _extract_identity(self) -> IdentityDict:
+        """Extract identity from current_user if in request context."""
+        try:
+            if not flask.has_request_context():
+                return {}
+            from flask_login import current_user
+
+            # Check if user is authenticated using the proxy
+            if not current_user.is_authenticated:
+                return {}
+
+            # Access the underlying user object
+            user = current_user
+
+            from models import Account
+            from models.model import EndUser
+
+            identity: IdentityDict = {}
+
+            match user:
+                case Account():
+                    if user.current_tenant_id:
+                        identity["tenant_id"] = user.current_tenant_id
+                    identity["user_id"] = user.id
+                    identity["user_type"] = "account"
+                case EndUser():
+                    identity["tenant_id"] = user.tenant_id
+                    identity["user_id"] = user.id
+                    identity["user_type"] = user.type or "end_user"
+
+            return identity
+        except Exception:
+            return {}

api/core/plugin/backwards_invocation/app.py

@@ -76,11 +76,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
         if not user_id:
             user = EndUserService.get_or_create_end_user(app)
         else:
-            try:
-                user = cls._get_user(user_id, app)
-            except ValueError:
-                # Plugins such as WeCom Bot pass external sender IDs rather than EndUser UUIDs.
-                user = EndUserService.get_or_create_end_user(app, user_id=user_id)
+            user = cls._get_user(user_id, app)
 
         conversation_id = conversation_id or ""
 
@@ -230,13 +226,6 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
                 EndUser.app_id == app.id,
             )
             user = session.scalar(stmt)
-            if not user:
-                stmt = select(EndUser).where(
-                    EndUser.session_id == user_id,
-                    EndUser.tenant_id == app.tenant_id,
-                    EndUser.app_id == app.id,
-                )
-                user = session.scalar(stmt)
             if not user:
                 stmt = select(Account).where(
                     Account.id == user_id,

api/core/plugin/entities/plugin_daemon.py

@@ -228,7 +228,7 @@ class CredentialType(enum.StrEnum):
     OAUTH2 = "oauth2"
     UNAUTHORIZED = "unauthorized"
 
-    def get_name(self) -> str:
+    def get_name(self):
         if self == CredentialType.API_KEY:
             return "API KEY"
         elif self == CredentialType.OAUTH2:

api/core/plugin/plugin_service.py

@@ -4,8 +4,6 @@ This module owns plugin daemon management calls that are shared by API services
 and core runtimes. Plugin model provider discovery is cached here, alongside
 plugin install, uninstall, and upgrade invalidation, so all cache mutations for
 plugin-owned provider metadata stay tenant-scoped and in one place.
-Provider cache payloads may be stored as prefixed zstd bytes; readers also
-accept legacy plain JSON payloads for rolling upgrades and existing Redis keys.
 
 The console plugin list also normalizes endpoint setup counters against live
 endpoint records. Some plugin daemon builds return stale ``endpoints_*``
@@ -16,15 +14,12 @@ metadata.
 
 import logging
 import time
-from collections.abc import Iterator, Mapping, Sequence
-from contextlib import contextmanager
+from collections.abc import Mapping, Sequence
 from mimetypes import guess_type
-from typing import Literal, Protocol, cast
+from typing import ClassVar
 
-import zstandard
 from pydantic import BaseModel, TypeAdapter, ValidationError
 from redis import RedisError
-from redis.exceptions import LockError
 from sqlalchemy import delete, select, update
 from sqlalchemy.orm import Session
 from yarl import URL
@@ -72,18 +67,14 @@ logger = logging.getLogger(__name__)
 _provider_entities_adapter: TypeAdapter[list[ProviderEntity]] = TypeAdapter(list[ProviderEntity])
 
 
-class _RedisLock(Protocol):
-    def acquire(self, *, blocking: bool = True, blocking_timeout: float | None = None) -> bool: ...
-
-    def release(self) -> None: ...
-
-
 class PluginService:
+    _plugin_model_providers_memory_cache: ClassVar[dict[str, tuple[int, float, tuple[ProviderEntity, ...]]]] = {}
+
     class LatestPluginCache(BaseModel):
         plugin_id: str
         version: str
         unique_identifier: str
-        status: Literal["active", "deleted"]
+        status: str
         deprecated_reason: str
         alternative_plugin_id: str
 
@@ -91,13 +82,6 @@ class PluginService:
     REDIS_TTL = 60 * 5  # 5 minutes
     PLUGIN_MODEL_PROVIDERS_REDIS_KEY_PREFIX = "plugin_model_providers:tenant_id:"
     PLUGIN_MODEL_PROVIDERS_GENERATION_REDIS_KEY_PREFIX = "plugin_model_providers_generation:tenant_id:"
-    PLUGIN_MODEL_PROVIDERS_LOCK_REDIS_KEY_PREFIX = "plugin_model_providers_refresh_lock:tenant_id:"
-    PLUGIN_MODEL_PROVIDERS_REMOTE_DEBUG_REDIS_KEY_PREFIX = "plugin_model_providers_remote_debug:tenant_id:"
-    PLUGIN_MODEL_PROVIDERS_LOCK_TTL = 30
-    PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT = 2.0
-    PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL = 0.05
-    PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX = b"\x00dify-plugin-model-providers-zstd-v1:"
-    PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_MIN_BYTES = 64 * 1024
     PLUGIN_INSTALL_TASK_TERMINAL_STATUSES = (PluginInstallTaskStatus.Success, PluginInstallTaskStatus.Failed)
     # Mirror the detail-panel endpoint query size so list reconciliation and
     # the visible endpoint drawer exercise the same daemon pagination path.
@@ -114,14 +98,6 @@ class PluginService:
     def _get_plugin_model_providers_generation_cache_key(cls, tenant_id: str) -> str:
         return f"{cls.PLUGIN_MODEL_PROVIDERS_GENERATION_REDIS_KEY_PREFIX}{tenant_id}"
 
-    @classmethod
-    def _get_plugin_model_providers_lock_key(cls, tenant_id: str, generation: int) -> str:
-        return f"{cls.PLUGIN_MODEL_PROVIDERS_LOCK_REDIS_KEY_PREFIX}{tenant_id}:generation:{generation}"
-
-    @classmethod
-    def _get_plugin_model_providers_remote_debug_cache_key(cls, tenant_id: str) -> str:
-        return f"{cls.PLUGIN_MODEL_PROVIDERS_REMOTE_DEBUG_REDIS_KEY_PREFIX}{tenant_id}"
-
     @staticmethod
     def _get_provider_short_name_alias(provider: PluginModelProviderEntity) -> str:
         """
@@ -153,25 +129,8 @@ class PluginService:
         return declaration
 
     @classmethod
-    def _encode_plugin_model_providers_cache_payload(cls, payload: bytes) -> bytes:
-        if len(payload) < cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_MIN_BYTES:
-            return payload
-
-        return cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX + zstandard.compress(payload, level=1)
-
-    @classmethod
-    def _decode_plugin_model_providers_cache_payload(cls, payload: bytes | bytearray | str) -> bytes | bytearray | str:
-        if isinstance(payload, str):
-            return payload
-
-        prefix = cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX
-        if not payload.startswith(prefix):
-            return payload
-
-        try:
-            return zstandard.decompress(payload[len(prefix) :])
-        except zstandard.ZstdError as exc:
-            raise ValueError("Invalid compressed plugin model providers cache payload.") from exc
+    def _copy_provider_entities(cls, providers: Sequence[ProviderEntity]) -> tuple[ProviderEntity, ...]:
+        return tuple(provider.model_copy(deep=True) for provider in providers)
 
     @classmethod
     def _load_plugin_model_providers_generation(cls, tenant_id: str) -> int | None:
@@ -204,35 +163,76 @@ class PluginService:
             return None
 
     @classmethod
-    def _load_cached_plugin_model_providers_for_generation(
-        cls, tenant_id: str, generation: int | None
-    ) -> tuple[tuple[ProviderEntity, ...] | None, bool]:
-        if generation is None:
-            return None, False
+    def _load_in_memory_plugin_model_providers(
+        cls, memory_cache_key: str, generation: int
+    ) -> tuple[ProviderEntity, ...] | None:
+        cached_entry = cls._plugin_model_providers_memory_cache.get(memory_cache_key)
+        if cached_entry is None:
+            return None
 
-        cache_keys = [cls._get_plugin_model_providers_cache_key(tenant_id, generation)]
+        cached_generation, expires_at, providers = cached_entry
+        if cached_generation != generation or time.monotonic() >= expires_at:
+            cls._plugin_model_providers_memory_cache.pop(memory_cache_key, None)
+            return None
+
+        return cls._copy_provider_entities(providers)
+
+    @classmethod
+    def _store_in_memory_plugin_model_providers(
+        cls, memory_cache_key: str, generation: int, providers: Sequence[ProviderEntity]
+    ) -> None:
+        ttl = dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL
+        if ttl <= 0:
+            cls._plugin_model_providers_memory_cache.pop(memory_cache_key, None)
+            return
+
+        cls._plugin_model_providers_memory_cache[memory_cache_key] = (
+            generation,
+            time.monotonic() + ttl,
+            cls._copy_provider_entities(providers),
+        )
+
+    @classmethod
+    def _load_cached_plugin_model_providers(
+        cls, tenant_id: str, *, client: PluginModelClient | None = None
+    ) -> tuple[ProviderEntity, ...] | None:
+        generation = cls._load_plugin_model_providers_generation(tenant_id)
+        if generation is not None:
+            in_memory_cached_providers = cls._load_in_memory_plugin_model_providers(tenant_id, generation)
+            if in_memory_cached_providers is not None:
+                return in_memory_cached_providers
+
+        cache_keys = []
+        if generation is not None:
+            cache_keys.append(cls._get_plugin_model_providers_cache_key(tenant_id, generation))
+            if generation == 0:
+                cache_keys.append(cls._get_plugin_model_providers_cache_key(tenant_id))
+
+        if not cache_keys:
+            return None
 
         try:
             cached_provider_entries = redis_client.mget(cache_keys)
-        except (LockError, RedisError, RuntimeError):
+        except (RedisError, RuntimeError):
             logger.warning("Failed to read cached plugin model providers for tenant %s.", tenant_id, exc_info=True)
-            return None, False
+            return None
 
         if len(cached_provider_entries) != len(cache_keys):
             logger.warning(
                 "Unexpected cached plugin model providers response size for tenant %s.",
                 tenant_id,
             )
-            return None, False
+            return None
 
         for cache_key, cached_providers in zip(cache_keys, cached_provider_entries):
             if not cached_providers:
                 continue
 
             try:
-                payload = cls._decode_plugin_model_providers_cache_payload(cached_providers)
-                providers = tuple(_provider_entities_adapter.validate_json(payload))
-                return providers, True
+                providers = tuple(_provider_entities_adapter.validate_json(cached_providers))
+                if generation is not None:
+                    cls._store_in_memory_plugin_model_providers(tenant_id, generation, providers)
+                return providers
             except (TypeError, ValueError, ValidationError):
                 logger.warning(
                     "Invalid cached plugin model providers for tenant %s; deleting cache key %s.",
@@ -249,7 +249,7 @@ class PluginService:
                         exc_info=True,
                     )
 
-        return None, True
+        return None
 
     @classmethod
     def _store_cached_plugin_model_providers(
@@ -257,199 +257,15 @@ class PluginService:
     ) -> None:
         cache_key = cls._get_plugin_model_providers_cache_key(tenant_id, generation)
         try:
-            payload = cls._encode_plugin_model_providers_cache_payload(
-                _provider_entities_adapter.dump_json(list(providers))
-            )
+            payload = _provider_entities_adapter.dump_json(list(providers)).decode("utf-8")
             redis_client.setex(cache_key, dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL, payload)
         except (RedisError, RuntimeError):
             logger.warning("Failed to cache plugin model providers for tenant %s.", tenant_id, exc_info=True)
 
-    @classmethod
-    def _get_remote_model_plugin_cache_marker(cls, plugins: Sequence[PluginEntity]) -> str | None:
-        remote_model_plugins = sorted(
-            f"{plugin.plugin_id}:{plugin.plugin_unique_identifier}"
-            for plugin in plugins
-            if plugin.source == PluginInstallationSource.Remote
-        )
-        if not remote_model_plugins:
-            return None
-
-        return "\n".join(remote_model_plugins)
-
-    @classmethod
-    def _load_cached_remote_model_plugin_marker(cls, tenant_id: str) -> str | None:
-        cache_key = cls._get_plugin_model_providers_remote_debug_cache_key(tenant_id)
-        try:
-            cached_marker = redis_client.get(cache_key)
-        except (RedisError, RuntimeError):
-            logger.warning("Failed to read remote debug model plugin marker for tenant %s.", tenant_id, exc_info=True)
-            return None
-
-        if cached_marker is None:
-            return None
-        if isinstance(cached_marker, bytes):
-            try:
-                return cached_marker.decode()
-            except UnicodeDecodeError:
-                logger.warning(
-                    "Invalid remote debug model plugin marker for tenant %s; deleting cache marker.",
-                    tenant_id,
-                    exc_info=True,
-                )
-                try:
-                    redis_client.delete(cache_key)
-                except (RedisError, RuntimeError):
-                    logger.warning(
-                        "Failed to delete invalid remote debug model plugin marker for tenant %s.",
-                        tenant_id,
-                        exc_info=True,
-                    )
-                return None
-        if isinstance(cached_marker, str):
-            return cached_marker
-
-        logger.warning("Invalid remote debug model plugin marker for tenant %s; deleting cache marker.", tenant_id)
-        try:
-            redis_client.delete(cache_key)
-        except (RedisError, RuntimeError):
-            logger.warning(
-                "Failed to delete invalid remote debug model plugin marker for tenant %s.",
-                tenant_id,
-                exc_info=True,
-            )
-        return None
-
-    @classmethod
-    def _store_cached_remote_model_plugin_marker(cls, tenant_id: str, marker: str | None) -> None:
-        cache_key = cls._get_plugin_model_providers_remote_debug_cache_key(tenant_id)
-        try:
-            if marker is None:
-                redis_client.delete(cache_key)
-            else:
-                redis_client.setex(cache_key, dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL, marker)
-        except (RedisError, RuntimeError):
-            logger.warning("Failed to cache remote debug model plugin marker for tenant %s.", tenant_id, exc_info=True)
-
-    @classmethod
-    def _load_cached_plugin_model_provider_plugin_ids(cls, tenant_id: str) -> set[str] | None:
-        """Return plugin ids represented by the current provider cache, or None when no usable cache exists."""
-        generation = cls._load_plugin_model_providers_generation(tenant_id)
-        cached_providers, _ = cls._load_cached_plugin_model_providers_for_generation(tenant_id, generation)
-        if cached_providers is None:
-            return None
-
-        plugin_ids: set[str] = set()
-        for provider in cached_providers:
-            last_slash = provider.provider.rfind("/")
-            if last_slash > 0:
-                plugin_ids.add(provider.provider[:last_slash])
-
-        return plugin_ids
-
-    @classmethod
-    def _should_invalidate_model_provider_cache_for_remote_model_plugins(
-        cls,
-        tenant_id: str,
-        plugins: Sequence[PluginEntity],
-    ) -> bool:
-        remote_model_plugin_marker = cls._get_remote_model_plugin_cache_marker(plugins)
-        cached_remote_model_plugin_marker = cls._load_cached_remote_model_plugin_marker(tenant_id)
-        if remote_model_plugin_marker is None:
-            return cached_remote_model_plugin_marker is not None
-
-        if remote_model_plugin_marker != cached_remote_model_plugin_marker:
-            return True
-
-        remote_model_plugin_ids = {
-            plugin.plugin_id for plugin in plugins if plugin.source == PluginInstallationSource.Remote
-        }
-        cached_plugin_ids = cls._load_cached_plugin_model_provider_plugin_ids(tenant_id)
-        if cached_plugin_ids is None:
-            return False
-
-        return not remote_model_plugin_ids.issubset(cached_plugin_ids)
-
-    @classmethod
-    @contextmanager
-    def _plugin_model_providers_refresh_lock(
-        cls, tenant_id: str, generation: int, *, wait_timeout: float
-    ) -> Iterator[bool]:
-        lock_key = cls._get_plugin_model_providers_lock_key(tenant_id, generation)
-        try:
-            refresh_lock: _RedisLock = redis_client.lock(
-                lock_key,
-                timeout=cls.PLUGIN_MODEL_PROVIDERS_LOCK_TTL,
-                sleep=cls.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL,
-            )
-        except (RedisError, RuntimeError):
-            logger.warning(
-                "Failed to create plugin model providers refresh lock for tenant %s.",
-                tenant_id,
-                exc_info=True,
-            )
-            yield False
-            return
-
-        try:
-            lock_acquired = refresh_lock.acquire(blocking=True, blocking_timeout=wait_timeout)
-        except LockError:
-            logger.warning(
-                "Provider refresh lock timed out; direct daemon fallback. tenant_id=%s generation=%s",
-                tenant_id,
-                generation,
-                exc_info=True,
-            )
-            yield False
-            return
-        except (RedisError, RuntimeError):
-            # Redis failures should not block provider discovery; callers fetch directly from the daemon.
-            logger.warning(
-                "Failed to acquire plugin model providers refresh lock for tenant %s.",
-                tenant_id,
-                exc_info=True,
-            )
-            yield False
-            return
-
-        if not lock_acquired:
-            logger.warning(
-                "Provider refresh lock timed out; direct daemon fallback. tenant_id=%s generation=%s",
-                tenant_id,
-                generation,
-            )
-            yield False
-            return
-
-        try:
-            yield True
-        finally:
-            try:
-                refresh_lock.release()
-            except (LockError, RedisError, RuntimeError):
-                # Release failures must not hide the daemon result or the original exception.
-                logger.warning(
-                    "Failed to release plugin model providers refresh lock for tenant %s generation %s.",
-                    tenant_id,
-                    generation,
-                    exc_info=True,
-                )
-
-    @classmethod
-    def _fetch_and_cache_plugin_model_providers(
-        cls, tenant_id: str, client: PluginModelClient | None, *, refresh_generation: int | None
-    ) -> tuple[ProviderEntity, ...]:
-        model_client = client or PluginModelClient()
-        providers = tuple(
-            cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id)
-        )
-        generation = cls._load_plugin_model_providers_generation(tenant_id)
-        if generation is not None and generation == refresh_generation:
-            cls._store_cached_plugin_model_providers(tenant_id, generation, providers)
-        return providers
-
     @classmethod
     def invalidate_plugin_model_providers_cache(cls, tenant_id: str) -> None:
-        """Invalidate tenant-scoped provider metadata stored in Redis."""
+        """Invalidate tenant-scoped provider metadata across Redis and worker-local mirrors."""
+        cls._plugin_model_providers_memory_cache.pop(tenant_id, None)
         cache_key = cls._get_plugin_model_providers_cache_key(tenant_id)
         generation_key = cls._get_plugin_model_providers_generation_cache_key(tenant_id)
         try:
@@ -471,68 +287,21 @@ class PluginService:
         are intentionally owned by this service so tenant isolation and cache
         expiry are handled in one place.
         """
-        deadline = time.monotonic() + cls.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT
+        cached_providers = cls._load_cached_plugin_model_providers(tenant_id, client=client)
+        if cached_providers is not None:
+            return cached_providers
 
-        while True:
-            generation = cls._load_plugin_model_providers_generation(tenant_id)
-            cached_providers, cache_available = cls._load_cached_plugin_model_providers_for_generation(
-                tenant_id, generation
-            )
-            if cached_providers is not None:
-                return cached_providers
-
-            if generation is None or not cache_available:
-                return cls._fetch_and_cache_plugin_model_providers(
-                    tenant_id,
-                    client,
-                    refresh_generation=generation,
-                )
-
-            wait_timeout = deadline - time.monotonic()
-            if wait_timeout < 0:
-                logger.warning(
-                    "Provider refresh lock timed out; direct daemon fallback. tenant_id=%s generation=%s",
-                    tenant_id,
-                    generation,
-                )
-                return cls._fetch_and_cache_plugin_model_providers(
-                    tenant_id,
-                    client,
-                    refresh_generation=generation,
-                )
-
-            with cls._plugin_model_providers_refresh_lock(
-                tenant_id,
-                generation,
-                wait_timeout=wait_timeout,
-            ) as lock_acquired:
-                if not lock_acquired:
-                    return cls._fetch_and_cache_plugin_model_providers(
-                        tenant_id,
-                        client,
-                        refresh_generation=generation,
-                    )
-
-                latest_generation = cls._load_plugin_model_providers_generation(tenant_id)
-                cached_providers, cache_available = cls._load_cached_plugin_model_providers_for_generation(
-                    tenant_id, latest_generation
-                )
-                if cached_providers is not None:
-                    return cached_providers
-                if latest_generation is None or not cache_available:
-                    return cls._fetch_and_cache_plugin_model_providers(
-                        tenant_id,
-                        client,
-                        refresh_generation=latest_generation,
-                    )
-                if latest_generation != generation:
-                    continue
-
-                return cls._fetch_and_cache_plugin_model_providers(
-                    tenant_id,
-                    client,
-                    refresh_generation=generation,
-                )
+        model_client = client or PluginModelClient()
+        providers = tuple(
+            cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id)
+        )
+        if not providers:
+            return providers
+        generation = cls._load_plugin_model_providers_generation(tenant_id)
+        if generation is not None:
+            cls._store_in_memory_plugin_model_providers(tenant_id, generation, providers)
+            cls._store_cached_plugin_model_providers(tenant_id, generation, providers)
+        return providers
 
     @staticmethod
     def fetch_latest_plugin_version(plugin_ids: Sequence[str]) -> Mapping[str, LatestPluginCache | None]:
@@ -571,7 +340,7 @@ class PluginService:
                             plugin_id=plugin_id,
                             version=manifest.latest_version,
                             unique_identifier=manifest.latest_package_identifier,
-                            status=cast(Literal["active", "deleted"], manifest.status),
+                            status=manifest.status,
                             deprecated_reason=manifest.deprecated_reason,
                             alternative_plugin_id=manifest.alternative_plugin_id,
                         )
@@ -681,21 +450,7 @@ class PluginService:
         This keeps pagination usable before category is persisted on installation rows.
         """
         manager = PluginInstaller()
-        plugins = manager.list_plugins_by_category(tenant_id, category, page, page_size)
-        if category == PluginCategory.Model:
-            should_invalidate_model_provider_cache = (
-                PluginService._should_invalidate_model_provider_cache_for_remote_model_plugins(
-                    tenant_id,
-                    plugins.list,
-                )
-            )
-            if should_invalidate_model_provider_cache:
-                PluginService.invalidate_plugin_model_providers_cache(tenant_id)
-
-            remote_model_plugin_marker = PluginService._get_remote_model_plugin_cache_marker(plugins.list)
-            PluginService._store_cached_remote_model_plugin_marker(tenant_id, remote_model_plugin_marker)
-
-        return plugins
+        return manager.list_plugins_by_category(tenant_id, category, page, page_size)
 
     @staticmethod
     def _normalize_endpoint_count(value: object) -> int:

api/core/rag/datasource/vdb/vector_backend_registry.py

@@ -44,13 +44,11 @@ def _load_plugin_factory(vector_type: str) -> type[AbstractVectorFactory] | None
     for ep in _vector_backend_entry_points():
         if ep.name != vector_type:
             continue
-        logger.info("loading vector backend entry point %s (%s)", ep.name, ep.value)
         try:
             loaded = ep.load()
         except Exception:
             logger.exception("Failed to load vector backend entry point %s", ep.name)
             raise
-        logger.info("vector backend entry point %s loaded", ep.name)
         return loaded  # type: ignore[return-value]
     return None
 
@@ -79,7 +77,6 @@ def get_vector_factory_class(vector_type: str) -> type[AbstractVectorFactory]:
     if vector_type in _VECTOR_FACTORY_CACHE:
         return _VECTOR_FACTORY_CACHE[vector_type]
 
-    logger.info("vector backend cache miss, discovering entry points: %s", vector_type)
     plugin_cls = _load_plugin_factory(vector_type)
     if plugin_cls is not None:
         _VECTOR_FACTORY_CACHE[vector_type] = plugin_cls

api/core/rag/datasource/vdb/vector_factory.py

@@ -144,20 +144,7 @@ class Vector:
         if not vector_type:
             raise ValueError("Vector store must be specified.")
 
-        logger.info(
-            "resolving vector backend factory: type=%s tenant_id=%s dataset_id=%s",
-            vector_type,
-            self._dataset.tenant_id,
-            self._dataset.id,
-        )
         vector_factory_cls = self.get_vector_factory(vector_type)
-        logger.info(
-            "vector backend factory resolved: type=%s cls=%s tenant_id=%s dataset_id=%s",
-            vector_type,
-            vector_factory_cls.__name__,
-            self._dataset.tenant_id,
-            self._dataset.id,
-        )
         return vector_factory_cls().init_vector(self._dataset, self._attributes, self._embeddings)
 
     @staticmethod

api/core/rag/retrieval/dataset_retrieval.py

@@ -1030,10 +1030,6 @@ class DatasetRetrieval:
     ):
         """
         Persist dataset query audit rows for retrieval requests.
-
-        Query audit logging is a side effect of retrieval. Keep it in an
-        independent transaction so failures or commits here do not affect the
-        request/workflow transaction that called the retriever.
         """
         if not query and not attachment_ids:
             return
@@ -1045,9 +1041,6 @@ class DatasetRetrieval:
                 app_id,
             )
             return
-        created_by_role = self._resolve_creator_user_role(user_from)
-        if created_by_role is None:
-            return
         dataset_queries = []
         for dataset_id in dataset_ids:
             contents = []
@@ -1062,16 +1055,13 @@ class DatasetRetrieval:
                     content=json.dumps(contents),
                     source=DatasetQuerySource.APP,
                     source_app_id=app_id,
-                    created_by_role=created_by_role,
+                    created_by_role=CreatorUserRole(user_from),
                     created_by=created_by,
                 )
                 dataset_queries.append(dataset_query)
-
-        if not dataset_queries:
-            return
-
-        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
-            session.add_all(dataset_queries)
+            if dataset_queries:
+                db.session.add_all(dataset_queries)
+        db.session.commit()
 
     def _retriever(
         self,

api/core/rbac/__init__.py

@@ -1,3 +1,3 @@
-from core.rbac.entities import RBACPermission, RBACResourceScope, RBACResourceWhitelistScope
+from core.rbac.entities import RBACPermission, RBACResourceScope
 
-__all__ = ["RBACPermission", "RBACResourceScope", "RBACResourceWhitelistScope"]
+__all__ = ["RBACPermission", "RBACResourceScope"]

api/core/rbac/entities.py

@@ -13,14 +13,6 @@ class RBACResourceScope(StrEnum):
     WORKSPACE = "workspace"
 
 
-class RBACResourceWhitelistScope(StrEnum):
-    """Whitelist scopes accepted by RBAC app and dataset access config APIs."""
-
-    ALL = "all"
-    SPECIFIC = "specific"
-    ONLY_ME = "only_me"
-
-
 class RBACPermission(StrEnum):
     """Permission points (RBAC scenes) checked by ``rbac_permission_required``.
 

api/core/repositories/celery_workflow_execution_repository.py

@@ -13,6 +13,7 @@ from sqlalchemy.orm import sessionmaker
 
 from core.repositories.factory import WorkflowExecutionRepository
 from graphon.entities import WorkflowExecution
+from libs.helper import extract_tenant_id
 from models import Account, CreatorUserRole, EndUser
 from models.enums import WorkflowRunTriggeredFrom
 from tasks.workflow_execution_tasks import (
@@ -46,7 +47,6 @@ class CeleryWorkflowExecutionRepository(WorkflowExecutionRepository):
     def __init__(
         self,
         session_factory: sessionmaker | Engine,
-        tenant_id: str,
         user: Account | EndUser,
         app_id: str | None,
         triggered_from: WorkflowRunTriggeredFrom | None,
@@ -56,8 +56,7 @@ class CeleryWorkflowExecutionRepository(WorkflowExecutionRepository):
 
         Args:
             session_factory: SQLAlchemy sessionmaker or engine for fallback operations
-            tenant_id: Tenant that owns the workflow execution
-            user: Account or EndUser used for creator attribution
+            user: Account or EndUser object containing tenant_id, user ID, and role information
             app_id: App ID for filtering by application (can be None)
             triggered_from: Source of the execution trigger (DEBUGGING or APP_RUN)
         """
@@ -72,8 +71,10 @@ class CeleryWorkflowExecutionRepository(WorkflowExecutionRepository):
                     f"Invalid session_factory type {type(session_factory).__name__}; expected sessionmaker or Engine"
                 )
 
+        # Extract tenant_id from user
+        tenant_id = extract_tenant_id(user)
         if not tenant_id:
-            raise ValueError("tenant_id is required")
+            raise ValueError("User must have a tenant_id or current_tenant_id")
         self._tenant_id = tenant_id
 
         # Store app context

api/core/repositories/celery_workflow_node_execution_repository.py

@@ -17,6 +17,7 @@ from core.repositories.factory import (
     WorkflowNodeExecutionRepository,
 )
 from graphon.entities import WorkflowNodeExecution
+from libs.helper import extract_tenant_id
 from models import Account, CreatorUserRole, EndUser
 from models.workflow import WorkflowNodeExecutionTriggeredFrom
 from tasks.workflow_node_execution_tasks import (
@@ -53,7 +54,6 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
     def __init__(
         self,
         session_factory: sessionmaker | Engine,
-        tenant_id: str,
         user: Account | EndUser,
         app_id: str | None,
         triggered_from: WorkflowNodeExecutionTriggeredFrom | None,
@@ -63,8 +63,7 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
 
         Args:
             session_factory: SQLAlchemy sessionmaker or engine for fallback operations
-            tenant_id: Tenant that owns the workflow node execution
-            user: Account or EndUser used for creator attribution
+            user: Account or EndUser object containing tenant_id, user ID, and role information
             app_id: App ID for filtering by application (can be None)
             triggered_from: Source of the execution trigger (SINGLE_STEP or WORKFLOW_RUN)
         """
@@ -79,8 +78,10 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
                     f"Invalid session_factory type {type(session_factory).__name__}; expected sessionmaker or Engine"
                 )
 
+        # Extract tenant_id from user
+        tenant_id = extract_tenant_id(user)
         if not tenant_id:
-            raise ValueError("tenant_id is required")
+            raise ValueError("User must have a tenant_id or current_tenant_id")
         self._tenant_id = tenant_id
 
         # Store app context

api/core/repositories/factory.py

@@ -62,7 +62,6 @@ class DifyCoreRepositoryFactory:
     def create_workflow_execution_repository(
         cls,
         session_factory: sessionmaker | Engine,
-        tenant_id: str,
         user: Account | EndUser,
         app_id: str,
         triggered_from: WorkflowRunTriggeredFrom,
@@ -72,8 +71,7 @@ class DifyCoreRepositoryFactory:
 
         Args:
             session_factory: SQLAlchemy sessionmaker or engine
-            tenant_id: Tenant that owns the workflow execution
-            user: Account or EndUser used for creator attribution
+            user: Account or EndUser object
             app_id: Application ID
             triggered_from: Source of the execution trigger
 
@@ -89,7 +87,6 @@ class DifyCoreRepositoryFactory:
             repository_class = import_string(class_path)
             return repository_class(
                 session_factory=session_factory,
-                tenant_id=tenant_id,
                 user=user,
                 app_id=app_id,
                 triggered_from=triggered_from,
@@ -101,7 +98,6 @@ class DifyCoreRepositoryFactory:
     def create_workflow_node_execution_repository(
         cls,
         session_factory: sessionmaker | Engine,
-        tenant_id: str,
         user: Account | EndUser,
         app_id: str,
         triggered_from: WorkflowNodeExecutionTriggeredFrom,
@@ -111,8 +107,7 @@ class DifyCoreRepositoryFactory:
 
         Args:
             session_factory: SQLAlchemy sessionmaker or engine
-            tenant_id: Tenant that owns the workflow node execution
-            user: Account or EndUser used for creator attribution
+            user: Account or EndUser object
             app_id: Application ID
             triggered_from: Source of the execution trigger
 
@@ -128,7 +123,6 @@ class DifyCoreRepositoryFactory:
             repository_class = import_string(class_path)
             return repository_class(
                 session_factory=session_factory,
-                tenant_id=tenant_id,
                 user=user,
                 app_id=app_id,
                 triggered_from=triggered_from,

api/core/repositories/sqlalchemy_workflow_execution_repository.py

@@ -13,6 +13,7 @@ from core.repositories.factory import WorkflowExecutionRepository
 from graphon.entities import WorkflowExecution
 from graphon.enums import WorkflowExecutionStatus, WorkflowType
 from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
+from libs.helper import extract_tenant_id
 from models import (
     Account,
     CreatorUserRole,
@@ -39,7 +40,6 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
     def __init__(
         self,
         session_factory: sessionmaker | Engine,
-        tenant_id: str,
         user: Account | EndUser,
         app_id: str | None,
         triggered_from: WorkflowRunTriggeredFrom | None,
@@ -49,8 +49,7 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
 
         Args:
             session_factory: SQLAlchemy sessionmaker or engine for creating sessions
-            tenant_id: Tenant that owns the workflow execution
-            user: Account or EndUser used for creator attribution
+            user: Account or EndUser object containing tenant_id, user ID, and role information
             app_id: App ID for filtering by application (can be None)
             triggered_from: Source of the execution trigger (DEBUGGING or APP_RUN)
         """
@@ -65,8 +64,10 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
                     f"Invalid session_factory type {type(session_factory).__name__}; expected sessionmaker or Engine"
                 )
 
+        # Extract tenant_id from user
+        tenant_id = extract_tenant_id(user)
         if not tenant_id:
-            raise ValueError("tenant_id is required")
+            raise ValueError("User must have a tenant_id or current_tenant_id")
         self._tenant_id = tenant_id
 
         # Store app context

api/core/repositories/sqlalchemy_workflow_node_execution_repository.py

@@ -23,6 +23,7 @@ from graphon.entities import WorkflowNodeExecution
 from graphon.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
+from libs.helper import extract_tenant_id
 from libs.uuid_utils import uuidv7
 from models import (
     Account,
@@ -62,7 +63,6 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
     def __init__(
         self,
         session_factory: sessionmaker | Engine,
-        tenant_id: str,
         user: Account | EndUser,
         app_id: str | None,
         triggered_from: WorkflowNodeExecutionTriggeredFrom | None,
@@ -72,8 +72,7 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
 
         Args:
             session_factory: SQLAlchemy sessionmaker or engine for creating sessions
-            tenant_id: Tenant that owns the workflow node execution
-            user: Account or EndUser used for creator attribution
+            user: Account or EndUser object containing tenant_id, user ID, and role information
             app_id: App ID for filtering by application (can be None)
             triggered_from: Source of the execution trigger (SINGLE_STEP or WORKFLOW_RUN)
         """
@@ -88,8 +87,10 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
                     f"Invalid session_factory type {type(session_factory).__name__}; expected sessionmaker or Engine"
                 )
 
+        # Extract tenant_id from user
+        tenant_id = extract_tenant_id(user)
         if not tenant_id:
-            raise ValueError("tenant_id is required")
+            raise ValueError("User must have a tenant_id or current_tenant_id")
         self._tenant_id = tenant_id
 
         # Store app context
@@ -298,7 +299,6 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
             content=value_json.encode("utf-8"),
             mimetype="application/json",
             user=self._user,
-            tenant_id=self._tenant_id,
         )
         offload = WorkflowNodeExecutionOffload(
             id=uuidv7(),

api/core/tools/custom_tool/tool.py

@@ -102,17 +102,16 @@ class ApiTool(Tool):
             elif not isinstance(credentials["api_key_value"], str):
                 raise ToolProviderCredentialValidationError("api_key_value must be a string")
 
-            api_key_value = credentials["api_key_value"]
             if "api_key_header_prefix" in credentials:
                 api_key_header_prefix = credentials["api_key_header_prefix"]
-                if api_key_header_prefix == "basic" and api_key_value:
-                    api_key_value = f"Basic {api_key_value}"
-                elif api_key_header_prefix == "bearer" and api_key_value:
-                    api_key_value = f"Bearer {api_key_value}"
+                if api_key_header_prefix == "basic" and credentials["api_key_value"]:
+                    credentials["api_key_value"] = f"Basic {credentials['api_key_value']}"
+                elif api_key_header_prefix == "bearer" and credentials["api_key_value"]:
+                    credentials["api_key_value"] = f"Bearer {credentials['api_key_value']}"
                 elif api_key_header_prefix == "custom":
                     pass
 
-            headers[api_key_header] = api_key_value
+            headers[api_key_header] = credentials["api_key_value"]
 
         elif credentials["auth_type"] == "api_key_query":
             # For query parameter authentication, we don't add anything to headers