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/frontend-testing/SKILL.md

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

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

@@ -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/style.yml

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

README.md

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

api/controllers/console/app/app.py

@@ -1094,7 +1094,7 @@ class AppTraceApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_MONITOR)
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
     def post(self, app_model: App):
         # add app trace

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

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

api/controllers/console/app/workflow_app_log.py

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

api/controllers/console/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/workspace/endpoint.py

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

api/controllers/console/workspace/plugin.py

@@ -793,7 +793,6 @@ class PluginFetchInstallTasksApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def get(self, tenant_id: str):
@@ -811,7 +810,6 @@ class PluginFetchInstallTaskApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def get(self, tenant_id: str, task_id: str):
@@ -827,7 +825,6 @@ class PluginDeleteInstallTaskApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str, task_id: str):
@@ -843,7 +840,6 @@ class PluginDeleteAllInstallTaskItemsApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -859,7 +855,6 @@ class PluginDeleteInstallTaskItemApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str, task_id: str, identifier: str):
@@ -876,7 +871,7 @@ class PluginUpgradeFromMarketplaceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -899,7 +894,7 @@ class PluginUpgradeFromGithubApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -927,7 +922,7 @@ class PluginUninstallApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_INSTALL, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_DELETE, resource_required=False)
     @plugin_permission_required(install_required=True)
     @with_current_tenant_id
     def post(self, tenant_id: str):
@@ -995,7 +990,7 @@ class PluginFetchDynamicSelectOptionsApi(Resource):
     @setup_required
     @login_required
     @is_admin_or_owner_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MANAGE, resource_required=False)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
     @account_initialization_required
     @with_current_user
     @with_current_tenant_id

api/controllers/console/workspace/rbac.py

@@ -201,21 +201,23 @@ def _legacy_workspace_roles(
     This keeps the new `/rbac/roles` endpoint compatible with the original
     Dify role model when enterprise RBAC is disabled.
     """
-
-    legacy_roles = [
-        svc.RBACRole(
-            id=role_name,
-            tenant_id="",
-            type=svc.RBACRoleType.WORKSPACE.value,
-            category="global_system_default",
-            name=role_name,
-            description="",
-            is_builtin=True,
-            permission_keys=list(dict.fromkeys(_LEGACY_ROLE_PERMISSION_KEYS[role_name])),
-            role_tag="owner" if role_name == "owner" else "",
+    legacy_roles = []
+    for role_name in ("owner", "admin", "editor", "normal", "dataset_operator"):
+        if not dify_config.DATASET_OPERATOR_ENABLED and role_name == "dataset_operator":
+            continue
+        legacy_roles.append(
+            svc.RBACRole(
+                id=role_name,
+                tenant_id="",
+                type=svc.RBACRoleType.WORKSPACE.value,
+                category="global_system_default",
+                name=role_name,
+                description="",
+                is_builtin=True,
+                permission_keys=list(dict.fromkeys(_LEGACY_ROLE_PERMISSION_KEYS[role_name])),
+                role_tag="owner" if role_name == "owner" else "",
+            )
         )
-        for role_name in ("owner", "admin", "editor", "normal", "dataset_operator")
-    ]
 
     if not include_owner:
         legacy_roles = [r for r in legacy_roles if r.name != "owner"]

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/web/completion.py

@@ -32,6 +32,7 @@ from libs.helper import uuid_value
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 from services.app_task_service import AppTaskService
+from services.conversation_service import ConversationService
 from services.errors.llm import InvokeRateLimitError
 
 logger = logging.getLogger(__name__)
@@ -202,6 +203,12 @@ class ChatApi(WebApiResource):
         args["auto_generate_name"] = False
 
         try:
+            # Eagerly validate conversation to avoid hanging on invalid conversation_id
+            if payload.conversation_id:
+                ConversationService.get_conversation(
+                    app_model=app_model, conversation_id=payload.conversation_id, user=end_user
+                )
+
             response = AppGenerateService.generate(
                 app_model=app_model, user=end_user, args=args, invoke_from=InvokeFrom.WEB_APP, streaming=streaming
             )

api/core/hosting_configuration.py

@@ -240,7 +240,8 @@ class HostingConfiguration:
         if len(quotas) > 0:
             credentials = {
                 "dashscope_api_key": dify_config.HOSTED_TONGYI_API_KEY,
-                "use_international_endpoint": dify_config.HOSTED_TONGYI_USE_INTERNATIONAL_ENDPOINT,
+                # SNP-494: keep temporary compatibility with tongyi plugin string credential checks.
+                "use_international_endpoint": str(dify_config.HOSTED_TONGYI_USE_INTERNATIONAL_ENDPOINT).lower(),
             }
 
             return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)

api/core/rbac/entities.py

@@ -28,6 +28,8 @@ class RBACPermission(StrEnum):
     APP_IMPORT_EXPORT_DSL = "app_import_export_dsl"
     APP_EDIT = "app_edit"
     APP_MONITOR = "app_monitor"
+    APP_TRACING_CONFIG = "app_tracing_config"
+    APP_LOG_AND_ANNOTATION = "app_log_and_annotation"
     APP_DELETE = "app_delete"
     APP_ACCESS_CONFIG = "app_access_config"
 
@@ -57,7 +59,9 @@ class RBACPermission(StrEnum):
 
     PLUGIN_INSTALL = "plugin_install"
     PLUGIN_PREFERENCES = "plugin_preferences"
+    PLUGIN_MODEL_CONFIG = "plugin_model_config"
     PLUGIN_MANAGE = "plugin_manage"
+    PLUGIN_DELETE = "plugin_delete"
     PLUGIN_DEBUG = "plugin_debug"
 
     CREDENTIAL_USE = "credential_use"

api/dev/lint_response_contracts.py

@@ -2,8 +2,8 @@
 
 This checker intentionally stays conservative. It only reports a hard schema
 mismatch when both sides are statically known for the same 2xx status code:
-a documented ``@ns.response(..., Model)`` and an actual ``dump_response(Model, ...)``
-or ``Model.model_validate(...).model_dump()`` return.
+a documented ``@ns.response(..., Model)`` and an actual ``dump_response(Model, ...)``,
+``Model(...).model_dump()``, or ``Model.model_validate(...).model_dump()`` return.
 
 Raw dictionaries, raw lists, ``None`` responses, streaming helpers, missing
 response schemas, and returns with non-literal status codes are classified as
@@ -28,6 +28,7 @@ from typing import Any, Literal
 HTTP_METHODS = {"delete", "get", "head", "options", "patch", "post", "put"}
 NO_BODY_STATUSES = {HTTPStatus.NO_CONTENT.value, HTTPStatus.RESET_CONTENT.value, HTTPStatus.NOT_MODIFIED.value}
 DEFAULT_CONTROLLER_DIRS = ("controllers/console", "controllers/service_api", "controllers/web")
+IGNORE_COMMENT_MARKERS = ("response-contract:ignore",)
 
 type Classification = Literal["valid", "mismatch", "unknown", "refactorable"]
 type ActualKind = Literal[
@@ -41,6 +42,7 @@ type ActualKind = Literal[
     "unknown",
 ]
 type MethodNode = ast.FunctionDef | ast.AsyncFunctionDef
+type ModelValueSource = Literal["constructor", "model_validate"]
 
 HTTP_STATUS_NAMES = {status.name: status.value for status in HTTPStatus}
 HTTP_STATUS_NAMES.update({f"HTTP_{status.value}_{status.name}": status.value for status in HTTPStatus})
@@ -109,18 +111,22 @@ class VariableAssignmentSummary:
     """Track whether a local name is safe to treat as one specific response model."""
 
     known_models: set[str] = field(default_factory=set)
+    known_sources: set[ModelValueSource] = field(default_factory=set)
     has_unknown_assignment: bool = False
 
-    def add_known(self, model: str) -> None:
+    def add_known(self, model: str, source: ModelValueSource) -> None:
         self.known_models.add(model)
+        self.known_sources.add(source)
 
     def add_unknown(self) -> None:
         self.has_unknown_assignment = True
 
-    def single_known_model(self) -> str | None:
+    def single_known_model(self) -> tuple[str, ModelValueSource] | None:
         if self.has_unknown_assignment or len(self.known_models) != 1:
             return None
-        return next(iter(self.known_models))
+        model = next(iter(self.known_models))
+        source: ModelValueSource = "constructor" if self.known_sources == {"constructor"} else "model_validate"
+        return model, source
 
 
 def dotted_name(node: ast.AST) -> str | None:
@@ -249,6 +255,12 @@ def model_name_from_model_validate_call(node: ast.AST) -> str | None:
     return None
 
 
+def model_value_from_model_validate_call(node: ast.AST) -> tuple[str, ModelValueSource] | None:
+    if model_name := model_name_from_model_validate_call(node):
+        return model_name, "model_validate"
+    return None
+
+
 def model_name_from_constructor_call(node: ast.AST) -> str | None:
     if not isinstance(node, ast.Call):
         return None
@@ -257,6 +269,12 @@ def model_name_from_constructor_call(node: ast.AST) -> str | None:
     return None
 
 
+def model_value_from_constructor_call(node: ast.AST) -> tuple[str, ModelValueSource] | None:
+    if model_name := model_name_from_constructor_call(node):
+        return model_name, "constructor"
+    return None
+
+
 def model_name_from_model_dump(node: ast.AST) -> str | None:
     if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute) or node.func.attr != "model_dump":
         return None
@@ -272,6 +290,10 @@ def model_name_from_model_value(node: ast.AST) -> str | None:
     return model_name_from_model_validate_call(node) or model_name_from_constructor_call(node)
 
 
+def model_value_from_model_value(node: ast.AST) -> tuple[str, ModelValueSource] | None:
+    return model_value_from_model_validate_call(node) or model_value_from_constructor_call(node)
+
+
 def model_name_from_dump_response(node: ast.AST) -> str | None:
     if not isinstance(node, ast.Call):
         return None
@@ -287,7 +309,7 @@ def model_name_from_dump_response(node: ast.AST) -> str | None:
 
 
 def actual_kind_from_expr(
-    expr: ast.AST | None, variable_models: dict[str, str] | None = None
+    expr: ast.AST | None, variable_models: dict[str, tuple[str, ModelValueSource]] | None = None
 ) -> tuple[ActualKind, str | None]:
     if expr is None:
         return "none", None
@@ -299,10 +321,14 @@ def actual_kind_from_expr(
     if isinstance(expr, ast.Call) and isinstance(expr.func, ast.Attribute) and expr.func.attr == "model_dump":
         dumped_value = expr.func.value
         if isinstance(dumped_value, ast.Name) and variable_models:
-            # A variable dump can match today, but it bypasses dump_response and
-            # is easier to drift; keep it visible as refactorable.
-            model_name = variable_models.get(dumped_value.id)
-            if model_name:
+            model_assignment = variable_models.get(dumped_value.id)
+            if model_assignment:
+                model_name, source = model_assignment
+                if source == "constructor":
+                    return "model", model_name
+                # A variable dump from model_validate can match today, but it
+                # bypasses dump_response and is easier to drift; keep it visible
+                # as refactorable.
                 return "model_dump_variable", model_name
 
     model_dump_model = model_name_from_model_dump(expr)
@@ -325,7 +351,9 @@ def actual_kind_from_expr(
     return "unknown", None
 
 
-def actual_response_from_return(return_node: ast.Return, variable_models: dict[str, str]) -> ActualResponse:
+def actual_response_from_return(
+    return_node: ast.Return, variable_models: dict[str, tuple[str, ModelValueSource]]
+) -> ActualResponse:
     status: int | None = 200
     body_expr = return_node.value
 
@@ -363,18 +391,21 @@ def target_names(target: ast.AST) -> Iterable[str]:
 
 
 def record_assignment(
-    assignments: defaultdict[str, VariableAssignmentSummary], targets: Iterable[str], model_name: str | None
+    assignments: defaultdict[str, VariableAssignmentSummary],
+    targets: Iterable[str],
+    model_assignment: tuple[str, ModelValueSource] | None,
 ) -> None:
     for target in targets:
-        if model_name is None:
+        if model_assignment is None:
             # Once a name receives an unknown value, later model_dump() calls on it
             # are no longer a reliable signal for the returned schema.
             assignments[target].add_unknown()
         else:
-            assignments[target].add_known(model_name)
+            model_name, source = model_assignment
+            assignments[target].add_known(model_name, source)
 
 
-def variable_model_assignments_for_method(method: MethodNode) -> dict[str, str]:
+def variable_model_assignments_for_method(method: MethodNode) -> dict[str, tuple[str, ModelValueSource]]:
     """Infer local variables that are unambiguously assigned one response model."""
 
     assignments: defaultdict[str, VariableAssignmentSummary] = defaultdict(VariableAssignmentSummary)
@@ -385,10 +416,10 @@ def variable_model_assignments_for_method(method: MethodNode) -> dict[str, str]:
                 record_assignment(
                     assignments,
                     (name for target in targets for name in target_names(target)),
-                    model_name_from_model_value(value),
+                    model_value_from_model_value(value),
                 )
             case ast.AnnAssign(target=target, value=value) if value is not None:
-                record_assignment(assignments, target_names(target), model_name_from_model_value(value))
+                record_assignment(assignments, target_names(target), model_value_from_model_value(value))
             case ast.AugAssign(target=target) | ast.For(target=target) | ast.AsyncFor(target=target):
                 # Mutation and loop targets overwrite prior values with runtime-dependent data.
                 record_assignment(assignments, target_names(target), None)
@@ -399,9 +430,13 @@ def variable_model_assignments_for_method(method: MethodNode) -> dict[str, str]:
             case ast.ExceptHandler(name=name) if name:
                 assignments[name].add_unknown()
             case ast.NamedExpr(target=target, value=value):
-                record_assignment(assignments, target_names(target), model_name_from_model_value(value))
+                record_assignment(assignments, target_names(target), model_value_from_model_value(value))
 
-    return {name: model for name, summary in assignments.items() if (model := summary.single_known_model()) is not None}
+    return {
+        name: assignment
+        for name, summary in assignments.items()
+        if (assignment := summary.single_known_model()) is not None
+    }
 
 
 def actual_responses_for_method(method: MethodNode) -> list[ActualResponse]:
@@ -545,13 +580,52 @@ def iter_controller_files(paths: Iterable[Path]) -> Iterable[Path]:
             yield from sorted(child for child in path.rglob("*.py") if child.is_file())
 
 
+def node_start_lineno(node: ast.ClassDef | MethodNode) -> int:
+    decorator_lines = [decorator.lineno for decorator in node.decorator_list]
+    if decorator_lines:
+        return min(decorator_lines)
+    return node.lineno
+
+
+def line_has_ignore_marker(line: str) -> bool:
+    _, marker, comment = line.partition("#")
+    if not marker:
+        return False
+    normalized = comment.lower()
+    return any(ignore_marker in normalized for ignore_marker in IGNORE_COMMENT_MARKERS)
+
+
+def node_has_ignore_comment(lines: Sequence[str], node: ast.ClassDef | MethodNode) -> bool:
+    start = node_start_lineno(node)
+    end = node.end_lineno or node.lineno
+    if any(line_has_ignore_marker(line) for line in lines[start - 1 : end]):
+        return True
+
+    line_index = start - 2
+    while line_index >= 0:
+        stripped = lines[line_index].strip()
+        if not stripped:
+            line_index -= 1
+            continue
+        if not stripped.startswith("#"):
+            break
+        if line_has_ignore_marker(lines[line_index]):
+            return True
+        line_index -= 1
+    return False
+
+
 def checks_for_file(file_path: Path, repo_root: Path) -> list[ContractCheck]:
-    module = ast.parse(file_path.read_text(encoding="utf-8"), filename=str(file_path))
+    source = file_path.read_text(encoding="utf-8")
+    lines = source.splitlines()
+    module = ast.parse(source, filename=str(file_path))
     checks: list[ContractCheck] = []
 
     for node in module.body:
         if not isinstance(node, ast.ClassDef):
             continue
+        if node_has_ignore_comment(lines, node):
+            continue
 
         class_routes = routes_from_decorators(node.decorator_list)
         class_documented = response_docs_from_decorators(node.decorator_list)
@@ -559,6 +633,8 @@ def checks_for_file(file_path: Path, repo_root: Path) -> list[ContractCheck]:
         for item in node.body:
             if not isinstance(item, ast.FunctionDef | ast.AsyncFunctionDef) or item.name not in HTTP_METHODS:
                 continue
+            if node_has_ignore_comment(lines, item):
+                continue
 
             routes = routes_from_decorators(item.decorator_list) or class_routes
             if not routes:

api/fields/base.py

@@ -7,7 +7,8 @@ class ResponseModel(BaseModel):
     model_config = ConfigDict(
         from_attributes=True,
         extra="ignore",
-        populate_by_name=True,
+        validate_by_name=True,
+        validate_by_alias=True,
         serialize_by_alias=True,
         protected_namespaces=(),
     )

api/libs/oauth_bearer.py

@@ -236,6 +236,16 @@ class TokenExpiredError(Exception):
     """Hard-expire bookkeeping is the resolver's job before raising."""
 
 
+class NegativeCache(StrEnum):
+    """Negative cache markers. ``EXPIRED`` is distinct from ``INVALID`` so a
+    retry inside ``NEGATIVE_TTL`` still reports expiry instead of collapsing
+    into a generic unknown-token miss.
+    """
+
+    INVALID = "invalid"
+    EXPIRED = "expired"
+
+
 # ============================================================================
 # Registry
 # ============================================================================
@@ -343,13 +353,15 @@ class OAuthAccessTokenResolver:
     def _cache_key(self, token_hash: str) -> str:
         return TOKEN_CACHE_KEY_FMT.format(hash=token_hash)
 
-    def cache_get(self, token_hash: str) -> ResolvedRow | None | Literal["invalid"]:
+    def cache_get(self, token_hash: str) -> ResolvedRow | None | NegativeCache:
         raw = self._redis.get(self._cache_key(token_hash))
         if raw is None:
             return None
         text = raw.decode() if isinstance(raw, (bytes, bytearray)) else raw
-        if text == "invalid":
-            return "invalid"
+        try:
+            return NegativeCache(text)
+        except ValueError:
+            pass
         try:
             return ResolvedRow.from_cache(json.loads(text))
         except (ValueError, KeyError):
@@ -363,8 +375,8 @@ class OAuthAccessTokenResolver:
             json.dumps(row.to_cache()),
         )
 
-    def cache_set_negative(self, token_hash: str) -> None:
-        self._redis.setex(self._cache_key(token_hash), self._negative_ttl, "invalid")
+    def cache_set_negative(self, token_hash: str, marker: NegativeCache = NegativeCache.INVALID) -> None:
+        self._redis.setex(self._cache_key(token_hash), self._negative_ttl, str(marker))
 
     def hard_expire(self, session: Session, row_id: uuid.UUID | str, token_hash: str) -> None:
         """Atomic CAS — only the worker that flips revoked_at emits audit;
@@ -385,7 +397,7 @@ class OAuthAccessTokenResolver:
                 extra={"audit": True, "token_id": str(row_id)},
             )
         self._redis.delete(self._cache_key(token_hash))
-        self.cache_set_negative(token_hash)
+        self.cache_set_negative(token_hash, NegativeCache.EXPIRED)
 
 
 class _VariantResolver:
@@ -395,9 +407,11 @@ class _VariantResolver:
 
     def resolve(self, token_hash: str) -> ResolvedRow | None:
         cached = self._parent.cache_get(token_hash)
-        if cached == "invalid":
+        if isinstance(cached, NegativeCache):
+            if cached is NegativeCache.EXPIRED:
+                raise TokenExpiredError("token_expired")
             return None
-        if cached is not None and not isinstance(cached, str):
+        if cached is not None:
             if not self._matches_variant(cached):
                 return None
             return cached
@@ -413,7 +427,7 @@ class _VariantResolver:
         now = datetime.now(UTC)
         if row.expires_at is not None and row.expires_at <= now:
             self._parent.hard_expire(session, row.id, token_hash)
-            return None
+            raise TokenExpiredError("token_expired")
 
         if not self._matches_variant_model(row):
             logger.error(
@@ -472,7 +486,7 @@ def record_layer0_verdict(token_hash: str, tenant_id: str, verdict: bool) -> Non
     if raw is None:
         return
     text = raw.decode() if isinstance(raw, (bytes, bytearray)) else raw
-    if text == "invalid":
+    if text in (NegativeCache.INVALID, NegativeCache.EXPIRED):
         return
     try:
         data = json.loads(text)
@@ -601,6 +615,8 @@ def validate_bearer(*, accept: frozenset[Accepts]) -> Callable[[Callable[_DP, _D
 
             try:
                 ctx = get_authenticator().authenticate(token)
+            except TokenExpiredError:
+                raise Unauthorized("token_expired")
             except InvalidBearerError as e:
                 raise Unauthorized(str(e))
 

api/libs/pyrefly_diagnostics.py

@@ -2,9 +2,10 @@
 
 from __future__ import annotations
 
+import argparse
 import sys
 
-_DIAGNOSTIC_PREFIXES = ("ERROR ", "WARNING ")
+_DIAGNOSTIC_PREFIXES = ("ERROR ", "WARN ", "WARNING ")
 _LOCATION_PREFIX = "-->"
 
 
@@ -13,7 +14,7 @@ def extract_diagnostics(raw_output: str) -> str:
 
     The full pyrefly output includes code excerpts and carets, which create noisy
     diffs. This helper keeps only:
-    - diagnostic headline lines (``ERROR ...`` / ``WARNING ...``)
+    - diagnostic headline lines (``ERROR ...`` / ``WARN ...`` / ``WARNING ...``)
     - the following location line (``--> path:line:column``), when present
     """
 
@@ -36,11 +37,28 @@ def extract_diagnostics(raw_output: str) -> str:
     return "\n".join(diagnostics) + "\n"
 
 
+def render_diagnostics(raw_output: str, exit_code: int) -> str:
+    """Render concise diagnostics and fall back to raw output on unmatched failures."""
+
+    diagnostics = extract_diagnostics(raw_output)
+    if diagnostics:
+        return diagnostics
+
+    if exit_code != 0:
+        return raw_output
+
+    return ""
+
+
 def main() -> int:
     """Read pyrefly output from stdin and print normalized diagnostics."""
 
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--status", type=int, default=0)
+    args = parser.parse_args()
+
     raw_output = sys.stdin.read()
-    sys.stdout.write(extract_diagnostics(raw_output))
+    sys.stdout.write(render_diagnostics(raw_output, exit_code=args.status))
     return 0
 
 

api/openapi/markdown/console-openapi.md

@@ -2868,6 +2868,7 @@ Delete an existing tracing configuration for an application
 | ---- | ----------- |
 | 204 | Tracing configuration deleted successfully |
 | 400 | Invalid request parameters or configuration not found |
+| 403 | Insufficient permissions |
 
 ### [GET] /apps/{app_id}/trace-config
 Get tracing configuration for an application
@@ -2909,6 +2910,7 @@ Update an existing tracing configuration for an application
 | ---- | ----------- | ------ |
 | 200 | Tracing configuration updated successfully | **application/json**: [TraceAppConfigResponse](#traceappconfigresponse)<br> |
 | 400 | Invalid request parameters or configuration not found |  |
+| 403 | Insufficient permissions |  |
 
 ### [POST] /apps/{app_id}/trace-config
 **Create a new trace app configuration**
@@ -2933,6 +2935,7 @@ Create a new tracing configuration for an application
 | ---- | ----------- | ------ |
 | 201 | Tracing configuration created successfully | **application/json**: [TraceAppConfigResponse](#traceappconfigresponse)<br> |
 | 400 | Invalid request parameters or configuration already exists |  |
+| 403 | Insufficient permissions |  |
 
 ### [POST] /apps/{app_id}/trigger-enable
 **Update app trigger (enable/disable)**
@@ -13429,7 +13432,6 @@ Soft lifecycle state for Agent records.
 | created_at | integer |  | No |
 | files | [ string ] |  | Yes |
 | id | string |  | Yes |
-| message_chain_id | string |  | No |
 | message_id | string |  | Yes |
 | observation | string |  | No |
 | position | integer |  | Yes |
@@ -14540,8 +14542,8 @@ Enum class for configurate method of provider model.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | annotation_create_account | [SimpleAccount](#simpleaccount) |  | No |
+| annotation_id | string |  | Yes |
 | created_at | integer |  | No |
-| id | string |  | Yes |
 
 #### ConversationDetail
 
@@ -16710,6 +16712,7 @@ Input field definition for snippet parameters.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
+| description | string |  | No |
 | icon | string |  | No |
 | icon_background | string |  | No |
 | icon_type | string |  | No |
@@ -17079,6 +17082,7 @@ Enum class for large language model mode.
 | agent_thoughts | [ [AgentThought](#agentthought) ] |  | No |
 | annotation | [ConversationAnnotation](#conversationannotation) |  | No |
 | annotation_hit_history | [ConversationAnnotationHitHistory](#conversationannotationhithistory) |  | No |
+| answer | string |  | Yes |
 | answer_tokens | integer |  | No |
 | conversation_id | string |  | Yes |
 | created_at | integer |  | No |
@@ -17092,12 +17096,11 @@ Enum class for large language model mode.
 | inputs | object |  | Yes |
 | message | [JSONValue](#jsonvalue) |  | No |
 | message_files | [ [MessageFile](#messagefile) ] |  | No |
-| message_metadata_dict | [JSONValue](#jsonvalue) |  | No |
 | message_tokens | integer |  | No |
+| metadata | [JSONValue](#jsonvalue) |  | No |
 | parent_message_id | string |  | No |
 | provider_response_latency | number |  | No |
 | query | string |  | Yes |
-| re_sign_file_url_answer | string |  | Yes |
 | status | string |  | Yes |
 | workflow_run_id | string |  | No |
 

api/openapi/markdown/openapi-openapi.md

@@ -990,6 +990,12 @@ Pagination for GET /account/sessions. Strict (extra='forbid').
 | last_used_at | string |  | No |
 | prefix | string |  | Yes |
 
+#### SimpleResultResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| result | string |  | Yes |
+
 #### SupportedAppType
 
 App types the ``app`` usage face (``get app``) lists and filters.

api/providers/trace/trace-tencent/tests/unit_tests/tencent_trace/test_tencent_trace.py

@@ -181,30 +181,34 @@ class TestTencentDataTrace:
         mock_trace_utils.convert_to_trace_id.return_value = 123
         mock_trace_utils.create_link.return_value = "link"
 
-        with patch.object(tencent_data_trace, "_get_user_id", return_value="user-1"):
-            with patch.object(tencent_data_trace, "_process_workflow_nodes") as mock_proc:
-                with patch.object(tencent_data_trace, "_record_workflow_trace_duration") as mock_dur:
-                    mock_span_builder.build_workflow_spans.return_value = [MagicMock(), MagicMock()]
+        with (
+            patch.object(tencent_data_trace, "_get_user_id", return_value="user-1"),
+            patch.object(tencent_data_trace, "_process_workflow_nodes") as mock_proc,
+            patch.object(tencent_data_trace, "_record_workflow_trace_duration") as mock_dur,
+        ):
+            mock_span_builder.build_workflow_spans.return_value = [MagicMock(), MagicMock()]
 
-                    tencent_data_trace.workflow_trace(trace_info)
+            tencent_data_trace.workflow_trace(trace_info)
 
-                    mock_trace_utils.convert_to_trace_id.assert_called_once_with("run-id")
-                    mock_trace_utils.create_link.assert_called_once_with("parent-trace-id")
-                    mock_span_builder.build_workflow_spans.assert_called_once()
-                    assert tencent_data_trace.trace_client.add_span.call_count == 2
-                    mock_proc.assert_called_once_with(trace_info, 123)
-                    mock_dur.assert_called_once_with(trace_info)
+            mock_trace_utils.convert_to_trace_id.assert_called_once_with("run-id")
+            mock_trace_utils.create_link.assert_called_once_with("parent-trace-id")
+            mock_span_builder.build_workflow_spans.assert_called_once()
+            assert tencent_data_trace.trace_client.add_span.call_count == 2
+            mock_proc.assert_called_once_with(trace_info, 123)
+            mock_dur.assert_called_once_with(trace_info)
 
-    def test_workflow_trace_exception(self, tencent_data_trace):
+    def test_workflow_trace_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
         trace_info.workflow_run_id = "run-id"
 
-        with patch(
-            "dify_trace_tencent.tencent_trace.TencentTraceUtils.convert_to_trace_id", side_effect=Exception("error")
+        with (
+            patch(
+                "dify_trace_tencent.tencent_trace.TencentTraceUtils.convert_to_trace_id", side_effect=Exception("error")
+            ),
+            caplog.at_level(logging.ERROR),
         ):
-            with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
-                tencent_data_trace.workflow_trace(trace_info)
-                mock_log.assert_called_once_with("[Tencent APM] Failed to process workflow trace")
+            tencent_data_trace.workflow_trace(trace_info)
+        assert "[Tencent APM] Failed to process workflow trace" in caplog.text
 
     def test_message_trace(self, tencent_data_trace, mock_trace_utils, mock_span_builder):
         trace_info = MagicMock(spec=MessageTraceInfo)
@@ -214,29 +218,33 @@ class TestTencentDataTrace:
         mock_trace_utils.convert_to_trace_id.return_value = 123
         mock_trace_utils.create_link.return_value = "link"
 
-        with patch.object(tencent_data_trace, "_get_user_id", return_value="user-1"):
-            with patch.object(tencent_data_trace, "_record_message_llm_metrics") as mock_metrics:
-                with patch.object(tencent_data_trace, "_record_message_trace_duration") as mock_dur:
-                    mock_span_builder.build_message_span.return_value = MagicMock()
+        with (
+            patch.object(tencent_data_trace, "_get_user_id", return_value="user-1"),
+            patch.object(tencent_data_trace, "_record_message_llm_metrics") as mock_metrics,
+            patch.object(tencent_data_trace, "_record_message_trace_duration") as mock_dur,
+        ):
+            mock_span_builder.build_message_span.return_value = MagicMock()
 
-                    tencent_data_trace.message_trace(trace_info)
+            tencent_data_trace.message_trace(trace_info)
 
-                    mock_trace_utils.convert_to_trace_id.assert_called_once_with("msg-id")
-                    mock_trace_utils.create_link.assert_called_once_with("parent-trace-id")
-                    mock_span_builder.build_message_span.assert_called_once()
-                    tencent_data_trace.trace_client.add_span.assert_called_once()
-                    mock_metrics.assert_called_once_with(trace_info)
-                    mock_dur.assert_called_once_with(trace_info)
+            mock_trace_utils.convert_to_trace_id.assert_called_once_with("msg-id")
+            mock_trace_utils.create_link.assert_called_once_with("parent-trace-id")
+            mock_span_builder.build_message_span.assert_called_once()
+            tencent_data_trace.trace_client.add_span.assert_called_once()
+            mock_metrics.assert_called_once_with(trace_info)
+            mock_dur.assert_called_once_with(trace_info)
 
-    def test_message_trace_exception(self, tencent_data_trace):
+    def test_message_trace_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=MessageTraceInfo)
 
-        with patch(
-            "dify_trace_tencent.tencent_trace.TencentTraceUtils.convert_to_trace_id", side_effect=Exception("error")
+        with (
+            patch(
+                "dify_trace_tencent.tencent_trace.TencentTraceUtils.convert_to_trace_id", side_effect=Exception("error")
+            ),
+            caplog.at_level(logging.ERROR),
         ):
-            with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
-                tencent_data_trace.message_trace(trace_info)
-                mock_log.assert_called_once_with("[Tencent APM] Failed to process message trace")
+            tencent_data_trace.message_trace(trace_info)
+        assert "[Tencent APM] Failed to process message trace" in caplog.text
 
     def test_tool_trace(self, tencent_data_trace, mock_trace_utils, mock_span_builder):
         trace_info = MagicMock(spec=ToolTraceInfo)
@@ -259,16 +267,18 @@ class TestTencentDataTrace:
         tencent_data_trace.tool_trace(trace_info)
         tencent_data_trace.trace_client.add_span.assert_not_called()
 
-    def test_tool_trace_exception(self, tencent_data_trace):
+    def test_tool_trace_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=ToolTraceInfo)
         trace_info.message_id = "msg-id"
 
-        with patch(
-            "dify_trace_tencent.tencent_trace.TencentTraceUtils.convert_to_span_id", side_effect=Exception("error")
+        with (
+            patch(
+                "dify_trace_tencent.tencent_trace.TencentTraceUtils.convert_to_span_id", side_effect=Exception("error")
+            ),
+            caplog.at_level(logging.ERROR),
         ):
-            with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
-                tencent_data_trace.tool_trace(trace_info)
-                mock_log.assert_called_once_with("[Tencent APM] Failed to process tool trace")
+            tencent_data_trace.tool_trace(trace_info)
+        assert "[Tencent APM] Failed to process tool trace" in caplog.text
 
     def test_dataset_retrieval_trace(self, tencent_data_trace, mock_trace_utils, mock_span_builder):
         trace_info = MagicMock(spec=DatasetRetrievalTraceInfo)
@@ -291,29 +301,34 @@ class TestTencentDataTrace:
         tencent_data_trace.dataset_retrieval_trace(trace_info)
         tencent_data_trace.trace_client.add_span.assert_not_called()
 
-    def test_dataset_retrieval_trace_exception(self, tencent_data_trace):
+    def test_dataset_retrieval_trace_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=DatasetRetrievalTraceInfo)
         trace_info.message_id = "msg-id"
 
-        with patch(
-            "dify_trace_tencent.tencent_trace.TencentTraceUtils.convert_to_span_id", side_effect=Exception("error")
+        with (
+            patch(
+                "dify_trace_tencent.tencent_trace.TencentTraceUtils.convert_to_span_id", side_effect=Exception("error")
+            ),
+            caplog.at_level(logging.ERROR),
         ):
-            with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
-                tencent_data_trace.dataset_retrieval_trace(trace_info)
-                mock_log.assert_called_once_with("[Tencent APM] Failed to process dataset retrieval trace")
+            tencent_data_trace.dataset_retrieval_trace(trace_info)
+        assert "[Tencent APM] Failed to process dataset retrieval trace" in caplog.text
 
-    def test_suggested_question_trace(self, tencent_data_trace):
+    def test_suggested_question_trace(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=SuggestedQuestionTraceInfo)
-        with patch("dify_trace_tencent.tencent_trace.logger.info") as mock_log:
+        with caplog.at_level(logging.INFO):
             tencent_data_trace.suggested_question_trace(trace_info)
-            mock_log.assert_called_once_with("[Tencent APM] Processing suggested question trace")
+        assert "[Tencent APM] Processing suggested question trace" in caplog.text
 
-    def test_suggested_question_trace_exception(self, tencent_data_trace):
+    def test_suggested_question_trace_exception(
+        self, tencent_data_trace, monkeypatch, caplog: pytest.LogCaptureFixture
+    ):
         trace_info = MagicMock(spec=SuggestedQuestionTraceInfo)
-        with patch("dify_trace_tencent.tencent_trace.logger.info", side_effect=Exception("error")):
-            with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
-                tencent_data_trace.suggested_question_trace(trace_info)
-                mock_log.assert_called_once_with("[Tencent APM] Failed to process suggested question trace")
+        target_logger = logging.getLogger("dify_trace_tencent.tencent_trace")
+        monkeypatch.setattr(target_logger, "info", MagicMock(side_effect=Exception("error")))
+        with caplog.at_level(logging.ERROR):
+            tencent_data_trace.suggested_question_trace(trace_info)
+        assert "[Tencent APM] Failed to process suggested question trace" in caplog.text
 
     def test_process_workflow_nodes(self, tencent_data_trace, mock_trace_utils):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
@@ -327,35 +342,42 @@ class TestTencentDataTrace:
         node2.id = "n2"
         node2.node_type = BuiltinNodeTypes.TOOL
 
-        with patch.object(tencent_data_trace, "_get_workflow_node_executions", return_value=[node1, node2]):
-            with patch.object(tencent_data_trace, "_build_workflow_node_span", side_effect=["span1", "span2"]):
-                with patch.object(tencent_data_trace, "_record_llm_metrics") as mock_metrics:
-                    tencent_data_trace._process_workflow_nodes(trace_info, 123)
+        with (
+            patch.object(tencent_data_trace, "_get_workflow_node_executions", return_value=[node1, node2]),
+            patch.object(tencent_data_trace, "_build_workflow_node_span", side_effect=["span1", "span2"]),
+            patch.object(tencent_data_trace, "_record_llm_metrics") as mock_metrics,
+        ):
+            tencent_data_trace._process_workflow_nodes(trace_info, 123)
 
-                    assert tencent_data_trace.trace_client.add_span.call_count == 2
-                    mock_metrics.assert_called_once_with(node1)
+            assert tencent_data_trace.trace_client.add_span.call_count == 2
+            mock_metrics.assert_called_once_with(node1)
 
-    def test_process_workflow_nodes_node_exception(self, tencent_data_trace, mock_trace_utils):
+    def test_process_workflow_nodes_node_exception(
+        self, tencent_data_trace, mock_trace_utils, caplog: pytest.LogCaptureFixture
+    ):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
         mock_trace_utils.convert_to_span_id.return_value = 111
 
         node = MagicMock(spec=WorkflowNodeExecution)
         node.id = "n1"
 
-        with patch.object(tencent_data_trace, "_get_workflow_node_executions", return_value=[node]):
-            with patch.object(tencent_data_trace, "_build_workflow_node_span", side_effect=Exception("node error")):
-                with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
-                    tencent_data_trace._process_workflow_nodes(trace_info, 123)
-                    # The exception should be caught by the outer handler since convert_to_span_id is called first
-                    mock_log.assert_called_once_with("[Tencent APM] Failed to process workflow nodes")
+        with (
+            patch.object(tencent_data_trace, "_get_workflow_node_executions", return_value=[node]),
+            patch.object(tencent_data_trace, "_build_workflow_node_span", side_effect=Exception("node error")),
+            caplog.at_level(logging.ERROR),
+        ):
+            tencent_data_trace._process_workflow_nodes(trace_info, 123)
+        assert "[Tencent APM] Failed to process workflow nodes" in caplog.text
 
-    def test_process_workflow_nodes_exception(self, tencent_data_trace, mock_trace_utils):
+    def test_process_workflow_nodes_exception(
+        self, tencent_data_trace, mock_trace_utils, caplog: pytest.LogCaptureFixture
+    ):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
         mock_trace_utils.convert_to_span_id.side_effect = Exception("outer error")
 
-        with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
+        with caplog.at_level(logging.ERROR):
             tencent_data_trace._process_workflow_nodes(trace_info, 123)
-            mock_log.assert_called_once_with("[Tencent APM] Failed to process workflow nodes")
+        assert "[Tencent APM] Failed to process workflow nodes" in caplog.text
 
     def test_build_workflow_node_span(self, tencent_data_trace, mock_span_builder):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
@@ -377,16 +399,18 @@ class TestTencentDataTrace:
             assert result == "span"
             builder_method.assert_called_once_with(123, 456, trace_info, node)
 
-    def test_build_workflow_node_span_exception(self, tencent_data_trace, mock_span_builder):
+    def test_build_workflow_node_span_exception(
+        self, tencent_data_trace, mock_span_builder, caplog: pytest.LogCaptureFixture
+    ):
         node = MagicMock(spec=WorkflowNodeExecution)
         node.node_type = BuiltinNodeTypes.LLM
         node.id = "n1"
         mock_span_builder.build_workflow_llm_span.side_effect = Exception("error")
 
-        with patch("dify_trace_tencent.tencent_trace.logger.debug") as mock_log:
+        with caplog.at_level(logging.DEBUG):
             result = tencent_data_trace._build_workflow_node_span(node, 123, MagicMock(), 456)
-            assert result is None
-            mock_log.assert_called_once()
+        assert result is None
+        assert len([r for r in caplog.records if r.levelno == logging.DEBUG]) >= 1
 
     def test_get_workflow_node_executions(self, tencent_data_trace):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
@@ -419,16 +443,16 @@ class TestTencentDataTrace:
                     assert results == mock_executions
                     account.set_tenant_id.assert_called_once_with("tenant-1")
 
-    def test_get_workflow_node_executions_no_app_id(self, tencent_data_trace):
+    def test_get_workflow_node_executions_no_app_id(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
         trace_info.metadata = {}
 
-        with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
+        with caplog.at_level(logging.ERROR):
             results = tencent_data_trace._get_workflow_node_executions(trace_info)
-            assert results == []
-            mock_log.assert_called_once()
+        assert results == []
+        assert len([r for r in caplog.records if r.levelno == logging.ERROR]) >= 1
 
-    def test_get_workflow_node_executions_app_not_found(self, tencent_data_trace):
+    def test_get_workflow_node_executions_app_not_found(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
         trace_info.metadata = {"app_id": "app-1"}
 
@@ -439,23 +463,25 @@ class TestTencentDataTrace:
                 session = mock_session_ctx.return_value.__enter__.return_value
                 session.scalar.return_value = None
 
-                with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
+                with caplog.at_level(logging.ERROR):
                     results = tencent_data_trace._get_workflow_node_executions(trace_info)
-                    assert results == []
-                    mock_log.assert_called_once()
+                assert results == []
+                assert len([r for r in caplog.records if r.levelno == logging.ERROR]) >= 1
 
     def test_get_user_id_workflow(self, tencent_data_trace):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
         trace_info.tenant_id = "tenant-1"
         trace_info.metadata = {"user_id": "user-1"}
 
-        with patch("dify_trace_tencent.tencent_trace.sessionmaker", side_effect=Exception("Database error")):
-            with patch("dify_trace_tencent.tencent_trace.db") as mock_db:
-                mock_db.init_app = MagicMock()
-                mock_db.engine = MagicMock()
+        with (
+            patch("dify_trace_tencent.tencent_trace.sessionmaker", side_effect=Exception("Database error")),
+            patch("dify_trace_tencent.tencent_trace.db") as mock_db,
+        ):
+            mock_db.init_app = MagicMock()
+            mock_db.engine = MagicMock()
 
-                user_id = tencent_data_trace._get_user_id(trace_info)
-                assert user_id == "unknown"
+            user_id = tencent_data_trace._get_user_id(trace_info)
+            assert user_id == "unknown"
 
     def test_get_user_id_only_user_id(self, tencent_data_trace):
         trace_info = MagicMock(spec=MessageTraceInfo)
@@ -471,16 +497,18 @@ class TestTencentDataTrace:
         user_id = tencent_data_trace._get_user_id(trace_info)
         assert user_id == "anonymous"
 
-    def test_get_user_id_exception(self, tencent_data_trace):
+    def test_get_user_id_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
         trace_info.tenant_id = "t"
         trace_info.metadata = {"user_id": "u"}
 
-        with patch("dify_trace_tencent.tencent_trace.sessionmaker", side_effect=Exception("error")):
-            with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
-                user_id = tencent_data_trace._get_user_id(trace_info)
-                assert user_id == "unknown"
-                mock_log.assert_called_once_with("[Tencent APM] Failed to get user ID")
+        with (
+            patch("dify_trace_tencent.tencent_trace.sessionmaker", side_effect=Exception("error")),
+            caplog.at_level(logging.ERROR),
+        ):
+            user_id = tencent_data_trace._get_user_id(trace_info)
+        assert user_id == "unknown"
+        assert "[Tencent APM] Failed to get user ID" in caplog.text
 
     def test_record_llm_metrics_usage_in_process_data(self, tencent_data_trace):
         node = MagicMock(spec=WorkflowNodeExecution)
@@ -514,14 +542,14 @@ class TestTencentDataTrace:
         tencent_data_trace.trace_client.record_llm_duration.assert_called_once()
         tencent_data_trace.trace_client.record_token_usage.assert_called_once()
 
-    def test_record_llm_metrics_exception(self, tencent_data_trace):
+    def test_record_llm_metrics_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         node = MagicMock(spec=WorkflowNodeExecution)
         node.process_data = None
         node.outputs = None
 
-        with patch("dify_trace_tencent.tencent_trace.logger.debug") as mock_log:
+        with caplog.at_level(logging.DEBUG):
             tencent_data_trace._record_llm_metrics(node)
-            # Should not crash
+        # Should not crash
 
     def test_record_message_llm_metrics(self, tencent_data_trace):
         trace_info = MagicMock(spec=MessageTraceInfo)
@@ -553,13 +581,13 @@ class TestTencentDataTrace:
         tencent_data_trace._record_message_llm_metrics(trace_info)
         tencent_data_trace.trace_client.record_llm_duration.assert_called_once()
 
-    def test_record_message_llm_metrics_exception(self, tencent_data_trace):
+    def test_record_message_llm_metrics_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=MessageTraceInfo)
         trace_info.metadata = None
 
-        with patch("dify_trace_tencent.tencent_trace.logger.debug") as mock_log:
+        with caplog.at_level(logging.DEBUG):
             tencent_data_trace._record_message_llm_metrics(trace_info)
-            # Should not crash
+        # Should not crash
 
     def test_record_workflow_trace_duration(self, tencent_data_trace):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
@@ -605,11 +633,11 @@ class TestTencentDataTrace:
             attributes = kwargs["attributes"] if "attributes" in kwargs else args[1] if len(args) > 1 else {}
             assert attributes["has_conversation"] == "false"
 
-    def test_record_workflow_trace_duration_exception(self, tencent_data_trace):
+    def test_record_workflow_trace_duration_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=WorkflowTraceInfo)
         trace_info.start_time = MagicMock()  # This might cause total_seconds() to fail if not mocked right
 
-        with patch("dify_trace_tencent.tencent_trace.logger.debug") as mock_log:
+        with caplog.at_level(logging.DEBUG):
             tencent_data_trace._record_workflow_trace_duration(trace_info)
 
     def test_record_message_trace_duration(self, tencent_data_trace):
@@ -627,11 +655,11 @@ class TestTencentDataTrace:
             2.0, {"conversation_mode": "chat", "stream": "true"}
         )
 
-    def test_record_message_trace_duration_exception(self, tencent_data_trace):
+    def test_record_message_trace_duration_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         trace_info = MagicMock(spec=MessageTraceInfo)
         trace_info.start_time = None
 
-        with patch("dify_trace_tencent.tencent_trace.logger.debug") as mock_log:
+        with caplog.at_level(logging.DEBUG):
             tencent_data_trace._record_message_trace_duration(trace_info)
 
     def test_close(self, tencent_data_trace):
@@ -647,11 +675,11 @@ class TestTencentDataTrace:
 
         client.shutdown.assert_called_once()
 
-    def test_close_exception(self, tencent_data_trace):
+    def test_close_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
         tencent_data_trace.trace_client.shutdown.side_effect = Exception("error")
-        with patch("dify_trace_tencent.tencent_trace.logger.exception") as mock_log:
+        with caplog.at_level(logging.ERROR):
             tencent_data_trace.close()
-            mock_log.assert_called_once_with("[Tencent APM] Failed to shutdown trace client during cleanup")
+        assert "[Tencent APM] Failed to shutdown trace client during cleanup" in caplog.text
 
     def test_close_handles_async_shutdown_mock(self, tencent_data_trace):
         shutdown = AsyncMock()

api/providers/vdb/vdb-lindorm/src/dify_vdb_lindorm/lindorm_vector.py

@@ -113,7 +113,7 @@ class LindormVectorStore(BaseVector):
         )
         def _bulk_with_retry(actions):
             try:
-                response = self._client.bulk(actions, timeout=timeout)
+                response = self._client.bulk(body=actions, timeout=timeout)
                 if response["errors"]:
                     error_items = [item for item in response["items"] if "error" in item["index"]]
                     error_msg = f"Bulk indexing had {len(error_items)} errors"
@@ -231,7 +231,7 @@ class LindormVectorStore(BaseVector):
             routing_filter_query = {
                 "query": {"bool": {"must": [{"term": {f"{ROUTING_FIELD}.keyword": self._routing}}]}}
             }
-            self._client.delete_by_query(self._collection_name, body=routing_filter_query)
+            self._client.delete_by_query(index=self._collection_name, body=routing_filter_query)
             self.refresh()
         else:
             if self._client.indices.exists(index=self._collection_name):

api/providers/vdb/vdb-lindorm/tests/unit_tests/test_lindorm_vector.py

@@ -127,7 +127,7 @@ def test_create_refresh_and_add_texts_success(lindorm_module, monkeypatch: pytes
     vector.add_texts(docs, embeddings, batch_size=2, timeout=9)
 
     assert vector._client.bulk.call_count == 2
-    actions = vector._client.bulk.call_args_list[0].args[0]
+    actions = vector._client.bulk.call_args_list[0].kwargs["body"]
     assert actions[0]["index"]["routing"] == "route"
     assert actions[1][lindorm_module.ROUTING_FIELD] == "route"
     vector.refresh()

api/providers/vdb/vdb-weaviate/tests/unit_tests/test_weaviate_vector.py

@@ -268,9 +268,11 @@ class TestWeaviateVector(unittest.TestCase):
         wv._client = MagicMock()
         wv._client.collections.exists.side_effect = RuntimeError("create failed")
 
-        with patch.object(weaviate_vector_module.logger, "exception") as mock_exception:
-            with pytest.raises(RuntimeError, match="create failed"):
-                wv._create_collection()
+        with (
+            patch.object(weaviate_vector_module.logger, "exception") as mock_exception,
+            pytest.raises(RuntimeError, match="create failed"),
+        ):
+            wv._create_collection()
 
         mock_exception.assert_called_once()
 
@@ -835,9 +837,11 @@ class TestWeaviateVector(unittest.TestCase):
         wv._client.collections.use.return_value = mock_col
         mock_col.data.delete_by_id.side_effect = FakeUnexpectedStatusCodeError(500)
 
-        with patch.object(weaviate_vector_module, "UnexpectedStatusCodeError", FakeUnexpectedStatusCodeError):
-            with pytest.raises(FakeUnexpectedStatusCodeError, match="status=500"):
-                wv.delete_by_ids(["bad-id"])
+        with (
+            patch.object(weaviate_vector_module, "UnexpectedStatusCodeError", FakeUnexpectedStatusCodeError),
+            pytest.raises(FakeUnexpectedStatusCodeError, match="status=500"),
+        ):
+            wv.delete_by_ids(["bad-id"])
 
     def test_json_serializable_converts_datetime(self):
         wv = WeaviateVector.__new__(WeaviateVector)

api/pyproject.toml

@@ -1,11 +1,11 @@
 [project]
 name = "dify-api"
-version = "1.14.2"
+version = "1.15.0"
 requires-python = "~=3.12.0"
 
 dependencies = [
     # Legacy: mature and widely deployed
-    "bleach>=6.3.0,<7.0.0",
+    "bleach>=6.4.0,<7.0.0",
     "boto3>=1.43.24,<2.0.0",
     "celery>=5.6.3,<6.0.0",
     "croniter>=6.2.2,<7.0.0",

api/services/account_service.py

@@ -1788,6 +1788,9 @@ class TenantService:
                 account_id,
             )
 
+        if dify_config.RBAC_ENABLED:
+            RBACService.MemberRoles.delete_rbac_bindings(tenant_id=tenant.id, account_id=account_id)
+
     @staticmethod
     def update_member_role(
         tenant: Tenant, member: Account, new_role: str, operator: Account, *, session: scoped_session | Session

api/services/enterprise/rbac_service.py

@@ -309,7 +309,8 @@ _LEGACY_WORKSPACE_OWNER_KEYS: list[str] = [
     "customization.manage",
     "plugin.install",
     "plugin.plugin_preferences",
-    "plugin.manage",
+    "plugin.model_config",
+    "plugin.delete",
     "plugin.debug",
     "credential.use",
     "credential.create",
@@ -330,8 +331,6 @@ _LEGACY_WORKSPACE_OWNER_KEYS: list[str] = [
     "snippets.management",
     "tool.manage",
     "mcp.manage",
-    "snippets.create_and_modify",
-    "snippets.management",
 ]
 
 _LEGACY_WORKSPACE_ADMIN_KEYS: list[str] = [
@@ -342,7 +341,8 @@ _LEGACY_WORKSPACE_ADMIN_KEYS: list[str] = [
     "customization.manage",
     "plugin.install",
     "plugin.plugin_preferences",
-    "plugin.manage",
+    "plugin.model_config",
+    "plugin.delete",
     "plugin.debug",
     "credential.use",
     "credential.create",
@@ -361,8 +361,6 @@ _LEGACY_WORKSPACE_ADMIN_KEYS: list[str] = [
     "snippets.management",
     "tool.manage",
     "mcp.manage",
-    "snippets.create_and_modify",
-    "snippets.management",
 ]
 
 _LEGACY_WORKSPACE_EDITOR_KEYS: list[str] = [
@@ -378,7 +376,9 @@ _LEGACY_WORKSPACE_EDITOR_KEYS: list[str] = [
     "dataset.external.connect",
     "snippets.create_and_modify",
     "tool.manage",
-    "snippets.create_and_modify",
+    "billing.view",
+    "billing.subscription.manage",
+    "billing.manage",
 ]
 
 _LEGACY_WORKSPACE_NORMAL_KEYS: list[str] = [
@@ -386,6 +386,9 @@ _LEGACY_WORKSPACE_NORMAL_KEYS: list[str] = [
     "plugin.install",
     "credential.use",
     "app_library.access",
+    "billing.view",
+    "billing.subscription.manage",
+    "billing.manage",
 ]
 
 _LEGACY_WORKSPACE_DATASET_OPERATOR_KEYS: list[str] = [
@@ -404,6 +407,8 @@ _LEGACY_APP_OWNER_KEYS: list[str] = [
     "app.acl.release_and_version",
     "app.acl.monitor",
     "app.acl.access_config",
+    "app.acl.tracing_config",
+    "app.acl.log_and_annotation",
 ]
 
 _LEGACY_APP_ADMIN_KEYS: list[str] = [
@@ -416,6 +421,9 @@ _LEGACY_APP_ADMIN_KEYS: list[str] = [
     "app.acl.release_and_version",
     "app.acl.monitor",
     "app.acl.access_config",
+    "app.acl.access_config",
+    "app.acl.tracing_config",
+    "app.acl.log_and_annotation",
 ]
 
 _LEGACY_APP_EDITOR_KEYS: list[str] = [
@@ -431,9 +439,6 @@ _LEGACY_APP_EDITOR_KEYS: list[str] = [
 ]
 
 _LEGACY_APP_NORMAL_KEYS: list[str] = [
-    "app.acl.preview",
-    "app.acl.view_layout",
-    "app.acl.test_and_run",
     "app.acl.monitor",
 ]
 
@@ -834,6 +839,7 @@ class RBACService:
             options: ListOption | None = None,
         ) -> Paginated[RBACRole]:
             params = (options or ListOption()).to_params({"include_owner": include_owner})
+            params["dataset_operator_enabled"] = dify_config.DATASET_OPERATOR_ENABLED
             data = _inner_call(
                 "GET",
                 f"{_INNER_PREFIX}/roles",
@@ -1678,6 +1684,17 @@ class RBACService:
             )
             return MemberRolesResponse.model_validate(data or {})
 
+        @staticmethod
+        def delete_rbac_bindings(tenant_id: str, account_id: str):
+            data = _inner_call(
+                "DELETE",
+                f"{_INNER_PREFIX}/members/rbac-bindings",
+                tenant_id=tenant_id,
+                account_id=account_id,
+                params={"account_id": account_id},
+            )
+            return data
+
     class CheckAccess:
         """Call the ``/inner/api/rbac/check-access`` endpoint."""
 

api/tasks/app_generate/workflow_execute_task.py

@@ -19,11 +19,16 @@ from core.app.entities.app_invoke_entities import (
     InvokeFrom,
     WorkflowAppGenerateEntity,
 )
+from core.app.entities.task_entities import WorkflowFinishStreamResponse, WorkflowStartStreamResponse
 from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig, WorkflowResumptionContext
 from core.repositories import DifyCoreRepositoryFactory
 from extensions.ext_database import db
+from graphon.entities import WorkflowStartReason
+from graphon.enums import WorkflowExecutionStatus
 from graphon.runtime import GraphRuntimeState
+from libs.datetime_utils import naive_utc_now
 from libs.flask_utils import set_login_user
+from libs.helper import to_timestamp
 from models.account import Account
 from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom
 from models.model import App, AppMode, Conversation, EndUser, Message
@@ -173,14 +178,24 @@ class _AppRunner:
                 )
             except Exception as exc:
                 if exec_params.streaming:
-                    _publish_error_event(exc, exec_params.workflow_run_id, exec_params.app_mode)
+                    _publish_failed_workflow_terminal_events(
+                        exc=exc,
+                        exec_params=exec_params,
+                    )
                 raise
 
             if not exec_params.streaming:
                 return response
 
             assert isinstance(response, Generator)
-            _publish_streaming_response(response, exec_params.workflow_run_id, exec_params.app_mode)
+            _publish_streaming_response(
+                response,
+                exec_params.workflow_run_id,
+                exec_params.app_mode,
+                exec_params.workflow_id,
+                exec_params.args.get("inputs", {}),
+                WorkflowStartReason.INITIAL,
+            )
 
     def _run_app(
         self,
@@ -246,29 +261,197 @@ def _resolve_user_for_run(session: Session, workflow_run: WorkflowRun) -> Accoun
     return session.get(EndUser, workflow_run.created_by)
 
 
-def _publish_error_event(exc: Exception, workflow_run_id: str, app_mode: AppMode) -> None:
-    topic = MessageBasedAppGenerator.get_response_topic(app_mode, workflow_run_id)
-    payload = json.dumps({"event": "error", "message": str(exc), "status": 500})
-    topic.publish(payload.encode())
+def _publish_failed_workflow_terminal_events(exc: Exception, exec_params: AppExecutionParams) -> None:
+    """Publish synthetic workflow lifecycle events for pre-runtime failures.
+
+    Early failures can happen before the app generator creates a task entity or
+    emits any workflow queue events. In that window SSE consumers still need a
+    normal terminal event to close their state machines, so we synthesize a
+    minimal `workflow_started -> workflow_finished(failed)` sequence here.
+
+    `workflow_run_id` is reused as a synthetic `task_id` because no application
+    task id exists yet on this failure path.
+    """
+    timestamp = to_timestamp(naive_utc_now())
+    assert timestamp is not None
+
+    topic = MessageBasedAppGenerator.get_response_topic(exec_params.app_mode, exec_params.workflow_run_id)
+    started_payload = WorkflowStartStreamResponse(
+        task_id=exec_params.workflow_run_id,
+        workflow_run_id=exec_params.workflow_run_id,
+        data=WorkflowStartStreamResponse.Data(
+            id=exec_params.workflow_run_id,
+            workflow_id=exec_params.workflow_id,
+            inputs=exec_params.args.get("inputs", {}),
+            created_at=timestamp,
+            reason=WorkflowStartReason.INITIAL,
+        ),
+    )
+    topic.publish(json.dumps(started_payload.model_dump(mode="json"), ensure_ascii=False).encode())
+
+    finished_payload = WorkflowFinishStreamResponse(
+        task_id=exec_params.workflow_run_id,
+        workflow_run_id=exec_params.workflow_run_id,
+        data=WorkflowFinishStreamResponse.Data(
+            id=exec_params.workflow_run_id,
+            workflow_id=exec_params.workflow_id,
+            status=WorkflowExecutionStatus.FAILED,
+            outputs=None,
+            error=str(exc),
+            elapsed_time=0.0,
+            total_tokens=0,
+            total_steps=0,
+            created_by={},
+            created_at=timestamp,
+            finished_at=timestamp,
+            exceptions_count=1,
+            files=[],
+        ),
+    )
+    topic.publish(json.dumps(finished_payload.model_dump(mode="json"), ensure_ascii=False).encode())
+
+
+def _get_event_name(event: str | Mapping[str, Any] | BaseModel) -> str | None:
+    if isinstance(event, BaseModel):
+        # Temporary compatibility for legacy BaseModel stream events; remove after confirming generators always emit
+        # str / Mapping responses.
+        event_name = getattr(event, "event", None)
+    elif isinstance(event, Mapping):
+        event_name = event.get("event")
+    else:
+        return None
+
+    if event_name is None:
+        return None
+    return str(event_name)
+
+
+def _get_task_id(event: str | Mapping[str, Any] | BaseModel) -> str | None:
+    if isinstance(event, BaseModel):
+        # Temporary compatibility for legacy BaseModel stream events; remove after confirming generators always emit
+        # str / Mapping responses.
+        task_id = getattr(event, "task_id", None)
+    elif isinstance(event, Mapping):
+        task_id = event.get("task_id")
+    else:
+        return None
+
+    return task_id if isinstance(task_id, str) and task_id else None
 
 
 def _publish_streaming_response(
     response_stream: Generator[str | Mapping[str, Any] | BaseModel, None, None],
-    workflow_run_id: str,
+    workflow_run_id: str | uuid.UUID,
     app_mode: AppMode,
+    workflow_id: str,
+    inputs: Mapping[str, Any],
+    started_reason: WorkflowStartReason,
 ) -> None:
-    topic = MessageBasedAppGenerator.get_response_topic(app_mode, workflow_run_id)
-    for event in response_stream:
-        try:
-            if isinstance(event, BaseModel):
-                payload = json.dumps(event.model_dump(mode="json"), ensure_ascii=False)
-            else:
-                payload = json.dumps(event, ensure_ascii=False, default=str)
-        except (TypeError, ValueError):
-            logger.exception("error while encoding event")
-            continue
+    """Publish workflow stream events and close broken streams with a failed terminal event.
 
-        topic.publish(payload.encode())
+    `_AppRunner.run()` only handles failures before the generator is returned.
+    Once we start iterating the runtime stream, this helper becomes the last
+    place that can guarantee SSE consumers eventually see a terminal workflow
+    lifecycle event.
+    """
+    normalized_workflow_run_id = str(workflow_run_id)
+
+    def _publish_failed_terminal_event(error_message: str, task_id: str, publish_started: bool) -> None:
+        timestamp = to_timestamp(naive_utc_now())
+        assert timestamp is not None
+
+        if publish_started:
+            started_payload = WorkflowStartStreamResponse(
+                task_id=task_id,
+                workflow_run_id=normalized_workflow_run_id,
+                data=WorkflowStartStreamResponse.Data(
+                    id=normalized_workflow_run_id,
+                    workflow_id=workflow_id,
+                    inputs=inputs,
+                    created_at=timestamp,
+                    reason=started_reason,
+                ),
+            )
+            topic.publish(
+                json.dumps(
+                    started_payload.model_dump(mode="json", fallback=str),
+                    ensure_ascii=False,
+                ).encode()
+            )
+
+        finished_payload = WorkflowFinishStreamResponse(
+            task_id=task_id,
+            workflow_run_id=normalized_workflow_run_id,
+            data=WorkflowFinishStreamResponse.Data(
+                id=normalized_workflow_run_id,
+                workflow_id=workflow_id,
+                status=WorkflowExecutionStatus.FAILED,
+                outputs=None,
+                error=error_message,
+                elapsed_time=0.0,
+                total_tokens=0,
+                total_steps=0,
+                created_by={},
+                created_at=timestamp,
+                finished_at=timestamp,
+                exceptions_count=1,
+                files=[],
+            ),
+        )
+        topic.publish(json.dumps(finished_payload.model_dump(mode="json"), ensure_ascii=False).encode())
+
+    terminal_events = {"workflow_finished", "workflow_paused"}
+    unexpected_stream_end_message = "Workflow stream ended without a terminal event"
+    topic = MessageBasedAppGenerator.get_response_topic(app_mode, normalized_workflow_run_id)
+    started_published = False
+    terminal_published = False
+    last_task_id = normalized_workflow_run_id
+
+    try:
+        for event in response_stream:
+            event_name = _get_event_name(event)
+            task_id = _get_task_id(event)
+            if task_id is not None:
+                last_task_id = task_id
+
+            try:
+                if isinstance(event, BaseModel):
+                    payload = json.dumps(event.model_dump(mode="json"), ensure_ascii=False)
+                else:
+                    payload = json.dumps(event, ensure_ascii=False, default=str)
+            except (TypeError, ValueError):
+                logger.exception("error while encoding event")
+                continue
+
+            topic.publish(payload.encode())
+
+            if event_name == "workflow_started":
+                started_published = True
+            elif event_name in terminal_events:
+                terminal_published = True
+    except Exception as exc:
+        if not terminal_published:
+            logger.exception(
+                "Workflow stream for run %s failed before terminal event; publishing fallback terminal event",
+                normalized_workflow_run_id,
+            )
+            _publish_failed_terminal_event(
+                error_message=str(exc) or exc.__class__.__name__,
+                task_id=last_task_id,
+                publish_started=not started_published,
+            )
+        raise
+
+    if not terminal_published:
+        logger.warning(
+            "Workflow stream for run %s ended without a terminal event; publishing fallback terminal event",
+            normalized_workflow_run_id,
+        )
+        _publish_failed_terminal_event(
+            error_message=unexpected_stream_end_message,
+            task_id=last_task_id,
+            publish_started=not started_published,
+        )
 
 
 @shared_task(queue=WORKFLOW_BASED_APP_EXECUTION_QUEUE)
@@ -454,7 +637,14 @@ def _resume_advanced_chat(
         raise
 
     assert isinstance(response, Generator)
-    _publish_streaming_response(response, workflow_run_id, AppMode.ADVANCED_CHAT)
+    _publish_streaming_response(
+        response,
+        workflow_run_id,
+        AppMode.ADVANCED_CHAT,
+        workflow.id,
+        generate_entity.inputs,
+        WorkflowStartReason.RESUMPTION,
+    )
 
 
 def _resume_workflow(
@@ -509,7 +699,14 @@ def _resume_workflow(
         raise
 
     assert isinstance(response, Generator)
-    _publish_streaming_response(response, workflow_run_id, AppMode.WORKFLOW)
+    _publish_streaming_response(
+        response,
+        workflow_run_id,
+        AppMode.WORKFLOW,
+        workflow.id,
+        generate_entity.inputs,
+        WorkflowStartReason.RESUMPTION,
+    )
 
     try:
         workflow_run_repo.delete_workflow_pause(pause_entity)

api/tests/integration_tests/controllers/openapi/conftest.py

@@ -72,6 +72,7 @@ def mint_token(flask_app: Flask):
         prefix: str,
         subject_email: str,
         subject_issuer: str | None,
+        expires_at: datetime | None = None,
     ) -> OAuthAccessToken:
         with flask_app.app_context():
             row = OAuthAccessToken(
@@ -82,7 +83,7 @@ def mint_token(flask_app: Flask):
                 subject_issuer=subject_issuer,
                 client_id="difyctl",
                 device_label="test-device",
-                expires_at=datetime.now(UTC) + timedelta(hours=1),
+                expires_at=expires_at or (datetime.now(UTC) + timedelta(hours=1)),
             )
             db.session.add(row)
             db.session.commit()
@@ -111,6 +112,21 @@ def account_token(workspace_account, mint_token) -> str:
     return token
 
 
+@pytest.fixture
+def expired_account_token(workspace_account, mint_token) -> str:
+    account, _, _ = workspace_account
+    token = "dfoa_" + uuid.uuid4().hex
+    mint_token(
+        token,
+        account_id=account.id,
+        prefix="dfoa_",
+        subject_email=account.email,
+        subject_issuer="dify:account",
+        expires_at=datetime.now(UTC) - timedelta(minutes=1),
+    )
+    return token
+
+
 @pytest.fixture(autouse=True)
 def _flush_auth_redis(flask_app: Flask) -> Generator[None, None, None]:
     def _flush():

api/tests/integration_tests/controllers/openapi/test_auth.py

@@ -6,6 +6,7 @@ acceptance/rejection on app-scoped routes.
 
 from __future__ import annotations
 
+import uuid
 from collections.abc import Generator
 
 import pytest
@@ -16,6 +17,50 @@ from extensions.ext_database import db
 from models import App, Tenant
 
 
+def test_expired_token_returns_401_token_expired(
+    test_client: FlaskClient,
+    expired_account_token: str,
+) -> None:
+    """An expired bearer is distinguishable from an unknown one: 401 with the
+    domain code ``token_expired`` (+ actionable hint), not a generic 401 or 500."""
+    res = test_client.get(
+        "/openapi/v1/account",
+        headers={"Authorization": f"Bearer {expired_account_token}"},
+    )
+    assert res.status_code == 401
+    assert res.json["code"] == "token_expired"
+    assert res.json["hint"]
+
+
+def test_expired_token_replay_stays_token_expired(
+    test_client: FlaskClient,
+    expired_account_token: str,
+) -> None:
+    """The distinct ``expired`` negative-cache marker keeps the second hit (served
+    from cache, inside NEGATIVE_TTL) reporting ``token_expired`` rather than
+    collapsing into a generic unknown-token 401."""
+    headers = {"Authorization": f"Bearer {expired_account_token}"}
+    first = test_client.get("/openapi/v1/account", headers=headers)
+    second = test_client.get("/openapi/v1/account", headers=headers)
+    assert first.json["code"] == "token_expired"
+    assert second.status_code == 401
+    assert second.json["code"] == "token_expired"
+
+
+def test_unknown_token_returns_401_unauthorized_not_500(
+    test_client: FlaskClient,
+    workspace_account,
+) -> None:
+    """An unknown bearer is a clean 401 ``unauthorized`` — not the latent 500 the
+    pipeline used to leak for unmapped InvalidBearerError."""
+    res = test_client.get(
+        "/openapi/v1/account",
+        headers={"Authorization": "Bearer dfoa_" + uuid.uuid4().hex},
+    )
+    assert res.status_code == 401
+    assert res.json["code"] == "unauthorized"
+
+
 def test_info_accepts_account_bearer_with_apps_read_scope(
     test_client: FlaskClient,
     app_in_workspace: App,

api/tests/test_containers_integration_tests/.ruff.toml

@@ -27,7 +27,6 @@ extend-select = ["ANN401", "ARG", "TID251"]
 "controllers/web/test_wraps.py" = ["ARG"]
 "core/app/layers/test_pause_state_persist_layer.py" = ["ARG"]
 "core/rag/retrieval/test_dataset_retrieval_integration.py" = ["ARG"]
-"models/test_account.py" = ["ARG"]
 "models/test_conversation_message_inputs.py" = ["ARG"]
 "models/test_conversation_status_count.py" = ["ARG"]
 "repositories/test_sqlalchemy_api_workflow_run_repository.py" = ["ARG"]

api/tests/test_containers_integration_tests/models/test_account.py

@@ -3,6 +3,7 @@ Integration tests for Account and Tenant model methods that interact with the da
 
 Migrated from unit_tests/models/test_account_models.py, replacing
 @patch("models.account.db") mock patches with real PostgreSQL operations.
+Also absorbs unit_tests/models/test_account.py role helper coverage.
 
 Covers:
 - Account.current_tenant setter (sets _current_tenant and role from TenantAccountJoin)
@@ -12,6 +13,7 @@ Covers:
 """
 
 from collections.abc import Generator
+from typing import cast
 from uuid import uuid4
 
 import pytest
@@ -20,8 +22,10 @@ from sqlalchemy.orm import Session
 
 from models.account import Account, AccountIntegrate, Tenant, TenantAccountJoin, TenantAccountRole
 
+TrackedRow = Account | AccountIntegrate | Tenant | TenantAccountJoin
 
-def _cleanup_tracked_rows(db_session: Session, tracked: list) -> None:
+
+def _cleanup_tracked_rows(db_session: Session, tracked: list[TrackedRow]) -> None:
     """Delete rows tracked during the test so committed state does not leak into the DB.
 
     Rolls back any pending (uncommitted) session state first, then issues DELETE
@@ -52,7 +56,7 @@ def _build_account(email_prefix: str = "account") -> Account:
 class _DBTrackingTestBase:
     """Base class providing a tracker list and shared row factories for account/tenant tests."""
 
-    _tracked: list
+    _tracked: list[TrackedRow]
 
     @pytest.fixture(autouse=True)
     def _setup_cleanup(self, db_session_with_containers: Session) -> Generator[None, None, None]:
@@ -84,6 +88,22 @@ class _DBTrackingTestBase:
         return join
 
 
+class TestTenantAccountRole:
+    """Tests for TenantAccountRole helper methods."""
+
+    def test_account_is_privileged_role(self) -> None:
+        assert TenantAccountRole.ADMIN == "admin"
+        assert TenantAccountRole.OWNER == "owner"
+        assert TenantAccountRole.EDITOR == "editor"
+        assert TenantAccountRole.NORMAL == "normal"
+
+        assert TenantAccountRole.is_privileged_role(TenantAccountRole.ADMIN)
+        assert TenantAccountRole.is_privileged_role(TenantAccountRole.OWNER)
+        assert not TenantAccountRole.is_privileged_role(TenantAccountRole.NORMAL)
+        assert not TenantAccountRole.is_privileged_role(TenantAccountRole.EDITOR)
+        assert not TenantAccountRole.is_privileged_role(cast(TenantAccountRole, ""))
+
+
 class TestAccountCurrentTenantSetter(_DBTrackingTestBase):
     """Integration tests for Account.current_tenant property setter."""
 
@@ -176,7 +196,7 @@ class TestAccountGetByOpenId(_DBTrackingTestBase):
         assert result is not None
         assert result.id == account.id
 
-    def test_get_by_openid_returns_none_when_no_integrate_exists(self, db_session_with_containers: Session) -> None:
+    def test_get_by_openid_returns_none_when_no_integrate_exists(self) -> None:
         """get_by_openid returns None when no AccountIntegrate row matches."""
         result = Account.get_by_openid("github", f"github_{uuid4()}")
 

api/tests/test_containers_integration_tests/pyrefly.toml

@@ -50,7 +50,6 @@ project-excludes = [
   "libs/broadcast_channel/redis/test_streams_channel.py",
   "libs/test_auto_renew_redis_lock_integration.py",
   "libs/test_rate_limiter_integration.py",
-  "models/test_account.py",
   "models/test_conversation_message_inputs.py",
   "models/test_types_enum_text.py",
   "repositories/test_sqlalchemy_api_workflow_node_execution_repository.py",

api/tests/unit_tests/commands/test_lint_response_contracts.py

@@ -77,6 +77,25 @@ class AnnotationApi(Resource):
     assert "prefer dump_response" in checks[0].reason
 
 
+def test_constructor_variable_model_dump_is_valid(tmp_path: Path):
+    checks = _checks_for_source(
+        tmp_path,
+        """
+@ns.route("/annotations")
+class AnnotationApi(Resource):
+    @ns.response(201, "Created", ns.models[AnnotationResponse.__name__])
+    def post(self):
+        response = AnnotationResponse(id="new", name=name)
+        return response.model_dump(mode="json"), 201
+""",
+    )
+
+    assert len(checks) == 1
+    assert checks[0].classification == "valid"
+    assert checks[0].actual[0].kind == "model"
+    assert checks[0].actual[0].model == "AnnotationResponse"
+
+
 def test_variable_model_dump_with_wrong_documented_schema_is_mismatch(tmp_path: Path):
     checks = _checks_for_source(
         tmp_path,
@@ -117,6 +136,38 @@ class StreamApi(Resource):
     assert {actual.model for actual in checks[0].actual} == {"StreamResponse"}
 
 
+def test_response_contract_ignore_comment_skips_route_method(tmp_path: Path):
+    checks = _checks_for_source(
+        tmp_path,
+        """
+@ns.route("/binary")
+class BinaryApi(Resource):
+    # response-contract:ignore binary response
+    @ns.response(200, "Binary file")
+    def get(self):
+        return send_file(path)
+
+
+# response-contract:ignore compact Flask response
+@ns.route("/compact")
+class CompactApi(Resource):
+    def get(self):
+        return make_response({"url": "https://example.com"})
+
+
+@ns.route("/regular")
+class RegularApi(Resource):
+    @ns.response(200, "OK", ns.models[RegularResponse.__name__])
+    def get(self):
+        return dump_response(RegularResponse, {})
+""",
+    )
+
+    assert len(checks) == 1
+    assert checks[0].class_name == "RegularApi"
+    assert checks[0].classification == "valid"
+
+
 def test_main_is_report_only_by_default_for_mismatches(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
     module = _load_lint_response_contracts_module()
     controller_path = tmp_path / "controllers" / "sample.py"

api/tests/unit_tests/controllers/common/test_fields.py

@@ -1,5 +1,4 @@
 import builtins
-from types import SimpleNamespace
 from unittest.mock import patch
 
 from flask.views import MethodView as FlaskMethodView
@@ -22,7 +21,7 @@ def test_parameters_model_round_trip():
 
 
 def test_site_icon_url_uses_signed_url_for_image_icon():
-    site = SimpleNamespace(
+    site = Site(
         title="Example",
         chat_color_theme=None,
         chat_color_theme_inverted=False,
@@ -46,7 +45,7 @@ def test_site_icon_url_uses_signed_url_for_image_icon():
 
 
 def test_site_icon_url_is_none_for_non_image_icon():
-    site = SimpleNamespace(
+    site = Site(
         title="Example",
         chat_color_theme=None,
         chat_color_theme_inverted=False,

api/tests/unit_tests/controllers/console/app/test_ops_trace_api.py

@@ -0,0 +1,154 @@
+from __future__ import annotations
+
+from contextlib import nullcontext
+from types import SimpleNamespace
+from unittest.mock import MagicMock, PropertyMock, patch
+
+import pytest
+from flask import Flask
+from werkzeug.exceptions import Forbidden
+
+from controllers.common import wraps as common_wraps
+from controllers.console import console_ns
+from controllers.console import wraps as console_wraps
+from controllers.console.app import ops_trace as ops_trace_module
+from controllers.console.app import wraps as app_wraps
+from libs import login as login_lib
+from models.account import Account, AccountStatus, TenantAccountRole
+
+
+def _make_account(role: TenantAccountRole) -> Account:
+    account = Account(name="tester", email="tester@example.com")
+    account.id = "account-123"  # type: ignore[assignment]
+    account.status = AccountStatus.ACTIVE
+    account.role = role
+    account._current_tenant = SimpleNamespace(id="tenant-123")  # type: ignore[assignment]
+    account._get_current_object = lambda: account  # type: ignore[attr-defined]
+    return account
+
+
+def _make_app() -> SimpleNamespace:
+    return SimpleNamespace(id="app-123", tenant_id="tenant-123", status="normal", mode="chat")
+
+
+def _patch_console_guards(
+    monkeypatch: pytest.MonkeyPatch,
+    account: Account,
+    app_model: SimpleNamespace,
+    *,
+    rbac_enabled: bool = False,
+) -> None:
+    monkeypatch.setattr(login_lib.dify_config, "LOGIN_DISABLED", True)
+    monkeypatch.setattr(login_lib.dify_config, "RBAC_ENABLED", rbac_enabled)
+    monkeypatch.setattr(console_wraps.dify_config, "EDITION", "CLOUD")
+    monkeypatch.setattr(login_lib, "current_user", account)
+    monkeypatch.setattr(login_lib, "current_account_with_tenant", lambda: (account, account.current_tenant_id))
+    monkeypatch.setattr(console_wraps, "current_account_with_tenant", lambda: (account, account.current_tenant_id))
+    monkeypatch.setattr(common_wraps, "current_account_with_tenant", lambda: (account, account.current_tenant_id))
+    monkeypatch.setattr(app_wraps, "_load_app_model_from_scoped_session", lambda _app_id: app_model)
+
+
+def _patch_payload(payload: dict[str, object] | None):
+    if payload is None:
+        return nullcontext()
+    return patch.object(type(console_ns), "payload", new_callable=PropertyMock, return_value=payload)
+
+
+@pytest.mark.parametrize(
+    ("method_name", "path", "payload", "service_method_name", "service_result"),
+    [
+        (
+            "post",
+            "/console/api/apps/app-123/trace-config",
+            {"tracing_provider": "mlflow", "tracing_config": {"endpoint": "https://trace.example.com"}},
+            "create_tracing_app_config",
+            {"id": "trace-config-1"},
+        ),
+        (
+            "patch",
+            "/console/api/apps/app-123/trace-config",
+            {"tracing_provider": "mlflow", "tracing_config": {"endpoint": "https://trace.example.com"}},
+            "update_tracing_app_config",
+            True,
+        ),
+        (
+            "delete",
+            "/console/api/apps/app-123/trace-config?tracing_provider=mlflow",
+            None,
+            "delete_tracing_app_config",
+            True,
+        ),
+    ],
+)
+def test_trace_config_mutations_require_edit_permission(
+    app: Flask,
+    monkeypatch: pytest.MonkeyPatch,
+    method_name: str,
+    path: str,
+    payload: dict[str, object] | None,
+    service_method_name: str,
+    service_result: object,
+) -> None:
+    app.config.setdefault("RESTX_MASK_HEADER", "X-Fields")
+    account = _make_account(TenantAccountRole.NORMAL)
+    _patch_console_guards(monkeypatch, account, _make_app())
+    service_mock = MagicMock(return_value=service_result)
+    monkeypatch.setattr(ops_trace_module.OpsService, service_method_name, service_mock)
+
+    with app.test_request_context(path, method=method_name.upper(), json=payload):
+        with _patch_payload(payload):
+            with pytest.raises(Forbidden):
+                getattr(ops_trace_module.TraceAppConfigApi(), method_name)(app_id="app-123")
+
+    service_mock.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("method_name", "path", "payload", "service_method_name", "service_result"),
+    [
+        (
+            "post",
+            "/console/api/apps/app-123/trace-config",
+            {"tracing_provider": "mlflow", "tracing_config": {"endpoint": "https://trace.example.com"}},
+            "create_tracing_app_config",
+            {"id": "trace-config-1"},
+        ),
+        (
+            "patch",
+            "/console/api/apps/app-123/trace-config",
+            {"tracing_provider": "mlflow", "tracing_config": {"endpoint": "https://trace.example.com"}},
+            "update_tracing_app_config",
+            True,
+        ),
+        (
+            "delete",
+            "/console/api/apps/app-123/trace-config?tracing_provider=mlflow",
+            None,
+            "delete_tracing_app_config",
+            True,
+        ),
+    ],
+)
+def test_trace_config_mutations_require_rbac_permission(
+    app: Flask,
+    monkeypatch: pytest.MonkeyPatch,
+    method_name: str,
+    path: str,
+    payload: dict[str, object] | None,
+    service_method_name: str,
+    service_result: object,
+) -> None:
+    app.config.setdefault("RESTX_MASK_HEADER", "X-Fields")
+    account = _make_account(TenantAccountRole.NORMAL)
+    _patch_console_guards(monkeypatch, account, _make_app(), rbac_enabled=True)
+    monkeypatch.setattr(common_wraps.db, "session", SimpleNamespace(scalar=lambda _stmt: "other-account"))
+    monkeypatch.setattr(common_wraps.RBACService.CheckAccess, "check", MagicMock(return_value=False))
+    service_mock = MagicMock(return_value=service_result)
+    monkeypatch.setattr(ops_trace_module.OpsService, service_method_name, service_mock)
+
+    with app.test_request_context(path, method=method_name.upper(), json=payload):
+        with _patch_payload(payload):
+            with pytest.raises(Forbidden):
+                getattr(ops_trace_module.TraceAppConfigApi(), method_name)(app_id="app-123")
+
+    service_mock.assert_not_called()

api/tests/unit_tests/controllers/console/app/test_workflow_convert_api.py

@@ -2,21 +2,14 @@
 
 from __future__ import annotations
 
+from inspect import unwrap
 from types import SimpleNamespace
 
 import pytest
 from flask import Flask
 
 from controllers.console.app import workflow as workflow_module
-
-
-def _unwrap(func):
-    bound_self = getattr(func, "__self__", None)
-    while hasattr(func, "__wrapped__"):
-        func = func.__wrapped__
-    if bound_self is not None:
-        return func.__get__(bound_self, bound_self.__class__)
-    return func
+from controllers.console.app.workflow import ConvertToWorkflowApi
 
 
 class TestConvertToWorkflowApi:
@@ -25,9 +18,9 @@ class TestConvertToWorkflowApi:
         return workflow_module.ConvertToWorkflowApi()
 
     def test_convert_to_workflow_attaches_permission_keys_when_rbac_enabled(
-        self, api, app: Flask, monkeypatch: pytest.MonkeyPatch
+        self, api: ConvertToWorkflowApi, app: Flask, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        method = _unwrap(api.post)
+        method = unwrap(api.post)
 
         monkeypatch.setattr(
             workflow_module,
@@ -46,6 +39,7 @@ class TestConvertToWorkflowApi:
             json={},
         ):
             response = method(
+                api,
                 current_tenant_id="tenant-1",
                 current_user=SimpleNamespace(id="u1"),
                 app_model=SimpleNamespace(id="app-1"),

api/tests/unit_tests/controllers/console/auth/test_login_logout.py

@@ -9,6 +9,7 @@ This module tests the core authentication endpoints including:
 """
 
 import base64
+import logging
 from unittest.mock import ANY, MagicMock, Mock, patch
 
 import pytest
@@ -191,7 +192,9 @@ class TestLoginApi:
     @patch("controllers.console.auth.login.dify_config.BILLING_ENABLED", False)
     @patch("controllers.console.auth.login.AccountService.is_login_error_rate_limit")
     @patch("controllers.console.auth.login.RegisterService.get_invitation_with_case_fallback")
-    def test_login_fails_when_rate_limited(self, mock_get_invitation, mock_is_rate_limit, mock_db, app: Flask):
+    def test_login_fails_when_rate_limited(
+        self, mock_get_invitation, mock_is_rate_limit, mock_db, app: Flask, caplog: pytest.LogCaptureFixture
+    ):
         """
         Test login rejection when rate limit is exceeded.
 
@@ -204,22 +207,26 @@ class TestLoginApi:
         mock_get_invitation.return_value = None
 
         # Act & Assert
-        with patch("controllers.console.auth.login.logger.warning") as mock_log_warning:
-            with app.test_request_context(
-                "/login", method="POST", json={"email": "test@example.com", "password": encode_password("password")}
-            ):
-                login_api = LoginApi()
-                with pytest.raises(EmailPasswordLoginLimitError):
-                    login_api.post()
+        with app.test_request_context(
+            "/login", method="POST", json={"email": "test@example.com", "password": encode_password("password")}
+        ):
+            login_api = LoginApi()
+            with pytest.raises(EmailPasswordLoginLimitError):
+                login_api.post()
 
-        assert mock_log_warning.call_count == 1
-        assert mock_log_warning.call_args.args[1] == "test@example.com"
-        assert mock_log_warning.call_args.args[2] == LoginFailureReason.LOGIN_RATE_LIMITED
+        warn_records = [
+            r for r in caplog.records if r.name == "controllers.console.auth.login" and r.levelno == logging.WARNING
+        ]
+        assert len(warn_records) == 1
+        assert warn_records[0].args[0] == "test@example.com"
+        assert warn_records[0].args[1] == LoginFailureReason.LOGIN_RATE_LIMITED
 
     @patch("controllers.console.wraps.db")
     @patch("controllers.console.auth.login.dify_config.BILLING_ENABLED", True)
     @patch("controllers.console.auth.login.BillingService.is_email_in_freeze")
-    def test_login_fails_when_account_frozen(self, mock_is_frozen, mock_db, app: Flask):
+    def test_login_fails_when_account_frozen(
+        self, mock_is_frozen, mock_db, app: Flask, caplog: pytest.LogCaptureFixture
+    ):
         """
         Test login rejection for frozen accounts.
 
@@ -231,17 +238,19 @@ class TestLoginApi:
         mock_is_frozen.return_value = True
 
         # Act & Assert
-        with patch("controllers.console.auth.login.logger.warning") as mock_log_warning:
-            with app.test_request_context(
-                "/login", method="POST", json={"email": "frozen@example.com", "password": encode_password("password")}
-            ):
-                login_api = LoginApi()
-                with pytest.raises(AccountInFreezeError):
-                    login_api.post()
+        with app.test_request_context(
+            "/login", method="POST", json={"email": "frozen@example.com", "password": encode_password("password")}
+        ):
+            login_api = LoginApi()
+            with pytest.raises(AccountInFreezeError):
+                login_api.post()
 
-        assert mock_log_warning.call_count == 1
-        assert mock_log_warning.call_args.args[1] == "frozen@example.com"
-        assert mock_log_warning.call_args.args[2] == LoginFailureReason.ACCOUNT_IN_FREEZE
+        warn_records = [
+            r for r in caplog.records if r.name == "controllers.console.auth.login" and r.levelno == logging.WARNING
+        ]
+        assert len(warn_records) == 1
+        assert warn_records[0].args[0] == "frozen@example.com"
+        assert warn_records[0].args[1] == LoginFailureReason.ACCOUNT_IN_FREEZE
 
     @patch("controllers.console.wraps.db")
     @patch("controllers.console.auth.login.dify_config.BILLING_ENABLED", False)
@@ -257,6 +266,7 @@ class TestLoginApi:
         mock_is_rate_limit,
         mock_db,
         app: Flask,
+        caplog: pytest.LogCaptureFixture,
     ):
         """
         Test login failure with invalid credentials.
@@ -272,20 +282,22 @@ class TestLoginApi:
         mock_authenticate.side_effect = AccountPasswordError("Invalid password")
 
         # Act & Assert
-        with patch("controllers.console.auth.login.logger.warning") as mock_log_warning:
-            with app.test_request_context(
-                "/login",
-                method="POST",
-                json={"email": "test@example.com", "password": encode_password("WrongPass123!")},
-            ):
-                login_api = LoginApi()
-                with pytest.raises(AuthenticationFailedError):
-                    login_api.post()
+        with app.test_request_context(
+            "/login",
+            method="POST",
+            json={"email": "test@example.com", "password": encode_password("WrongPass123!")},
+        ):
+            login_api = LoginApi()
+            with pytest.raises(AuthenticationFailedError):
+                login_api.post()
 
         mock_add_rate_limit.assert_called_once_with("test@example.com")
-        assert mock_log_warning.call_count == 1
-        assert mock_log_warning.call_args.args[1] == "test@example.com"
-        assert mock_log_warning.call_args.args[2] == LoginFailureReason.INVALID_CREDENTIALS
+        warn_records = [
+            r for r in caplog.records if r.name == "controllers.console.auth.login" and r.levelno == logging.WARNING
+        ]
+        assert len(warn_records) == 1
+        assert warn_records[0].args[0] == "test@example.com"
+        assert warn_records[0].args[1] == LoginFailureReason.INVALID_CREDENTIALS
 
     @patch("controllers.console.wraps.db")
     @patch("controllers.console.auth.login.dify_config.BILLING_ENABLED", False)
@@ -293,7 +305,7 @@ class TestLoginApi:
     @patch("controllers.console.auth.login.RegisterService.get_invitation_with_case_fallback")
     @patch("controllers.console.auth.login.AccountService.authenticate")
     def test_login_fails_for_banned_account(
-        self, mock_authenticate, mock_get_invitation, mock_is_rate_limit, mock_db, app: Flask
+        self, mock_authenticate, mock_get_invitation, mock_is_rate_limit, mock_db, app: Flask, caplog
     ):
         """
         Test login rejection for banned accounts.
@@ -308,19 +320,21 @@ class TestLoginApi:
         mock_authenticate.side_effect = AccountLoginError("Account is banned")
 
         # Act & Assert
-        with patch("controllers.console.auth.login.logger.warning") as mock_log_warning:
-            with app.test_request_context(
-                "/login",
-                method="POST",
-                json={"email": "banned@example.com", "password": encode_password("ValidPass123!")},
-            ):
-                login_api = LoginApi()
-                with pytest.raises(AccountBannedError):
-                    login_api.post()
+        with app.test_request_context(
+            "/login",
+            method="POST",
+            json={"email": "banned@example.com", "password": encode_password("ValidPass123!")},
+        ):
+            login_api = LoginApi()
+            with pytest.raises(AccountBannedError):
+                login_api.post()
 
-        assert mock_log_warning.call_count == 1
-        assert mock_log_warning.call_args.args[1] == "banned@example.com"
-        assert mock_log_warning.call_args.args[2] == LoginFailureReason.ACCOUNT_BANNED
+        warn_records = [
+            r for r in caplog.records if r.name == "controllers.console.auth.login" and r.levelno == logging.WARNING
+        ]
+        assert len(warn_records) == 1
+        assert warn_records[0].args[0] == "banned@example.com"
+        assert warn_records[0].args[1] == LoginFailureReason.ACCOUNT_BANNED
 
     @patch("controllers.console.wraps.db")
     @patch("controllers.console.auth.login.dify_config.BILLING_ENABLED", False)
@@ -452,23 +466,26 @@ class TestLoginApi:
         mock_get_token_data: MagicMock,
         mock_db: MagicMock,
         app: Flask,
+        caplog: pytest.LogCaptureFixture,
     ):
         mock_get_token_data.return_value = {"email": "User@Example.com", "code": "123456"}
         mock_get_account.side_effect = Unauthorized("Account is banned.")
 
-        with patch("controllers.console.auth.login.logger.warning") as mock_log_warning:
-            with app.test_request_context(
-                "/email-code-login/validity",
-                method="POST",
-                json={"email": "User@Example.com", "code": encode_code("123456"), "token": "token-123"},
-            ):
-                with pytest.raises(AccountBannedError):
-                    EmailCodeLoginApi().post()
+        with app.test_request_context(
+            "/email-code-login/validity",
+            method="POST",
+            json={"email": "User@Example.com", "code": encode_code("123456"), "token": "token-123"},
+        ):
+            with pytest.raises(AccountBannedError):
+                EmailCodeLoginApi().post()
 
         mock_revoke_token.assert_called_once_with("token-123")
-        assert mock_log_warning.call_count == 1
-        assert mock_log_warning.call_args.args[1] == "user@example.com"
-        assert mock_log_warning.call_args.args[2] == LoginFailureReason.ACCOUNT_BANNED
+        warn_records = [
+            r for r in caplog.records if r.name == "controllers.console.auth.login" and r.levelno == logging.WARNING
+        ]
+        assert len(warn_records) == 1
+        assert warn_records[0].args[0] == "user@example.com"
+        assert warn_records[0].args[1] == LoginFailureReason.ACCOUNT_BANNED
 
 
 class TestLogoutApi:

api/tests/unit_tests/controllers/console/snippets/test_snippet_workflow_draft_variable.py

@@ -1,3 +1,4 @@
+from inspect import unwrap
 from types import SimpleNamespace
 from unittest.mock import Mock
 
@@ -10,12 +11,6 @@ from models.account import Account, AccountStatus
 from services.workflow_draft_variable_service import WorkflowDraftVariableList
 
 
-def _unwrap(func):
-    while hasattr(func, "__wrapped__"):
-        func = func.__wrapped__
-    return func
-
-
 def _make_account() -> Account:
     account = Account(
         name="tester",
@@ -66,7 +61,7 @@ def test_ensure_snippet_draft_variable_row_allowed_accepts_canvas_node_variable(
 
 def test_conversation_variables_returns_empty_list(app: Flask):
     api = module.SnippetConversationVariableCollectionApi()
-    handler = _unwrap(api.get)
+    handler = unwrap(api.get)
 
     with app.test_request_context("/"):
         result = handler(api, _make_account(), snippet=SimpleNamespace(id="snippet-1"))
@@ -76,7 +71,7 @@ def test_conversation_variables_returns_empty_list(app: Flask):
 
 def test_system_variables_returns_empty_list(app: Flask):
     api = module.SnippetSystemVariableCollectionApi()
-    handler = _unwrap(api.get)
+    handler = unwrap(api.get)
 
     with app.test_request_context("/"):
         result = handler(api, _make_account(), snippet=SimpleNamespace(id="snippet-1"))
@@ -91,7 +86,7 @@ def test_delete_variable_collection_deletes_current_user_variables(app: Flask, m
     db_session.return_value = SimpleNamespace()
     monkeypatch.setattr(module.db, "session", db_session)
     api = module.SnippetWorkflowVariableCollectionApi()
-    handler = _unwrap(api.delete)
+    handler = unwrap(api.delete)
 
     with app.test_request_context("/", method="DELETE"):
         response = handler(api, _make_account(), snippet=SimpleNamespace(id="snippet-1"))
@@ -109,7 +104,7 @@ def test_variable_collection_get_raises_when_draft_workflow_missing(app: Flask,
     )
 
     api = module.SnippetWorkflowVariableCollectionApi()
-    handler = _unwrap(api.get)
+    handler = unwrap(api.get)
 
     with app.test_request_context("/?page=1&limit=20"):
         with pytest.raises(module.DraftWorkflowNotExist):
@@ -140,7 +135,7 @@ def test_node_variable_collection_get_lists_node_variables(app: Flask, monkeypat
     )
 
     api = module.SnippetNodeVariableCollectionApi()
-    handler = _unwrap(api.get)
+    handler = unwrap(api.get)
 
     with app.test_request_context("/"):
         result = handler(api, _make_account(), snippet=SimpleNamespace(id="snippet-1"), node_id="llm-1")
@@ -158,7 +153,7 @@ def test_node_variable_collection_delete_deletes_node_variables(app: Flask, monk
     monkeypatch.setattr(module.db, "session", db_session)
 
     api = module.SnippetNodeVariableCollectionApi()
-    handler = _unwrap(api.delete)
+    handler = unwrap(api.delete)
 
     with app.test_request_context("/", method="DELETE"):
         response = handler(api, _make_account(), snippet=SimpleNamespace(id="snippet-1"), node_id="llm-1")
@@ -177,7 +172,7 @@ def test_variable_patch_returns_variable_when_no_changes(app: Flask, monkeypatch
     monkeypatch.setattr(module, "WorkflowDraftVariableService", Mock(return_value=draft_var_service))
 
     api = module.SnippetVariableApi()
-    handler = _unwrap(api.patch)
+    handler = unwrap(api.patch)
 
     with app.test_request_context("/", method="PATCH", json={}):
         result = handler(
@@ -202,7 +197,7 @@ def test_variable_delete_deletes_variable(app: Flask, monkeypatch: pytest.Monkey
     monkeypatch.setattr(module, "WorkflowDraftVariableService", Mock(return_value=draft_var_service))
 
     api = module.SnippetVariableApi()
-    handler = _unwrap(api.delete)
+    handler = unwrap(api.delete)
 
     with app.test_request_context("/", method="DELETE"):
         response = handler(api, _make_account(), snippet=SimpleNamespace(id="snippet-1"), variable_id="var-1")
@@ -230,7 +225,7 @@ def test_variable_reset_returns_no_content_when_reset_result_is_none(app: Flask,
     )
 
     api = module.SnippetVariableResetApi()
-    handler = _unwrap(api.put)
+    handler = unwrap(api.put)
 
     with app.test_request_context("/", method="PUT"):
         response = handler(api, _make_account(), snippet=SimpleNamespace(id="snippet-1"), variable_id="var-1")
@@ -260,7 +255,7 @@ def test_environment_variables_returns_workflow_environment_variables(app: Flask
     )
 
     api = module.SnippetEnvironmentVariableCollectionApi()
-    handler = _unwrap(api.get)
+    handler = unwrap(api.get)
 
     with app.test_request_context("/"):
         result = handler(api, _make_account(), snippet=SimpleNamespace(id="snippet-1"))

api/tests/unit_tests/controllers/console/test_spec.py

@@ -22,23 +22,17 @@ class TestSpecSchemaDefinitionsApi:
         assert status == 200
         assert resp == schema_definitions
 
-    def test_get_exception_returns_empty_list(self):
+    def test_get_exception_returns_empty_list(self, caplog):
         api = spec_module.SpecSchemaDefinitionsApi()
         method = unwrap(api.get)
 
-        with (
-            patch.object(
-                spec_module,
-                "SchemaManager",
-                side_effect=Exception("boom"),
-            ),
-            patch.object(
-                spec_module.logger,
-                "exception",
-            ) as log_exception,
+        with patch.object(
+            spec_module,
+            "SchemaManager",
+            side_effect=Exception("boom"),
         ):
             resp, status = method(api)
 
         assert status == 200
         assert resp == []
-        log_exception.assert_called_once()
+        assert "boom" in caplog.text

api/tests/unit_tests/controllers/console/workspace/test_rbac.py

@@ -201,10 +201,10 @@ class TestPaginationMapping:
             },
         ]
         assert response["pagination"] == {
-            "total_count": 5,
+            "total_count": 4,
             "per_page": 2,
             "current_page": 1,
-            "total_pages": 3,
+            "total_pages": 2,
         }
         mock_list.assert_not_called()
 

api/tests/unit_tests/controllers/console/workspace/test_workspace.py

@@ -1,4 +1,5 @@
 import inspect
+import logging
 from io import BytesIO
 from unittest.mock import MagicMock, patch
 
@@ -151,7 +152,9 @@ class TestTenantListApi:
         get_plan_bulk_mock.assert_called_once_with(["t1", "t2"])
         get_features_mock.assert_called_once_with("t2", exclude_vector_space=True)
 
-    def test_get_saas_path_falls_back_to_legacy_feature_path_on_bulk_error(self, app: Flask):
+    def test_get_saas_path_falls_back_to_legacy_feature_path_on_bulk_error(
+        self, app: Flask, caplog: pytest.LogCaptureFixture
+    ):
         """Test fallback to FeatureService when bulk billing returns empty result.
 
         BillingService.get_plan_bulk catches exceptions internally and returns empty dict,
@@ -170,6 +173,7 @@ class TestTenantListApi:
 
         with (
             app.test_request_context("/workspaces"),
+            caplog.at_level(logging.WARNING, logger="controllers.console.workspace.workspace"),
             patch(
                 "controllers.console.workspace.workspace.TenantService.get_workspaces_for_account",
                 return_value=[(tenant1, make_membership()), (tenant2, make_membership())],
@@ -185,7 +189,6 @@ class TestTenantListApi:
                 "controllers.console.workspace.workspace.FeatureService.get_features",
                 return_value=features,
             ) as get_features_mock,
-            patch("controllers.console.workspace.workspace.logger.warning") as logger_warning_mock,
         ):
             result, status = method(api, "t2", user)
 
@@ -194,7 +197,7 @@ class TestTenantListApi:
         assert result["workspaces"][1]["plan"] == CloudPlan.TEAM
         get_plan_bulk_mock.assert_called_once_with(["t1", "t2"])
         assert get_features_mock.call_count == 2
-        logger_warning_mock.assert_called_once()
+        assert "get_plan_bulk returned empty result, falling back to legacy feature path" in caplog.messages
 
     def test_get_billing_disabled_community_path(self, app: Flask):
         api = TenantListApi()
@@ -365,7 +368,7 @@ class TestTenantApi:
             with pytest.raises(Unauthorized):
                 method(api, user)
 
-    def test_post_info_path(self, app: Flask):
+    def test_post_info_path(self, app: Flask, caplog: pytest.LogCaptureFixture):
         api = TenantApi()
         method = inspect.unwrap(api.post)
 
@@ -374,15 +377,15 @@ class TestTenantApi:
 
         with (
             app.test_request_context("/info"),
+            caplog.at_level(logging.WARNING, logger="controllers.console.workspace.workspace"),
             patch(
                 "controllers.console.workspace.workspace.WorkspaceService.get_tenant_info",
                 return_value={"id": "t1"},
             ),
-            patch("controllers.console.workspace.workspace.logger.warning") as warn_mock,
         ):
             result, status = method(api, user)
 
-        warn_mock.assert_called_once()
+        assert "Deprecated URL /info was used." in caplog.messages
         assert status == 200
 
 

api/tests/unit_tests/controllers/openapi/auth/test_pipeline.py

@@ -321,3 +321,56 @@ def test_guard_no_external_identity_when_subject_email_absent(app):
             view()
 
     assert received["data"].external_identity is None
+
+
+# --- auth-failure mapping (no raw 500 leak) ---
+
+
+def test_guard_expired_token_raises_session_expired_401(app):
+    from controllers.openapi._errors import OpenApiErrorCode, SessionExpired
+    from libs.oauth_bearer import TokenExpiredError
+
+    router = _make_router()
+
+    with app.test_request_context("/test", headers={"Authorization": "Bearer tok"}):
+        with (
+            patch("controllers.openapi.auth.pipeline.extract_bearer", return_value="tok"),
+            patch("controllers.openapi.auth.pipeline.get_authenticator") as mock_auth,
+            patch("controllers.openapi.auth.pipeline.current_edition", return_value=Edition.CE),
+        ):
+            mock_auth.return_value.authenticate.side_effect = TokenExpiredError("token_expired")
+
+            @router.guard(scope=Scope.FULL)
+            def view(*, auth_data):
+                pass
+
+            with pytest.raises(SessionExpired) as exc:
+                view()
+
+    assert exc.value.code == 401
+    assert exc.value.error_code == OpenApiErrorCode.TOKEN_EXPIRED
+
+
+def test_guard_invalid_token_raises_unified_401_not_500(app):
+    from controllers.openapi._errors import InvalidBearer, OpenApiErrorCode
+    from libs.oauth_bearer import InvalidBearerError
+
+    router = _make_router()
+
+    with app.test_request_context("/test", headers={"Authorization": "Bearer tok"}):
+        with (
+            patch("controllers.openapi.auth.pipeline.extract_bearer", return_value="tok"),
+            patch("controllers.openapi.auth.pipeline.get_authenticator") as mock_auth,
+            patch("controllers.openapi.auth.pipeline.current_edition", return_value=Edition.CE),
+        ):
+            mock_auth.return_value.authenticate.side_effect = InvalidBearerError("invalid_bearer")
+
+            @router.guard(scope=Scope.FULL)
+            def view(*, auth_data):
+                pass
+
+            with pytest.raises(InvalidBearer) as exc:
+                view()
+
+    assert exc.value.code == 401
+    assert exc.value.error_code == OpenApiErrorCode.UNAUTHORIZED

api/tests/unit_tests/controllers/openapi/test_app_run_streaming.py

@@ -3,13 +3,36 @@
 from __future__ import annotations
 
 import sys
-from types import SimpleNamespace
+import uuid
 from unittest.mock import Mock
 
 import pytest
 from flask import Flask
 
 from controllers.openapi._models import AppRunRequest
+from models import Account
+from models.model import App, AppMode
+
+_TEST_APP_ID = str(uuid.uuid4())
+_TEST_TENANT_ID = str(uuid.uuid4())
+_TEST_ACCOUNT_ID = str(uuid.uuid4())
+
+
+def _make_app() -> App:
+    app = App()
+    app.id = _TEST_APP_ID
+    app.tenant_id = _TEST_TENANT_ID
+    app.name = "Streaming app"
+    app.mode = AppMode.CHAT
+    app.enable_site = False
+    app.enable_api = True
+    return app
+
+
+def _make_account() -> Account:
+    account = Account(name="OpenAPI caller", email="caller@example.com")
+    account.id = _TEST_ACCOUNT_ID
+    return account
 
 
 def test_app_run_request_has_no_response_mode_field():
@@ -40,15 +63,19 @@ def test_run_chat_always_calls_generate_with_streaming_true(
     from controllers.openapi.app_run import _run_chat
 
     generate_mock = Mock(return_value=iter([]))
+
+    class GenerateService:
+        generate = generate_mock
+
     monkeypatch.setattr(
         sys.modules["controllers.openapi.app_run"],
         "AppGenerateService",
-        SimpleNamespace(generate=generate_mock),
+        GenerateService,
     )
-    with app.test_request_context("/openapi/v1/apps/app-1/run", method="POST"):
+    with app.test_request_context(f"/openapi/v1/apps/{_TEST_APP_ID}/run", method="POST"):
         _run_chat(
-            SimpleNamespace(id="app-1", tenant_id="t-1"),
-            SimpleNamespace(id="acct-1"),
+            _make_app(),
+            _make_account(),
             AppRunRequest(inputs={}, query="hello"),
         )
     _, kwargs = generate_mock.call_args
@@ -80,11 +107,11 @@ def test_stop_task_calls_queue_manager_and_graph_engine(app: Flask, bypass_pipel
 
     auth_data = AuthData.model_construct(
         token_type=TokenType.OAUTH_ACCOUNT,
-        account_id=uuid.uuid4(),
+        account_id=uuid.UUID(_TEST_ACCOUNT_ID),
         token_hash="test",
         scopes=frozenset({Scope.FULL}),
-        app=SimpleNamespace(id="app-1", tenant_id="t-1"),
-        caller=SimpleNamespace(id="acct-1"),
+        app=_make_app(),
+        caller=_make_account(),
         caller_kind="account",
     )
 

api/tests/unit_tests/controllers/openapi/test_contract.py

@@ -5,6 +5,7 @@ view function decorated with @accepts/@returns, driven inside a request context.
 """
 
 from functools import wraps
+from typing import Any, cast
 
 import pytest
 from pydantic import BaseModel, ConfigDict, Field
@@ -100,7 +101,7 @@ def test_accepts_validation_error_is_sanitized_and_structured(app):
         with pytest.raises(UnprocessableEntity) as exc_info:
             view()
 
-    data = exc_info.value.data
+    data = cast(dict[str, Any], cast(Any, exc_info.value).data)
     assert data["message"] == "Request validation failed"
     assert isinstance(data["errors"], list)
     assert data["errors"]

api/tests/unit_tests/controllers/openapi/test_error_contract.py

@@ -33,6 +33,7 @@ from controllers.openapi._errors import (
     OpenApiErrorCode,
     OpenApiErrorFormatter,
     RecipientSurfaceMismatch,
+    SessionExpired,
 )
 from controllers.service_api.app.error import (
     AppUnavailableError,
@@ -353,3 +354,20 @@ class TestErrorCodeEnumRegistration:
         schema = model.__schema__
         assert schema["type"] == "string"
         assert set(schema["enum"]) == {member.value for member in OpenApiErrorCode}
+
+
+class TestSessionExpired:
+    def test_session_expired_emits_token_expired_401_with_hint(self):
+        fmt = OpenApiErrorFormatter()
+        e = SessionExpired()
+        data = {"code": "unauthorized", "message": e.description, "status": 401}
+
+        wire = fmt.finalize(e, data, 401)
+
+        assert wire["code"] == OpenApiErrorCode.TOKEN_EXPIRED
+        assert wire["status"] == 401
+        assert wire["hint"]
+
+    def test_session_expired_code_is_401(self):
+        assert SessionExpired.code == 401
+        assert SessionExpired.error_code == OpenApiErrorCode.TOKEN_EXPIRED

api/tests/unit_tests/controllers/web/test_web_login.py

@@ -1,4 +1,5 @@
 import base64
+import logging
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
@@ -16,6 +17,13 @@ def encode_code(code: str) -> str:
     return base64.b64encode(code.encode("utf-8")).decode()
 
 
+def assert_login_failure_logged(caplog: pytest.LogCaptureFixture, email: str, reason: LoginFailureReason) -> None:
+    records = [record for record in caplog.records if record.name == "controllers.web.login"]
+    assert len(records) == 1
+    assert records[0].args[0] == email
+    assert records[0].args[1] == reason
+
+
 @pytest.fixture
 def app():
     flask_app = Flask(__name__)
@@ -114,10 +122,10 @@ class TestLoginApi:
         "controllers.web.login.WebAppAuthService.authenticate",
         side_effect=services.errors.account.AccountLoginError(),
     )
-    def test_login_banned_account(self, mock_auth: MagicMock, app: Flask) -> None:
+    def test_login_banned_account(self, mock_auth: MagicMock, app: Flask, caplog: pytest.LogCaptureFixture) -> None:
         from controllers.console.error import AccountBannedError
 
-        with patch("controllers.web.login.logger.warning") as mock_log_warning:
+        with caplog.at_level(logging.WARNING, logger="controllers.web.login"):
             with app.test_request_context(
                 "/web/login",
                 method="POST",
@@ -126,18 +134,16 @@ class TestLoginApi:
                 with pytest.raises(AccountBannedError):
                     LoginApi().post()
 
-        assert mock_log_warning.call_count == 1
-        assert mock_log_warning.call_args.args[1] == "user@example.com"
-        assert mock_log_warning.call_args.args[2] == LoginFailureReason.ACCOUNT_BANNED
+        assert_login_failure_logged(caplog, "user@example.com", LoginFailureReason.ACCOUNT_BANNED)
 
     @patch(
         "controllers.web.login.WebAppAuthService.authenticate",
         side_effect=services.errors.account.AccountPasswordError(),
     )
-    def test_login_wrong_password(self, mock_auth: MagicMock, app: Flask) -> None:
+    def test_login_wrong_password(self, mock_auth: MagicMock, app: Flask, caplog: pytest.LogCaptureFixture) -> None:
         from controllers.console.auth.error import AuthenticationFailedError
 
-        with patch("controllers.web.login.logger.warning") as mock_log_warning:
+        with caplog.at_level(logging.WARNING, logger="controllers.web.login"):
             with app.test_request_context(
                 "/web/login",
                 method="POST",
@@ -146,18 +152,16 @@ class TestLoginApi:
                 with pytest.raises(AuthenticationFailedError):
                     LoginApi().post()
 
-        assert mock_log_warning.call_count == 1
-        assert mock_log_warning.call_args.args[1] == "user@example.com"
-        assert mock_log_warning.call_args.args[2] == LoginFailureReason.INVALID_CREDENTIALS
+        assert_login_failure_logged(caplog, "user@example.com", LoginFailureReason.INVALID_CREDENTIALS)
 
     @patch(
         "controllers.web.login.WebAppAuthService.authenticate",
         side_effect=services.errors.account.AccountNotFoundError(),
     )
-    def test_login_account_not_found(self, mock_auth: MagicMock, app: Flask) -> None:
+    def test_login_account_not_found(self, mock_auth: MagicMock, app: Flask, caplog: pytest.LogCaptureFixture) -> None:
         from controllers.console.auth.error import AuthenticationFailedError
 
-        with patch("controllers.web.login.logger.warning") as mock_log_warning:
+        with caplog.at_level(logging.WARNING, logger="controllers.web.login"):
             with app.test_request_context(
                 "/web/login",
                 method="POST",
@@ -166,13 +170,13 @@ class TestLoginApi:
                 with pytest.raises(AuthenticationFailedError):
                     LoginApi().post()
 
-        assert mock_log_warning.call_count == 1
-        assert mock_log_warning.call_args.args[1] == "missing@example.com"
-        assert mock_log_warning.call_args.args[2] == LoginFailureReason.ACCOUNT_NOT_FOUND
+        assert_login_failure_logged(caplog, "missing@example.com", LoginFailureReason.ACCOUNT_NOT_FOUND)
 
     @patch("controllers.web.login.WebAppAuthService.get_email_code_login_data", return_value=None)
-    def test_email_code_login_logs_invalid_token(self, mock_get_token_data: MagicMock, app: Flask) -> None:
-        with patch("controllers.web.login.logger.warning") as mock_log_warning:
+    def test_email_code_login_logs_invalid_token(
+        self, mock_get_token_data: MagicMock, app: Flask, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        with caplog.at_level(logging.WARNING, logger="controllers.web.login"):
             with app.test_request_context(
                 "/web/email-code-login/validity",
                 method="POST",
@@ -182,9 +186,7 @@ class TestLoginApi:
                     EmailCodeLoginApi().post()
 
         mock_get_token_data.assert_called_once_with("token-123")
-        assert mock_log_warning.call_count == 1
-        assert mock_log_warning.call_args.args[1] == "user@example.com"
-        assert mock_log_warning.call_args.args[2] == LoginFailureReason.INVALID_EMAIL_CODE_TOKEN
+        assert_login_failure_logged(caplog, "user@example.com", LoginFailureReason.INVALID_EMAIL_CODE_TOKEN)
 
     @patch("controllers.web.login.WebAppAuthService.revoke_email_code_login_token")
     @patch(
@@ -201,10 +203,11 @@ class TestLoginApi:
         mock_get_user: MagicMock,
         mock_revoke_token: MagicMock,
         app: Flask,
+        caplog: pytest.LogCaptureFixture,
     ) -> None:
         from controllers.console.error import AccountBannedError
 
-        with patch("controllers.web.login.logger.warning") as mock_log_warning:
+        with caplog.at_level(logging.WARNING, logger="controllers.web.login"):
             with app.test_request_context(
                 "/web/email-code-login/validity",
                 method="POST",
@@ -215,9 +218,7 @@ class TestLoginApi:
 
         mock_get_token_data.assert_called_once_with("token-123")
         mock_revoke_token.assert_called_once_with("token-123")
-        assert mock_log_warning.call_count == 1
-        assert mock_log_warning.call_args.args[1] == "user@example.com"
-        assert mock_log_warning.call_args.args[2] == LoginFailureReason.ACCOUNT_BANNED
+        assert_login_failure_logged(caplog, "user@example.com", LoginFailureReason.ACCOUNT_BANNED)
 
 
 class TestLoginStatusApi:

api/tests/unit_tests/core/app/apps/advanced_chat/test_app_generator.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 from contextlib import contextmanager
 from types import SimpleNamespace
 from unittest.mock import MagicMock
@@ -961,7 +962,9 @@ class TestAdvancedChatAppGeneratorInternals:
                 stream=False,
             )
 
-    def test_handle_response_re_raises_value_error(self, monkeypatch: pytest.MonkeyPatch):
+    def test_handle_response_re_raises_value_error(
+        self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+    ):
         generator = AdvancedChatAppGenerator()
         generator._dialogue_count = 1
         app_config = self._build_app_config()
@@ -986,29 +989,28 @@ class TestAdvancedChatAppGeneratorInternals:
             def process(self):
                 raise ValueError("other error")
 
-        logger_exception = MagicMock()
-        monkeypatch.setattr("core.app.apps.advanced_chat.app_generator.logger.exception", logger_exception)
         monkeypatch.setattr("core.app.apps.advanced_chat.app_generator.AdvancedChatAppGenerateTaskPipeline", _Pipeline)
 
-        with pytest.raises(ValueError, match="other error"):
-            generator._handle_advanced_chat_response(
-                application_generate_entity=application_generate_entity,
-                workflow=WorkflowSnapshot(id="wf", tenant_id="tenant", features_dict={}),
-                queue_manager=SimpleNamespace(),
-                conversation=ConversationSnapshot(id="conv", mode=AppMode.ADVANCED_CHAT),
-                message=MessageSnapshot(
-                    id="msg",
-                    query="hello",
-                    created_at=naive_utc_now(),
-                    status=MessageStatus.NORMAL,
-                    answer="",
-                ),
-                user=SimpleNamespace(),
-                draft_var_saver_factory=lambda **kwargs: None,
-                stream=False,
-            )
+        with caplog.at_level(logging.ERROR, logger="core.app.apps.advanced_chat.app_generator"):
+            with pytest.raises(ValueError, match="other error"):
+                generator._handle_advanced_chat_response(
+                    application_generate_entity=application_generate_entity,
+                    workflow=WorkflowSnapshot(id="wf", tenant_id="tenant", features_dict={}),
+                    queue_manager=SimpleNamespace(),
+                    conversation=ConversationSnapshot(id="conv", mode=AppMode.ADVANCED_CHAT),
+                    message=MessageSnapshot(
+                        id="msg",
+                        query="hello",
+                        created_at=naive_utc_now(),
+                        status=MessageStatus.NORMAL,
+                        answer="",
+                    ),
+                    user=SimpleNamespace(),
+                    draft_var_saver_factory=lambda **kwargs: None,
+                    stream=False,
+                )
 
-        logger_exception.assert_called_once()
+        assert "Failed to process generate task pipeline, conversation_id: conv" in caplog.messages
 
     def test_generate_worker_handles_invoke_auth_error(self, monkeypatch: pytest.MonkeyPatch):
         generator = AdvancedChatAppGenerator()

api/tests/unit_tests/core/app/apps/agent_chat/test_agent_chat_app_generator.py

@@ -1,4 +1,5 @@
 import contextlib
+import logging
 
 import pytest
 from pydantic import ValidationError
@@ -274,7 +275,9 @@ class TestAgentChatAppGeneratorWorker:
 
         assert queue_manager.publish_error.called
 
-    def test_generate_worker_logs_value_error_when_debug(self, generator, mocker: MockerFixture):
+    def test_generate_worker_logs_value_error_when_debug(
+        self, generator, mocker: MockerFixture, caplog: pytest.LogCaptureFixture
+    ):
         queue_manager = mocker.MagicMock()
         generator._get_conversation = mocker.MagicMock(return_value=mocker.MagicMock())
         generator._get_message = mocker.MagicMock(return_value=mocker.MagicMock())
@@ -285,15 +288,15 @@ class TestAgentChatAppGeneratorWorker:
         mocker.patch("core.app.apps.agent_chat.app_generator.db.session.close")
 
         mocker.patch("core.app.apps.agent_chat.app_generator.dify_config", new=mocker.MagicMock(DEBUG=True))
-        logger = mocker.patch("core.app.apps.agent_chat.app_generator.logger")
 
-        generator._generate_worker(
-            flask_app=mocker.MagicMock(),
-            context=mocker.MagicMock(),
-            application_generate_entity=mocker.MagicMock(),
-            queue_manager=queue_manager,
-            conversation_id="conv",
-            message_id="msg",
-        )
+        with caplog.at_level(logging.ERROR, logger="core.app.apps.agent_chat.app_generator"):
+            generator._generate_worker(
+                flask_app=mocker.MagicMock(),
+                context=mocker.MagicMock(),
+                application_generate_entity=mocker.MagicMock(),
+                queue_manager=queue_manager,
+                conversation_id="conv",
+                message_id="msg",
+            )
 
-        logger.exception.assert_called_once()
+        assert "Error when generating" in caplog.messages

api/tests/unit_tests/core/app/apps/test_base_app_runner.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
@@ -263,11 +264,11 @@ class TestAppRunner:
                 files=[],
             )
 
-    def test_handle_invoke_result_stream_routes_chunks_and_builds_message(self, monkeypatch: pytest.MonkeyPatch):
+    def test_handle_invoke_result_stream_routes_chunks_and_builds_message(
+        self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+    ):
         runner = AppRunner()
         queue = _QueueRecorder()
-        warning_logger = MagicMock()
-        monkeypatch.setattr("core.app.apps.base_app_runner._logger.warning", warning_logger)
 
         image_content = ImagePromptMessageContent(
             url="https://example.com/image.png", format="png", mime_type="image/png"
@@ -290,23 +291,24 @@ class TestAppRunner:
                 ),
             )
 
-        runner._handle_invoke_result(
-            invoke_result=_stream(),
-            queue_manager=queue,
-            stream=True,
-            agent=False,
-        )
+        with caplog.at_level(logging.WARNING, logger="core.app.apps.base_app_runner"):
+            runner._handle_invoke_result(
+                invoke_result=_stream(),
+                queue_manager=queue,
+                stream=True,
+                agent=False,
+            )
 
         assert isinstance(queue.events[0], QueueLLMChunkEvent)
         assert isinstance(queue.events[-1], QueueMessageEndEvent)
         assert queue.events[-1].llm_result.message.content == "abc"
-        warning_logger.assert_called_once()
+        assert "Received multimodal output but missing required parameters" in caplog.messages
 
-    def test_handle_invoke_result_stream_agent_mode_handles_multimodal_errors(self, monkeypatch: pytest.MonkeyPatch):
+    def test_handle_invoke_result_stream_agent_mode_handles_multimodal_errors(
+        self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+    ):
         runner = AppRunner()
         queue = _QueueRecorder()
-        exception_logger = MagicMock()
-        monkeypatch.setattr("core.app.apps.base_app_runner._logger.exception", exception_logger)
 
         monkeypatch.setattr(
             runner,
@@ -335,19 +337,20 @@ class TestAppRunner:
                 ),
             )
 
-        runner._handle_invoke_result_stream(
-            invoke_result=_stream(),
-            queue_manager=queue,
-            agent=True,
-            message_id="message-id",
-            user_id="user-id",
-            tenant_id="tenant-id",
-        )
+        with caplog.at_level(logging.ERROR, logger="core.app.apps.base_app_runner"):
+            runner._handle_invoke_result_stream(
+                invoke_result=_stream(),
+                queue_manager=queue,
+                agent=True,
+                message_id="message-id",
+                user_id="user-id",
+                tenant_id="tenant-id",
+            )
 
         assert isinstance(queue.events[0], QueueAgentMessageEvent)
         assert isinstance(queue.events[-1], QueueMessageEndEvent)
         assert queue.events[-1].llm_result.usage == usage
-        exception_logger.assert_called_once()
+        assert "Failed to handle multimodal image output" in caplog.messages
 
     def test_handle_invoke_result_stream_closes_generator_when_stopped(self):
         runner = AppRunner()

api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline_core.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 from contextlib import contextmanager
 from types import SimpleNamespace
 from unittest.mock import MagicMock
@@ -639,7 +640,9 @@ class TestWorkflowGenerateTaskPipeline:
         assert sleep_spy
         assert any(isinstance(item, MessageAudioEndStreamResponse) for item in responses)
 
-    def test_wrapper_process_stream_response_handles_audio_exception(self, monkeypatch: pytest.MonkeyPatch):
+    def test_wrapper_process_stream_response_handles_audio_exception(
+        self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+    ):
         pipeline = _make_pipeline()
         pipeline._workflow_features_dict = {
             "text_to_speech": {"enabled": True, "autoPlay": "enabled", "voice": "v", "language": "en"}
@@ -659,20 +662,16 @@ class TestWorkflowGenerateTaskPipeline:
             def publish(self, message):
                 _ = message
 
-        logger_exception = []
         monkeypatch.setattr("core.app.apps.workflow.generate_task_pipeline.time.time", lambda: 0.0)
-        monkeypatch.setattr(
-            "core.app.apps.workflow.generate_task_pipeline.logger.exception",
-            lambda *args, **kwargs: logger_exception.append((args, kwargs)),
-        )
         monkeypatch.setattr(
             "core.app.apps.workflow.generate_task_pipeline.AppGeneratorTTSPublisher",
             _Publisher,
         )
 
-        responses = list(pipeline._wrapper_process_stream_response())
+        with caplog.at_level(logging.ERROR, logger="core.app.apps.workflow.generate_task_pipeline"):
+            responses = list(pipeline._wrapper_process_stream_response())
 
-        assert logger_exception
+        assert "Fails to get audio trunk, task_id: task" in caplog.messages
         assert any(isinstance(item, MessageAudioEndStreamResponse) for item in responses)
 
     def test_database_session_rolls_back_on_error(self, monkeypatch: pytest.MonkeyPatch):

api/tests/unit_tests/core/entities/test_entities_provider_configuration.py

@@ -2042,7 +2042,9 @@ def test_get_custom_provider_models_skips_schema_models_with_mismatched_type() -
     assert all(model.model != "embed-model" for model in models)
 
 
-def test_get_custom_provider_models_skips_custom_models_on_schema_error_or_none() -> None:
+def test_get_custom_provider_models_skips_custom_models_on_schema_error_or_none(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
     configuration = _build_provider_configuration()
     configuration.custom_configuration.models = [
         CustomModelConfiguration(model="error-custom", model_type=ModelType.LLM, credentials={"k": "v"}),
@@ -2064,7 +2066,7 @@ def test_get_custom_provider_models_skips_custom_models_on_schema_error_or_none(
             return None
         return _build_ai_model(model)
 
-    with patch("core.entities.provider_configuration.logger.warning") as mock_warning:
+    with caplog.at_level(logging.WARNING, logger="core.entities.provider_configuration"):
         with patch.object(ProviderConfiguration, "get_model_schema", side_effect=_schema):
             models = configuration._get_custom_provider_models(
                 model_types=[ModelType.LLM],
@@ -2072,6 +2074,6 @@ def test_get_custom_provider_models_skips_custom_models_on_schema_error_or_none(
                 model_setting_map={},
             )
 
-    assert mock_warning.call_count == 1
+    assert "get custom model schema failed, boom" in caplog.messages
     assert any(model.model == "ok-custom" for model in models)
     assert all(model.model != "none-custom" for model in models)

api/tests/unit_tests/enterprise/telemetry/test_enterprise_trace.py

@@ -21,6 +21,7 @@ from core.ops.entities.trace_entity import (
     WorkflowNodeTraceInfo,
     WorkflowTraceInfo,
 )
+from enterprise.telemetry.enterprise_trace import EnterpriseOtelTrace
 from enterprise.telemetry.entities import (
     EnterpriseTelemetryCounter,
     EnterpriseTelemetryEvent,
@@ -297,43 +298,43 @@ def test_init_succeeds_with_valid_exporter(mock_exporter):
 
 
 class TestSafePayloadValue:
-    def test_string_passthrough(self, trace_handler):
+    def test_string_passthrough(self, trace_handler: EnterpriseOtelTrace):
         assert trace_handler._safe_payload_value("hello") == "hello"
 
-    def test_dict_passthrough(self, trace_handler):
+    def test_dict_passthrough(self, trace_handler: EnterpriseOtelTrace):
         d = {"key": "val"}
         assert trace_handler._safe_payload_value(d) == d
 
-    def test_list_passthrough(self, trace_handler):
+    def test_list_passthrough(self, trace_handler: EnterpriseOtelTrace):
         lst = [1, 2, 3]
         assert trace_handler._safe_payload_value(lst) == lst
 
-    def test_none_returns_none(self, trace_handler):
+    def test_none_returns_none(self, trace_handler: EnterpriseOtelTrace):
         assert trace_handler._safe_payload_value(None) is None
 
-    def test_int_returns_none(self, trace_handler):
+    def test_int_returns_none(self, trace_handler: EnterpriseOtelTrace):
         assert trace_handler._safe_payload_value(42) is None
 
-    def test_bool_returns_none(self, trace_handler):
+    def test_bool_returns_none(self, trace_handler: EnterpriseOtelTrace):
         assert trace_handler._safe_payload_value(True) is None
 
 
 class TestMaybeJson:
-    def test_none_returns_none(self, trace_handler):
+    def test_none_returns_none(self, trace_handler: EnterpriseOtelTrace):
         assert trace_handler._maybe_json(None) is None
 
-    def test_string_passthrough(self, trace_handler):
+    def test_string_passthrough(self, trace_handler: EnterpriseOtelTrace):
         assert trace_handler._maybe_json("hello") == "hello"
 
-    def test_dict_serialised(self, trace_handler):
+    def test_dict_serialised(self, trace_handler: EnterpriseOtelTrace):
         result = trace_handler._maybe_json({"a": 1})
         assert result == json.dumps({"a": 1})
 
-    def test_list_serialised(self, trace_handler):
+    def test_list_serialised(self, trace_handler: EnterpriseOtelTrace):
         result = trace_handler._maybe_json([1, 2])
         assert result == "[1, 2]"
 
-    def test_non_serialisable_falls_back_to_str(self, trace_handler):
+    def test_non_serialisable_falls_back_to_str(self, trace_handler: EnterpriseOtelTrace):
         class Unserializable:
             def __repr__(self):
                 return "Unserializable()"
@@ -344,22 +345,22 @@ class TestMaybeJson:
 
 
 class TestContentOrRef:
-    def test_returns_content_when_include_content_true(self, trace_handler, mock_exporter):
+    def test_returns_content_when_include_content_true(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = True
         result = trace_handler._content_or_ref("actual content", "ref:x=1")
         assert result == "actual content"
 
-    def test_returns_ref_when_include_content_false(self, trace_handler, mock_exporter):
+    def test_returns_ref_when_include_content_false(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = False
         result = trace_handler._content_or_ref("actual content", "ref:x=1")
         assert result == "ref:x=1"
 
-    def test_dict_serialised_when_include_content_true(self, trace_handler, mock_exporter):
+    def test_dict_serialised_when_include_content_true(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = True
         result = trace_handler._content_or_ref({"key": "val"}, "ref:x=1")
         assert result == json.dumps({"key": "val"})
 
-    def test_none_returns_none_when_include_content_true(self, trace_handler, mock_exporter):
+    def test_none_returns_none_when_include_content_true(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = True
         result = trace_handler._content_or_ref(None, "ref:x=1")
         assert result is None
@@ -371,67 +372,67 @@ class TestContentOrRef:
 
 
 class TestTraceDispatcher:
-    def test_dispatches_workflow_trace(self, trace_handler):
+    def test_dispatches_workflow_trace(self, trace_handler: EnterpriseOtelTrace):
         with patch.object(trace_handler, "_workflow_trace") as mock_method:
             info = make_workflow_info()
             trace_handler.trace(info)
             mock_method.assert_called_once_with(info)
 
-    def test_dispatches_message_trace(self, trace_handler):
+    def test_dispatches_message_trace(self, trace_handler: EnterpriseOtelTrace):
         with patch.object(trace_handler, "_message_trace") as mock_method:
             info = make_message_info()
             trace_handler.trace(info)
             mock_method.assert_called_once_with(info)
 
-    def test_dispatches_tool_trace(self, trace_handler):
+    def test_dispatches_tool_trace(self, trace_handler: EnterpriseOtelTrace):
         with patch.object(trace_handler, "_tool_trace") as mock_method:
             info = make_tool_info()
             trace_handler.trace(info)
             mock_method.assert_called_once_with(info)
 
-    def test_dispatches_draft_node_execution_trace(self, trace_handler):
+    def test_dispatches_draft_node_execution_trace(self, trace_handler: EnterpriseOtelTrace):
         with patch.object(trace_handler, "_draft_node_execution_trace") as mock_method:
             info = make_draft_node_info()
             trace_handler.trace(info)
             mock_method.assert_called_once_with(info)
 
-    def test_dispatches_node_execution_trace(self, trace_handler):
+    def test_dispatches_node_execution_trace(self, trace_handler: EnterpriseOtelTrace):
         with patch.object(trace_handler, "_node_execution_trace") as mock_method:
             info = make_node_info()
             trace_handler.trace(info)
             mock_method.assert_called_once_with(info)
 
-    def test_dispatches_moderation_trace(self, trace_handler):
+    def test_dispatches_moderation_trace(self, trace_handler: EnterpriseOtelTrace):
         with patch.object(trace_handler, "_moderation_trace") as mock_method:
             info = make_moderation_info()
             trace_handler.trace(info)
             mock_method.assert_called_once_with(info)
 
-    def test_dispatches_suggested_question_trace(self, trace_handler):
+    def test_dispatches_suggested_question_trace(self, trace_handler: EnterpriseOtelTrace):
         with patch.object(trace_handler, "_suggested_question_trace") as mock_method:
             info = make_suggested_question_info()
             trace_handler.trace(info)
             mock_method.assert_called_once_with(info)
 
-    def test_dispatches_dataset_retrieval_trace(self, trace_handler):
+    def test_dispatches_dataset_retrieval_trace(self, trace_handler: EnterpriseOtelTrace):
         with patch.object(trace_handler, "_dataset_retrieval_trace") as mock_method:
             info = make_dataset_retrieval_info()
             trace_handler.trace(info)
             mock_method.assert_called_once_with(info)
 
-    def test_dispatches_generate_name_trace(self, trace_handler):
+    def test_dispatches_generate_name_trace(self, trace_handler: EnterpriseOtelTrace):
         with patch.object(trace_handler, "_generate_name_trace") as mock_method:
             info = make_generate_name_info()
             trace_handler.trace(info)
             mock_method.assert_called_once_with(info)
 
-    def test_dispatches_prompt_generation_trace(self, trace_handler):
+    def test_dispatches_prompt_generation_trace(self, trace_handler: EnterpriseOtelTrace):
         with patch.object(trace_handler, "_prompt_generation_trace") as mock_method:
             info = make_prompt_generation_info()
             trace_handler.trace(info)
             mock_method.assert_called_once_with(info)
 
-    def test_draft_node_dispatched_before_node(self, trace_handler):
+    def test_draft_node_dispatched_before_node(self, trace_handler: EnterpriseOtelTrace):
         """DraftNodeExecutionTrace is a subclass of WorkflowNodeTraceInfo;
         it must be dispatched to _draft_node_execution_trace, not _node_execution_trace."""
         with (
@@ -450,7 +451,7 @@ class TestTraceDispatcher:
 
 
 class TestWorkflowTrace:
-    def test_emits_correct_span_attributes(self, trace_handler, mock_exporter):
+    def test_emits_correct_span_attributes(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log") as mock_log:
             info = make_workflow_info()
             trace_handler._workflow_trace(info)
@@ -465,7 +466,7 @@ class TestWorkflowTrace:
         assert attrs["dify.workflow.status"] == "succeeded"
         assert attrs["gen_ai.usage.total_tokens"] == 100
 
-    def test_span_timing_passed_correctly(self, trace_handler, mock_exporter):
+    def test_span_timing_passed_correctly(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             info = make_workflow_info()
             trace_handler._workflow_trace(info)
@@ -474,7 +475,7 @@ class TestWorkflowTrace:
         assert span_call[1]["start_time"] == _T0
         assert span_call[1]["end_time"] == _T1
 
-    def test_emits_companion_log_with_event_name(self, trace_handler, mock_exporter):
+    def test_emits_companion_log_with_event_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log") as mock_log:
             trace_handler._workflow_trace(make_workflow_info())
 
@@ -482,7 +483,7 @@ class TestWorkflowTrace:
         assert mock_log.call_args[1]["event_name"] == EnterpriseTelemetryEvent.WORKFLOW_RUN
         assert mock_log.call_args[1]["tenant_id"] == "tenant-abc"
 
-    def test_companion_log_includes_content_when_enabled(self, trace_handler, mock_exporter):
+    def test_companion_log_includes_content_when_enabled(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = True
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log") as mock_log:
             trace_handler._workflow_trace(make_workflow_info())
@@ -491,7 +492,7 @@ class TestWorkflowTrace:
         assert log_attrs["dify.workflow.inputs"] == json.dumps({"query": "hello"})
         assert log_attrs["dify.workflow.outputs"] == json.dumps({"answer": "world"})
 
-    def test_companion_log_uses_ref_when_content_disabled(self, trace_handler, mock_exporter):
+    def test_companion_log_uses_ref_when_content_disabled(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = False
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log") as mock_log:
             trace_handler._workflow_trace(make_workflow_info())
@@ -500,7 +501,7 @@ class TestWorkflowTrace:
         assert log_attrs["dify.workflow.inputs"].startswith("ref:workflow_run_id=")
         assert log_attrs["dify.workflow.outputs"].startswith("ref:workflow_run_id=")
 
-    def test_increments_token_counter(self, trace_handler, mock_exporter):
+    def test_increments_token_counter(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._workflow_trace(make_workflow_info())
 
@@ -510,7 +511,7 @@ class TestWorkflowTrace:
         assert len(token_calls) == 1
         assert token_calls[0][0][1] == 100
 
-    def test_increments_input_and_output_token_counters(self, trace_handler, mock_exporter):
+    def test_increments_input_and_output_token_counters(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._workflow_trace(make_workflow_info())
 
@@ -519,7 +520,7 @@ class TestWorkflowTrace:
         assert EnterpriseTelemetryCounter.INPUT_TOKENS in counter_names
         assert EnterpriseTelemetryCounter.OUTPUT_TOKENS in counter_names
 
-    def test_no_input_token_counter_when_prompt_tokens_zero(self, trace_handler, mock_exporter):
+    def test_no_input_token_counter_when_prompt_tokens_zero(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             info = make_workflow_info(prompt_tokens=0)
             trace_handler._workflow_trace(info)
@@ -528,7 +529,7 @@ class TestWorkflowTrace:
         counter_names = [c[0][0] for c in all_calls]
         assert EnterpriseTelemetryCounter.INPUT_TOKENS not in counter_names
 
-    def test_records_workflow_duration_histogram(self, trace_handler, mock_exporter):
+    def test_records_workflow_duration_histogram(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._workflow_trace(make_workflow_info())
 
@@ -537,7 +538,9 @@ class TestWorkflowTrace:
         assert hist_call[0][0] == EnterpriseTelemetryHistogram.WORKFLOW_DURATION
         assert hist_call[0][1] == pytest.approx(5.0)
 
-    def test_duration_falls_back_to_elapsed_time_when_timestamps_missing(self, trace_handler, mock_exporter):
+    def test_duration_falls_back_to_elapsed_time_when_timestamps_missing(
+        self, trace_handler: EnterpriseOtelTrace, mock_exporter
+    ):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             info = make_workflow_info(start_time=None, end_time=None, workflow_run_elapsed_time=7.3)
             trace_handler._workflow_trace(info)
@@ -545,7 +548,7 @@ class TestWorkflowTrace:
         hist_call = mock_exporter.record_histogram.call_args
         assert hist_call[0][1] == pytest.approx(7.3)
 
-    def test_duration_defaults_to_zero_when_no_timing(self, trace_handler, mock_exporter):
+    def test_duration_defaults_to_zero_when_no_timing(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             info = make_workflow_info(start_time=None, end_time=None, workflow_run_elapsed_time=0)
             trace_handler._workflow_trace(info)
@@ -553,7 +556,7 @@ class TestWorkflowTrace:
         hist_call = mock_exporter.record_histogram.call_args
         assert hist_call[0][1] == pytest.approx(0.0)
 
-    def test_error_path_increments_error_counter(self, trace_handler, mock_exporter):
+    def test_error_path_increments_error_counter(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             info = make_workflow_info(error="Something went wrong", workflow_run_status="failed")
             trace_handler._workflow_trace(info)
@@ -563,7 +566,7 @@ class TestWorkflowTrace:
         ]
         assert len(error_calls) == 1
 
-    def test_no_error_counter_on_success(self, trace_handler, mock_exporter):
+    def test_no_error_counter_on_success(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._workflow_trace(make_workflow_info())
 
@@ -572,7 +575,7 @@ class TestWorkflowTrace:
         ]
         assert len(error_calls) == 0
 
-    def test_parent_trace_context_injected_into_span_attrs(self, trace_handler, mock_exporter):
+    def test_parent_trace_context_injected_into_span_attrs(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             info = make_workflow_info(
                 metadata={
@@ -601,14 +604,14 @@ class TestWorkflowTrace:
 
 
 class TestNodeExecutionTrace:
-    def test_emits_span_with_node_execution_span_name(self, trace_handler, mock_exporter):
+    def test_emits_span_with_node_execution_span_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._node_execution_trace(make_node_info())
 
         span_call = mock_exporter.export_span.call_args
         assert span_call[0][0] == EnterpriseTelemetrySpan.NODE_EXECUTION
 
-    def test_span_contains_core_node_attributes(self, trace_handler, mock_exporter):
+    def test_span_contains_core_node_attributes(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._node_execution_trace(make_node_info())
 
@@ -620,7 +623,7 @@ class TestNodeExecutionTrace:
         assert attrs["gen_ai.request.model"] == "gpt-4"
         assert attrs["gen_ai.provider.name"] == "openai"
 
-    def test_increments_token_counters_when_tokens_present(self, trace_handler, mock_exporter):
+    def test_increments_token_counters_when_tokens_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._node_execution_trace(make_node_info())
 
@@ -629,7 +632,7 @@ class TestNodeExecutionTrace:
         assert EnterpriseTelemetryCounter.INPUT_TOKENS in counter_names
         assert EnterpriseTelemetryCounter.OUTPUT_TOKENS in counter_names
 
-    def test_no_token_counters_when_total_tokens_zero(self, trace_handler, mock_exporter):
+    def test_no_token_counters_when_total_tokens_zero(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._node_execution_trace(make_node_info(total_tokens=0))
 
@@ -637,7 +640,7 @@ class TestNodeExecutionTrace:
         assert EnterpriseTelemetryCounter.TOKENS not in counter_names
         assert EnterpriseTelemetryCounter.INPUT_TOKENS not in counter_names
 
-    def test_records_node_duration_histogram(self, trace_handler, mock_exporter):
+    def test_records_node_duration_histogram(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._node_execution_trace(make_node_info())
 
@@ -645,7 +648,7 @@ class TestNodeExecutionTrace:
         assert hist_call[0][0] == EnterpriseTelemetryHistogram.NODE_DURATION
         assert hist_call[0][1] == pytest.approx(2.5)
 
-    def test_error_path_increments_error_counter(self, trace_handler, mock_exporter):
+    def test_error_path_increments_error_counter(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._node_execution_trace(make_node_info(error="Node failed", status="failed"))
 
@@ -654,14 +657,16 @@ class TestNodeExecutionTrace:
         ]
         assert len(error_calls) == 1
 
-    def test_emits_companion_log_with_span_name_as_event(self, trace_handler, mock_exporter):
+    def test_emits_companion_log_with_span_name_as_event(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log") as mock_log:
             trace_handler._node_execution_trace(make_node_info())
 
         mock_log.assert_called_once()
         assert mock_log.call_args[1]["event_name"] == EnterpriseTelemetrySpan.NODE_EXECUTION.value
 
-    def test_plugin_name_added_to_duration_labels_for_tool_node(self, trace_handler, mock_exporter):
+    def test_plugin_name_added_to_duration_labels_for_tool_node(
+        self, trace_handler: EnterpriseOtelTrace, mock_exporter
+    ):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             info = make_node_info(
                 node_type="tool",
@@ -677,7 +682,7 @@ class TestNodeExecutionTrace:
         duration_labels = hist_call[0][2]
         assert duration_labels.get("plugin_name") == "my-plugin"
 
-    def test_plugin_name_not_added_for_non_tool_node(self, trace_handler, mock_exporter):
+    def test_plugin_name_not_added_for_non_tool_node(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             info = make_node_info(
                 node_type="llm",
@@ -693,7 +698,9 @@ class TestNodeExecutionTrace:
         duration_labels = hist_call[0][2]
         assert "plugin_name" not in duration_labels
 
-    def test_companion_log_inputs_use_ref_when_content_disabled(self, trace_handler, mock_exporter):
+    def test_companion_log_inputs_use_ref_when_content_disabled(
+        self, trace_handler: EnterpriseOtelTrace, mock_exporter
+    ):
         mock_exporter.include_content = False
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log") as mock_log:
             trace_handler._node_execution_trace(
@@ -711,14 +718,14 @@ class TestNodeExecutionTrace:
 
 
 class TestDraftNodeExecutionTrace:
-    def test_uses_draft_span_name(self, trace_handler, mock_exporter):
+    def test_uses_draft_span_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             trace_handler._draft_node_execution_trace(make_draft_node_info())
 
         span_call = mock_exporter.export_span.call_args
         assert span_call[0][0] == EnterpriseTelemetrySpan.DRAFT_NODE_EXECUTION
 
-    def test_correlation_id_is_node_execution_id(self, trace_handler, mock_exporter):
+    def test_correlation_id_is_node_execution_id(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             info = make_draft_node_info()
             trace_handler._draft_node_execution_trace(info)
@@ -726,7 +733,7 @@ class TestDraftNodeExecutionTrace:
         span_call = mock_exporter.export_span.call_args
         assert span_call[1]["correlation_id"] == "ne-draft-001"
 
-    def test_trace_correlation_override_is_workflow_run_id(self, trace_handler, mock_exporter):
+    def test_trace_correlation_override_is_workflow_run_id(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"):
             info = make_draft_node_info()
             trace_handler._draft_node_execution_trace(info)
@@ -734,7 +741,7 @@ class TestDraftNodeExecutionTrace:
         span_call = mock_exporter.export_span.call_args
         assert span_call[1]["trace_correlation_override"] == "run-draft-001"
 
-    def test_companion_log_uses_draft_span_name(self, trace_handler, mock_exporter):
+    def test_companion_log_uses_draft_span_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log") as mock_log:
             trace_handler._draft_node_execution_trace(make_draft_node_info())
 
@@ -747,34 +754,36 @@ class TestDraftNodeExecutionTrace:
 
 
 class TestMessageTrace:
-    def test_emits_event_with_correct_name(self, trace_handler, mock_exporter):
+    def test_emits_event_with_correct_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._message_trace(make_message_info())
 
         mock_emit.assert_called_once()
         assert mock_emit.call_args[1]["event_name"] == EnterpriseTelemetryEvent.MESSAGE_RUN
 
-    def test_emits_correct_tenant_and_user(self, trace_handler, mock_exporter):
+    def test_emits_correct_tenant_and_user(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._message_trace(make_message_info())
 
         assert mock_emit.call_args[1]["tenant_id"] == "tenant-abc"
 
-    def test_duration_computed_from_timestamps(self, trace_handler, mock_exporter):
+    def test_duration_computed_from_timestamps(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._message_trace(make_message_info())
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.message.duration"] == pytest.approx(5.0)
 
-    def test_no_duration_when_timestamps_missing(self, trace_handler, mock_exporter):
+    def test_no_duration_when_timestamps_missing(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._message_trace(make_message_info(start_time=None, end_time=None))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert "dify.message.duration" not in attrs
 
-    def test_records_duration_histogram_when_timestamps_present(self, trace_handler, mock_exporter):
+    def test_records_duration_histogram_when_timestamps_present(
+        self, trace_handler: EnterpriseOtelTrace, mock_exporter
+    ):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._message_trace(make_message_info())
 
@@ -786,14 +795,14 @@ class TestMessageTrace:
         assert len(hist_calls) == 1
         assert hist_calls[0][0][1] == pytest.approx(5.0)
 
-    def test_no_duration_histogram_when_timestamps_missing(self, trace_handler, mock_exporter):
+    def test_no_duration_histogram_when_timestamps_missing(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._message_trace(make_message_info(start_time=None, end_time=None))
 
         hist_names = [c[0][0] for c in mock_exporter.record_histogram.call_args_list]
         assert EnterpriseTelemetryHistogram.MESSAGE_DURATION not in hist_names
 
-    def test_records_ttft_histogram_when_present(self, trace_handler, mock_exporter):
+    def test_records_ttft_histogram_when_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._message_trace(make_message_info(gen_ai_server_time_to_first_token=0.42))
 
@@ -805,14 +814,14 @@ class TestMessageTrace:
         assert len(ttft_calls) == 1
         assert ttft_calls[0][0][1] == pytest.approx(0.42)
 
-    def test_no_ttft_histogram_when_not_present(self, trace_handler, mock_exporter):
+    def test_no_ttft_histogram_when_not_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._message_trace(make_message_info(gen_ai_server_time_to_first_token=None))
 
         hist_names = [c[0][0] for c in mock_exporter.record_histogram.call_args_list]
         assert EnterpriseTelemetryHistogram.MESSAGE_TTFT not in hist_names
 
-    def test_increments_token_counters(self, trace_handler, mock_exporter):
+    def test_increments_token_counters(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._message_trace(make_message_info())
 
@@ -821,7 +830,7 @@ class TestMessageTrace:
         assert EnterpriseTelemetryCounter.INPUT_TOKENS in counter_names
         assert EnterpriseTelemetryCounter.OUTPUT_TOKENS in counter_names
 
-    def test_error_path_increments_error_counter(self, trace_handler, mock_exporter):
+    def test_error_path_increments_error_counter(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._message_trace(make_message_info(error="LLM failed"))
 
@@ -830,7 +839,7 @@ class TestMessageTrace:
         ]
         assert len(error_calls) == 1
 
-    def test_inputs_and_outputs_gated_by_include_content(self, trace_handler, mock_exporter):
+    def test_inputs_and_outputs_gated_by_include_content(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = False
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._message_trace(make_message_info())
@@ -846,27 +855,27 @@ class TestMessageTrace:
 
 
 class TestToolTrace:
-    def test_emits_event_with_correct_name(self, trace_handler, mock_exporter):
+    def test_emits_event_with_correct_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._tool_trace(make_tool_info())
 
         assert mock_emit.call_args[1]["event_name"] == EnterpriseTelemetryEvent.TOOL_EXECUTION
 
-    def test_status_is_succeeded_on_success(self, trace_handler, mock_exporter):
+    def test_status_is_succeeded_on_success(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._tool_trace(make_tool_info())
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.tool.status"] == "succeeded"
 
-    def test_status_is_failed_on_error(self, trace_handler, mock_exporter):
+    def test_status_is_failed_on_error(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._tool_trace(make_tool_info(error="Tool error"))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.tool.status"] == "failed"
 
-    def test_records_tool_duration_histogram(self, trace_handler, mock_exporter):
+    def test_records_tool_duration_histogram(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._tool_trace(make_tool_info())
 
@@ -874,7 +883,7 @@ class TestToolTrace:
         assert hist_call[0][0] == EnterpriseTelemetryHistogram.TOOL_DURATION
         assert hist_call[0][1] == pytest.approx(1.5)
 
-    def test_error_increments_error_counter(self, trace_handler, mock_exporter):
+    def test_error_increments_error_counter(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._tool_trace(make_tool_info(error="Tool crashed"))
 
@@ -883,7 +892,7 @@ class TestToolTrace:
         ]
         assert len(error_calls) == 1
 
-    def test_inputs_and_outputs_gated_by_include_content(self, trace_handler, mock_exporter):
+    def test_inputs_and_outputs_gated_by_include_content(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = False
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._tool_trace(make_tool_info())
@@ -892,7 +901,7 @@ class TestToolTrace:
         assert attrs["dify.tool.inputs"].startswith("ref:message_id=")
         assert attrs["dify.tool.outputs"].startswith("ref:message_id=")
 
-    def test_inputs_present_when_include_content_true(self, trace_handler, mock_exporter):
+    def test_inputs_present_when_include_content_true(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = True
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._tool_trace(make_tool_info())
@@ -901,7 +910,7 @@ class TestToolTrace:
         assert attrs["dify.tool.inputs"] == json.dumps({"query": "test"})
         assert attrs["dify.tool.outputs"] == "search results"
 
-    def test_increments_requests_counter(self, trace_handler, mock_exporter):
+    def test_increments_requests_counter(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._tool_trace(make_tool_info())
 
@@ -918,27 +927,27 @@ class TestToolTrace:
 
 
 class TestModerationTrace:
-    def test_emits_event_with_correct_name(self, trace_handler, mock_exporter):
+    def test_emits_event_with_correct_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._moderation_trace(make_moderation_info())
 
         assert mock_emit.call_args[1]["event_name"] == EnterpriseTelemetryEvent.MODERATION_CHECK
 
-    def test_flagged_true_sets_attribute(self, trace_handler, mock_exporter):
+    def test_flagged_true_sets_attribute(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._moderation_trace(make_moderation_info(flagged=True))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.moderation.flagged"] is True
 
-    def test_flagged_false_sets_attribute(self, trace_handler, mock_exporter):
+    def test_flagged_false_sets_attribute(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._moderation_trace(make_moderation_info(flagged=False))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.moderation.flagged"] is False
 
-    def test_query_gated_by_include_content(self, trace_handler, mock_exporter):
+    def test_query_gated_by_include_content(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = False
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._moderation_trace(make_moderation_info())
@@ -946,7 +955,7 @@ class TestModerationTrace:
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.moderation.query"].startswith("ref:message_id=")
 
-    def test_query_present_when_include_content_true(self, trace_handler, mock_exporter):
+    def test_query_present_when_include_content_true(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = True
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._moderation_trace(make_moderation_info())
@@ -954,7 +963,7 @@ class TestModerationTrace:
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.moderation.query"] == "is this ok?"
 
-    def test_increments_requests_counter(self, trace_handler, mock_exporter):
+    def test_increments_requests_counter(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._moderation_trace(make_moderation_info())
 
@@ -971,48 +980,48 @@ class TestModerationTrace:
 
 
 class TestSuggestedQuestionTrace:
-    def test_emits_event_with_correct_name(self, trace_handler, mock_exporter):
+    def test_emits_event_with_correct_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._suggested_question_trace(make_suggested_question_info())
 
         assert mock_emit.call_args[1]["event_name"] == EnterpriseTelemetryEvent.SUGGESTED_QUESTION_GENERATION
 
-    def test_duration_computed_from_timestamps(self, trace_handler, mock_exporter):
+    def test_duration_computed_from_timestamps(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._suggested_question_trace(make_suggested_question_info())
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.suggested_question.duration"] == pytest.approx(5.0)
 
-    def test_duration_is_none_when_timestamps_missing(self, trace_handler, mock_exporter):
+    def test_duration_is_none_when_timestamps_missing(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._suggested_question_trace(make_suggested_question_info(start_time=None, end_time=None))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.suggested_question.duration"] is None
 
-    def test_status_is_failed_when_error_present(self, trace_handler, mock_exporter):
+    def test_status_is_failed_when_error_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._suggested_question_trace(make_suggested_question_info(error="Generation failed"))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.suggested_question.status"] == "failed"
 
-    def test_status_falls_back_to_succeeded_when_no_error(self, trace_handler, mock_exporter):
+    def test_status_falls_back_to_succeeded_when_no_error(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._suggested_question_trace(make_suggested_question_info(status=None, error=None))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.suggested_question.status"] == "succeeded"
 
-    def test_question_count_attribute(self, trace_handler, mock_exporter):
+    def test_question_count_attribute(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._suggested_question_trace(make_suggested_question_info())
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.suggested_question.count"] == 2
 
-    def test_questions_gated_by_include_content(self, trace_handler, mock_exporter):
+    def test_questions_gated_by_include_content(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = False
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._suggested_question_trace(make_suggested_question_info())
@@ -1020,7 +1029,7 @@ class TestSuggestedQuestionTrace:
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.suggested_question.questions"].startswith("ref:message_id=")
 
-    def test_increments_requests_counter(self, trace_handler, mock_exporter):
+    def test_increments_requests_counter(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._suggested_question_trace(make_suggested_question_info())
 
@@ -1037,48 +1046,48 @@ class TestSuggestedQuestionTrace:
 
 
 class TestDatasetRetrievalTrace:
-    def test_emits_event_with_correct_name(self, trace_handler, mock_exporter):
+    def test_emits_event_with_correct_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(make_dataset_retrieval_info())
 
         assert mock_emit.call_args[1]["event_name"] == EnterpriseTelemetryEvent.DATASET_RETRIEVAL
 
-    def test_document_count_attribute(self, trace_handler, mock_exporter):
+    def test_document_count_attribute(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(make_dataset_retrieval_info())
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.retrieval.document_count"] == 1
 
-    def test_dataset_ids_extracted(self, trace_handler, mock_exporter):
+    def test_dataset_ids_extracted(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(make_dataset_retrieval_info())
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert "ds-001" in attrs["dify.dataset.id"]
 
-    def test_empty_documents_has_zero_count(self, trace_handler, mock_exporter):
+    def test_empty_documents_has_zero_count(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(make_dataset_retrieval_info(documents=[]))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.retrieval.document_count"] == 0
 
-    def test_status_succeeded_when_no_error(self, trace_handler, mock_exporter):
+    def test_status_succeeded_when_no_error(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(make_dataset_retrieval_info())
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.retrieval.status"] == "succeeded"
 
-    def test_status_failed_when_error_present(self, trace_handler, mock_exporter):
+    def test_status_failed_when_error_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(make_dataset_retrieval_info(error="DB error"))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.retrieval.status"] == "failed"
 
-    def test_embedding_model_attributes_set_when_present(self, trace_handler, mock_exporter):
+    def test_embedding_model_attributes_set_when_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(make_dataset_retrieval_info())
 
@@ -1086,7 +1095,7 @@ class TestDatasetRetrievalTrace:
         assert "dify.dataset.embedding_providers" in attrs
         assert "dify.dataset.embedding_models" in attrs
 
-    def test_no_embedding_model_attributes_when_not_provided(self, trace_handler, mock_exporter):
+    def test_no_embedding_model_attributes_when_not_provided(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(
                 make_dataset_retrieval_info(metadata={"app_id": "app-001", "tenant_id": "tenant-abc"})
@@ -1096,7 +1105,7 @@ class TestDatasetRetrievalTrace:
         assert "dify.dataset.embedding_providers" not in attrs
         assert "dify.dataset.embedding_models" not in attrs
 
-    def test_rerank_attributes_set_when_present(self, trace_handler, mock_exporter):
+    def test_rerank_attributes_set_when_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(
                 make_dataset_retrieval_info(
@@ -1113,7 +1122,7 @@ class TestDatasetRetrievalTrace:
         assert attrs["dify.retrieval.rerank_provider"] == "cohere"
         assert attrs["dify.retrieval.rerank_model"] == "rerank-english"
 
-    def test_no_rerank_attributes_when_not_present(self, trace_handler, mock_exporter):
+    def test_no_rerank_attributes_when_not_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(
                 make_dataset_retrieval_info(metadata={"app_id": "app-001", "tenant_id": "tenant-abc"})
@@ -1123,7 +1132,7 @@ class TestDatasetRetrievalTrace:
         assert "dify.retrieval.rerank_provider" not in attrs
         assert "dify.retrieval.rerank_model" not in attrs
 
-    def test_dataset_retrieval_counter_incremented_per_dataset(self, trace_handler, mock_exporter):
+    def test_dataset_retrieval_counter_incremented_per_dataset(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._dataset_retrieval_trace(make_dataset_retrieval_info())
 
@@ -1135,7 +1144,7 @@ class TestDatasetRetrievalTrace:
         assert len(ds_calls) == 1
         assert ds_calls[0][0][2]["dataset_id"] == "ds-001"
 
-    def test_no_dataset_retrieval_counter_when_no_documents(self, trace_handler, mock_exporter):
+    def test_no_dataset_retrieval_counter_when_no_documents(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._dataset_retrieval_trace(make_dataset_retrieval_info(documents=[]))
 
@@ -1146,7 +1155,7 @@ class TestDatasetRetrievalTrace:
         ]
         assert len(ds_calls) == 0
 
-    def test_query_gated_by_include_content(self, trace_handler, mock_exporter):
+    def test_query_gated_by_include_content(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = False
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._dataset_retrieval_trace(make_dataset_retrieval_info())
@@ -1161,34 +1170,34 @@ class TestDatasetRetrievalTrace:
 
 
 class TestGenerateNameTrace:
-    def test_emits_event_with_correct_name(self, trace_handler, mock_exporter):
+    def test_emits_event_with_correct_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._generate_name_trace(make_generate_name_info())
 
         assert mock_emit.call_args[1]["event_name"] == EnterpriseTelemetryEvent.GENERATE_NAME_EXECUTION
 
-    def test_duration_computed_from_timestamps(self, trace_handler, mock_exporter):
+    def test_duration_computed_from_timestamps(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._generate_name_trace(make_generate_name_info())
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.generate_name.duration"] == pytest.approx(5.0)
 
-    def test_no_duration_when_timestamps_missing(self, trace_handler, mock_exporter):
+    def test_no_duration_when_timestamps_missing(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._generate_name_trace(make_generate_name_info(start_time=None, end_time=None))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.generate_name.duration"] is None
 
-    def test_status_succeeded_on_success(self, trace_handler, mock_exporter):
+    def test_status_succeeded_on_success(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._generate_name_trace(make_generate_name_info())
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.generate_name.status"] == "succeeded"
 
-    def test_status_failed_when_metadata_has_error(self, trace_handler, mock_exporter):
+    def test_status_failed_when_metadata_has_error(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._generate_name_trace(
                 make_generate_name_info(
@@ -1203,7 +1212,7 @@ class TestGenerateNameTrace:
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.generate_name.status"] == "failed"
 
-    def test_inputs_and_outputs_gated_by_include_content(self, trace_handler, mock_exporter):
+    def test_inputs_and_outputs_gated_by_include_content(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = False
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._generate_name_trace(make_generate_name_info())
@@ -1212,7 +1221,7 @@ class TestGenerateNameTrace:
         assert attrs["dify.generate_name.inputs"].startswith("ref:conversation_id=")
         assert attrs["dify.generate_name.outputs"].startswith("ref:conversation_id=")
 
-    def test_increments_requests_counter(self, trace_handler, mock_exporter):
+    def test_increments_requests_counter(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._generate_name_trace(make_generate_name_info())
 
@@ -1229,27 +1238,27 @@ class TestGenerateNameTrace:
 
 
 class TestPromptGenerationTrace:
-    def test_emits_event_with_correct_name(self, trace_handler, mock_exporter):
+    def test_emits_event_with_correct_name(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._prompt_generation_trace(make_prompt_generation_info())
 
         assert mock_emit.call_args[1]["event_name"] == EnterpriseTelemetryEvent.PROMPT_GENERATION_EXECUTION
 
-    def test_status_succeeded_on_success(self, trace_handler, mock_exporter):
+    def test_status_succeeded_on_success(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._prompt_generation_trace(make_prompt_generation_info())
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.prompt_generation.status"] == "succeeded"
 
-    def test_status_failed_when_error_present(self, trace_handler, mock_exporter):
+    def test_status_failed_when_error_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._prompt_generation_trace(make_prompt_generation_info(error="Generation error"))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.prompt_generation.status"] == "failed"
 
-    def test_token_counters_incremented(self, trace_handler, mock_exporter):
+    def test_token_counters_incremented(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._prompt_generation_trace(make_prompt_generation_info())
 
@@ -1258,7 +1267,7 @@ class TestPromptGenerationTrace:
         assert EnterpriseTelemetryCounter.INPUT_TOKENS in counter_names
         assert EnterpriseTelemetryCounter.OUTPUT_TOKENS in counter_names
 
-    def test_records_duration_histogram(self, trace_handler, mock_exporter):
+    def test_records_duration_histogram(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._prompt_generation_trace(make_prompt_generation_info())
 
@@ -1270,7 +1279,7 @@ class TestPromptGenerationTrace:
         assert len(hist_calls) == 1
         assert hist_calls[0][0][1] == pytest.approx(3.2)
 
-    def test_total_price_attribute_set_when_present(self, trace_handler, mock_exporter):
+    def test_total_price_attribute_set_when_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._prompt_generation_trace(make_prompt_generation_info(total_price=0.05, currency="USD"))
 
@@ -1278,14 +1287,14 @@ class TestPromptGenerationTrace:
         assert attrs["dify.prompt_generation.total_price"] == pytest.approx(0.05)
         assert attrs["dify.prompt_generation.currency"] == "USD"
 
-    def test_no_total_price_attribute_when_none(self, trace_handler, mock_exporter):
+    def test_no_total_price_attribute_when_none(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._prompt_generation_trace(make_prompt_generation_info(total_price=None))
 
         attrs = mock_emit.call_args[1]["attributes"]
         assert "dify.prompt_generation.total_price" not in attrs
 
-    def test_error_increments_error_counter(self, trace_handler, mock_exporter):
+    def test_error_increments_error_counter(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._prompt_generation_trace(make_prompt_generation_info(error="Prompt failed"))
 
@@ -1294,7 +1303,7 @@ class TestPromptGenerationTrace:
         ]
         assert len(error_calls) == 1
 
-    def test_no_error_counter_on_success(self, trace_handler, mock_exporter):
+    def test_no_error_counter_on_success(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._prompt_generation_trace(make_prompt_generation_info())
 
@@ -1303,7 +1312,7 @@ class TestPromptGenerationTrace:
         ]
         assert len(error_calls) == 0
 
-    def test_instruction_gated_by_include_content(self, trace_handler, mock_exporter):
+    def test_instruction_gated_by_include_content(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         mock_exporter.include_content = False
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._prompt_generation_trace(make_prompt_generation_info())
@@ -1311,7 +1320,7 @@ class TestPromptGenerationTrace:
         attrs = mock_emit.call_args[1]["attributes"]
         assert attrs["dify.prompt_generation.instruction"].startswith("ref:trace_id=")
 
-    def test_operation_type_label_used_in_token_counters(self, trace_handler, mock_exporter):
+    def test_operation_type_label_used_in_token_counters(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._prompt_generation_trace(make_prompt_generation_info(operation_type="code_generate"))
 
@@ -1321,7 +1330,7 @@ class TestPromptGenerationTrace:
         assert len(token_calls) == 1
         assert token_calls[0][0][2]["operation_type"] == "code_generate"
 
-    def test_emits_correct_tenant_id(self, trace_handler, mock_exporter):
+    def test_emits_correct_tenant_id(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._prompt_generation_trace(make_prompt_generation_info())
 

api/tests/unit_tests/libs/test_oauth_bearer_token_expiry.py

@@ -0,0 +1,87 @@
+"""Resolver-level expiry signalling.
+
+An expired token must be distinguishable from an unknown/revoked one: the
+resolver raises ``TokenExpiredError`` for expiry and returns ``None`` for
+everything else. The signal survives the negative-cache window via a distinct
+``expired`` marker so a retry inside ``NEGATIVE_TTL`` still reports expiry.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from unittest.mock import MagicMock
+
+import pytest
+
+from libs.oauth_bearer import (
+    OAuthAccessTokenResolver,
+    TokenExpiredError,
+)
+
+
+def _row(expires_at: datetime):
+    row = MagicMock()
+    row.id = "11111111-1111-1111-1111-111111111111"
+    row.account_id = "22222222-2222-2222-2222-222222222222"
+    row.prefix = "dfoa_"
+    row.subject_email = None
+    row.subject_issuer = None
+    row.client_id = None
+    row.expires_at = expires_at
+    return row
+
+
+def _resolver(redis: MagicMock, db_row=None) -> OAuthAccessTokenResolver:
+    session = MagicMock()
+    session.query.return_value.filter.return_value.one_or_none.return_value = db_row
+    session.execute.return_value.rowcount = 1
+    return OAuthAccessTokenResolver(session_factory=lambda: session, redis_client=redis)
+
+
+def test_resolve_raises_token_expired_for_expired_db_row():
+    redis = MagicMock()
+    redis.get.return_value = None  # cache miss -> DB path
+    past = datetime.now(UTC) - timedelta(minutes=1)
+    resolver = _resolver(redis, db_row=_row(past))
+
+    with pytest.raises(TokenExpiredError):
+        resolver.for_account().resolve("expiredhash")
+
+
+def test_resolve_raises_token_expired_for_expired_cache_marker():
+    redis = MagicMock()
+    redis.get.return_value = b"expired"  # negative-cache replay
+    resolver = _resolver(redis, db_row=None)
+
+    with pytest.raises(TokenExpiredError):
+        resolver.for_account().resolve("expiredhash")
+
+
+def test_resolve_returns_none_for_invalid_cache_marker():
+    redis = MagicMock()
+    redis.get.return_value = b"invalid"
+    resolver = _resolver(redis, db_row=None)
+
+    assert resolver.for_account().resolve("revokedhash") is None
+
+
+def test_resolve_returns_none_for_unknown_token():
+    redis = MagicMock()
+    redis.get.return_value = None  # cache miss
+    resolver = _resolver(redis, db_row=None)  # no DB row
+
+    assert resolver.for_account().resolve("unknownhash") is None
+
+
+def test_hard_expire_caches_expired_marker_not_invalid():
+    redis = MagicMock()
+    redis.get.return_value = None
+    past = datetime.now(UTC) - timedelta(minutes=1)
+    resolver = _resolver(redis, db_row=_row(past))
+
+    with pytest.raises(TokenExpiredError):
+        resolver.for_account().resolve("expiredhash")
+
+    setex_values = [call.args[2] for call in redis.setex.call_args_list]
+    assert "expired" in setex_values
+    assert "invalid" not in setex_values

api/tests/unit_tests/libs/test_pyrefly_diagnostics.py

@@ -1,4 +1,4 @@
-from libs.pyrefly_diagnostics import extract_diagnostics
+from libs.pyrefly_diagnostics import extract_diagnostics, render_diagnostics
 
 
 def test_extract_diagnostics_keeps_only_summary_and_location_lines() -> None:
@@ -40,6 +40,37 @@ def test_extract_diagnostics_handles_error_without_location_line() -> None:
     assert diagnostics == "ERROR unexpected pyrefly output format [bad-format]\n"
 
 
+def test_extract_diagnostics_keeps_warn_headlines_and_location_lines() -> None:
+    # Arrange
+    raw_output = """INFO Checking project configured at `/tmp/project/pyrefly.toml`
+WARN Skipping include pattern `/tmp/project/tests` because it is matched by `project-excludes`.
+   --> tests/test_containers_integration_tests/pyrefly.toml:3:1
+"""
+
+    # Act
+    diagnostics = extract_diagnostics(raw_output)
+
+    # Assert
+    assert diagnostics == (
+        "WARN Skipping include pattern `/tmp/project/tests` because it is matched by `project-excludes`.\n"
+        "   --> tests/test_containers_integration_tests/pyrefly.toml:3:1\n"
+    )
+
+
+def test_render_diagnostics_falls_back_to_raw_output_for_nonzero_exit_without_matches() -> None:
+    # Arrange
+    raw_output = (
+        "INFO Checking project configured at `/tmp/project/pyrefly.toml`\n"
+        "No Python files matched pattern `/tmp/project/tests/test_containers_integration_tests`\n"
+    )
+
+    # Act
+    diagnostics = render_diagnostics(raw_output, exit_code=1)
+
+    # Assert
+    assert diagnostics == raw_output
+
+
 def test_extract_diagnostics_returns_empty_for_non_error_output() -> None:
     # Arrange
     raw_output = "INFO Checking project configured at `/tmp/project/pyrefly.toml`\n"

api/tests/unit_tests/models/test_account.py

@@ -1,14 +0,0 @@
-from models.account import TenantAccountRole
-
-
-def test_account_is_privileged_role():
-    assert TenantAccountRole.ADMIN == "admin"
-    assert TenantAccountRole.OWNER == "owner"
-    assert TenantAccountRole.EDITOR == "editor"
-    assert TenantAccountRole.NORMAL == "normal"
-
-    assert TenantAccountRole.is_privileged_role(TenantAccountRole.ADMIN)
-    assert TenantAccountRole.is_privileged_role(TenantAccountRole.OWNER)
-    assert not TenantAccountRole.is_privileged_role(TenantAccountRole.NORMAL)
-    assert not TenantAccountRole.is_privileged_role(TenantAccountRole.EDITOR)
-    assert not TenantAccountRole.is_privileged_role("")

api/tests/unit_tests/models/test_dataset_models.py

@@ -12,7 +12,6 @@ This test suite covers:
 import json
 import pickle
 from datetime import UTC, datetime
-from types import SimpleNamespace
 from unittest.mock import Mock, patch
 from urllib.parse import parse_qs, urlparse
 from uuid import uuid4
@@ -20,6 +19,7 @@ from uuid import uuid4
 import pytest
 
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
+from extensions.storage.storage_type import StorageType
 from models.dataset import (
     AppDatasetJoin,
     ChildChunk,
@@ -32,12 +32,14 @@ from models.dataset import (
     ExternalKnowledgeBindings,
 )
 from models.enums import (
+    CreatorUserRole,
     DataSourceType,
     DocumentCreatedFrom,
     IndexingStatus,
     ProcessRuleMode,
     SegmentStatus,
 )
+from models.model import UploadFile
 
 
 class TestDatasetModelValidation:
@@ -719,13 +721,20 @@ class TestDocumentSegmentIndexing:
             created_by="user-1",
         )
         segment.id = "segment-1"
-        attachment = SimpleNamespace(
-            id="upload-1",
+        attachment = UploadFile(
+            tenant_id="tenant-1",
+            storage_type=StorageType.LOCAL,
+            key="upload-1-key",
             name="image.png",
             size=128,
             extension="png",
             mime_type="image/png",
+            created_by_role=CreatorUserRole.ACCOUNT,
+            created_by="user-1",
+            created_at=datetime(2023, 11, 14, tzinfo=UTC),
+            used=False,
         )
+        attachment.id = "upload-1"
 
         monkeypatch.setattr("models.dataset.time.time", lambda: 1700000000)
         monkeypatch.setattr("models.dataset.os.urandom", lambda _: b"\x01" * 16)

api/tests/unit_tests/services/enterprise/test_rbac_service.py

@@ -86,21 +86,26 @@ class TestRoles:
         call = _call_args(mock_send)
         assert call.method == "GET"
         assert call.endpoint == "/rbac/roles"
-        assert call.params == {"page_number": 2, "results_per_page": 50, "reverse": "true"}
+        assert call.params == {
+            "dataset_operator_enabled": False,
+            "page_number": 2,
+            "results_per_page": 50,
+            "reverse": "true",
+        }
         assert out.pagination
         assert out.pagination.total_count == 1
 
     def test_list_omits_params_when_default(self, mock_send: MagicMock):
         mock_send.return_value = {"data": [], "pagination": None}
         svc.RBACService.Roles.list("tenant-1")
-        assert _call_args(mock_send).params is None
+        assert _call_args(mock_send).params is not None
 
     def test_list_forwards_include_owner(self, mock_send: MagicMock):
         mock_send.return_value = {"data": [], "pagination": None}
 
         svc.RBACService.Roles.list("tenant-1", include_owner=1)
 
-        assert _call_args(mock_send).params == {"include_owner": 1}
+        assert _call_args(mock_send).params == {"dataset_operator_enabled": False, "include_owner": 1}
 
     def test_list_coerces_null_permission_keys(self, mock_send: MagicMock):
         mock_send.return_value = {
@@ -616,6 +621,7 @@ class TestMyPermissions:
 
         mock_send.assert_not_called()
         assert out.workspace.permission_keys == workspace_keys
+        assert len(out.workspace.permission_keys) == len(set(out.workspace.permission_keys))
         assert out.app.default_permission_keys == app_keys
         assert out.dataset.default_permission_keys == dataset_keys
         assert out.app.overrides == []

api/tests/unit_tests/services/plugin/test_plugin_auto_upgrade_service.py

@@ -1,6 +1,9 @@
+import logging
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 from models.account import TenantPluginAutoUpgradeStrategy
 
 MODULE = "services.plugin.plugin_auto_upgrade_service"
@@ -227,7 +230,7 @@ class TestBackfillStrategyCategories:
         assert default_time % (15 * 60) == 0
         assert 0 <= default_time < 24 * 60 * 60
 
-    def test_creates_missing_categories_and_splits_known_plugins(self):
+    def test_creates_missing_categories_and_splits_known_plugins(self, caplog: pytest.LogCaptureFixture):
         p1, session = _patched_session()
         tool_strategy = SimpleNamespace(
             category=TenantPluginAutoUpgradeStrategy.PluginCategory.TOOL,
@@ -260,7 +263,11 @@ class TestBackfillStrategyCategories:
         installer = MagicMock()
         installer.list_plugins.return_value = installed_plugins
 
-        with p1, patch(f"{MODULE}.PluginInstaller", return_value=installer), patch(f"{MODULE}.logger") as logger:
+        with (
+            p1,
+            patch(f"{MODULE}.PluginInstaller", return_value=installer),
+            caplog.at_level(logging.WARNING, logger=MODULE),
+        ):
             from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService
 
             result = PluginAutoUpgradeService.backfill_strategy_categories("t1")
@@ -272,10 +279,7 @@ class TestBackfillStrategyCategories:
         assert tool_strategy.include_plugins == ["tool-plugin"]
         assert model_strategy.exclude_plugins == ["model-plugin"]
         assert model_strategy.include_plugins == ["model-plugin"]
-        logger.warning.assert_called_once_with(
+        assert (
             "Skipped unknown plugin IDs while backfilling plugin auto-upgrade strategies: "
-            "tenant_id=%s, field=%s, plugin_ids=%s",
-            "t1",
-            "exclude_plugins",
-            ["unknown-plugin"],
+            "tenant_id=t1, field=exclude_plugins, plugin_ids=['unknown-plugin']" in caplog.messages
         )

api/tests/unit_tests/services/test_clear_free_plan_tenant_expired_logs.py

@@ -1,4 +1,5 @@
 import datetime
+import logging
 from types import SimpleNamespace
 from unittest.mock import MagicMock, Mock, patch
 
@@ -347,7 +348,9 @@ def test_serialize_record_falls_back_to_table_columns() -> None:
     }
 
 
-def test_process_with_tenant_ids_filters_by_plan_and_logs_errors(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_process_with_tenant_ids_filters_by_plan_and_logs_errors(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
     monkeypatch.setattr(service_module, "db", SimpleNamespace(engine=object()))
 
     # Total tenant count query
@@ -381,14 +384,13 @@ def test_process_with_tenant_ids_filters_by_plan_and_logs_errors(monkeypatch: py
     process_tenant_mock = MagicMock(side_effect=lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("err")))
     monkeypatch.setattr(ClearFreePlanTenantExpiredLogs, "process_tenant", process_tenant_mock)
 
-    logger_exc = MagicMock()
-    monkeypatch.setattr(service_module.logger, "exception", logger_exc)
-
-    ClearFreePlanTenantExpiredLogs.process(days=7, batch=10, tenant_ids=["t_sandbox", "t_paid", "t_fail"])
+    with caplog.at_level(logging.ERROR, logger=service_module.logger.name):
+        ClearFreePlanTenantExpiredLogs.process(days=7, batch=10, tenant_ids=["t_sandbox", "t_paid", "t_fail"])
 
     # Only sandbox tenant should attempt processing, and its failure should be swallowed + logged.
     assert process_tenant_mock.call_count == 1
-    assert logger_exc.call_count >= 1
+    assert "Failed to process tenant t_sandbox" in caplog.messages
+    assert "Failed to process tenant t_fail" in caplog.messages
 
 
 def test_process_without_tenant_ids_batches_and_scales_interval(monkeypatch: pytest.MonkeyPatch) -> None:

api/tests/unit_tests/services/test_workflow_collaboration_service.py

@@ -1,3 +1,4 @@
+import logging
 from unittest.mock import Mock, patch
 
 import pytest
@@ -427,16 +428,20 @@ class TestWorkflowCollaborationService:
         repository.delete_leader.assert_not_called()
 
     def test_broadcast_leader_change_logs_emit_errors(
-        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+        self,
+        service: tuple[WorkflowCollaborationService, Mock, Mock],
+        caplog: pytest.LogCaptureFixture,
     ) -> None:
         collaboration_service, repository, socketio = service
         repository.get_session_sids.return_value = ["sid-1", "sid-2"]
         socketio.emit.side_effect = [RuntimeError("boom"), None]
 
-        with patch("services.workflow_collaboration_service.logging.exception") as exception_mock:
+        with caplog.at_level(logging.ERROR):
             collaboration_service.broadcast_leader_change("wf-1", "sid-2")
 
-        assert exception_mock.call_count == 1
+        error_records = [record for record in caplog.records if record.levelno == logging.ERROR]
+        assert len(error_records) == 1
+        assert "Failed to emit leader status to session sid-1" in error_records[0].getMessage()
 
     def test_broadcast_online_users_sorts_and_emits(
         self, service: tuple[WorkflowCollaborationService, Mock, Mock]

api/tests/unit_tests/tasks/test_dataset_indexing_task.py

@@ -9,6 +9,7 @@ This module tests the document indexing task functionality including:
 - Task cancellation and cleanup
 """
 
+import logging
 import uuid
 from contextlib import nullcontext
 from types import SimpleNamespace
@@ -758,7 +759,15 @@ class TestErrorHandling:
         assert mock_db_session.close.called
 
     def test_tenant_queue_error_handling_still_processes_next_task(
-        self, tenant_id, dataset_id, document_ids, mock_redis, mock_db_session, mock_dataset, mock_indexing_runner
+        self,
+        tenant_id,
+        dataset_id,
+        document_ids,
+        mock_redis,
+        mock_db_session,
+        mock_dataset,
+        mock_indexing_runner,
+        caplog: pytest.LogCaptureFixture,
     ):
         """
         Test that errors don't prevent processing next task in tenant queue.
@@ -778,14 +787,17 @@ class TestErrorHandling:
         with patch("tasks.document_indexing_task._document_indexing") as mock_indexing:
             mock_indexing.side_effect = Exception("Processing failed")
 
-            # Patch logger to avoid format string issue in actual code
-            with patch("tasks.document_indexing_task.logger"):
+            with caplog.at_level(logging.ERROR, logger="tasks.document_indexing_task"):
                 with patch("tasks.document_indexing_task.normal_document_indexing_task") as mock_task:
                     # Act
                     _document_indexing_with_tenant_queue(tenant_id, dataset_id, document_ids, mock_task)
 
                     # Assert - Next task should still be enqueued despite error
                     mock_task.apply_async.assert_called()
+                    assert (
+                        f"Error processing document indexing {dataset_id} for tenant {tenant_id}: {document_ids}"
+                        in caplog.messages
+                    )
 
     def test_concurrent_task_limit_respected(
         self, tenant_id, dataset_id, document_ids, mock_redis, mock_db_session, mock_dataset

api/tests/unit_tests/tasks/test_workflow_execute_task.py

@@ -1,18 +1,26 @@
 from __future__ import annotations
 
 import json
+import logging
 import uuid
+from contextlib import nullcontext
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
 import pytest
+from pydantic import BaseModel
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom, WorkflowAppGenerateEntity
+from graphon.entities import WorkflowStartReason
+from graphon.enums import WorkflowExecutionStatus
 from models.enums import CreatorUserRole
 from models.model import App, AppMode, Conversation
 from models.workflow import Workflow, WorkflowRun
 from repositories.sqlalchemy_api_workflow_run_repository import _WorkflowRunError
+from tasks.app_generate import workflow_execute_task as workflow_execute_task_module
 from tasks.app_generate.workflow_execute_task import (
+    AppExecutionParams,
+    _AppRunner,
     _publish_streaming_response,
     _resume_advanced_chat,
     _resume_app_execution,
@@ -31,6 +39,11 @@ class _FakeSessionContext:
         return False
 
 
+class _StreamEventModel(BaseModel):
+    event: object | None = None
+    task_id: object | None = None
+
+
 def _build_advanced_chat_generate_entity(conversation_id: str | None) -> AdvancedChatAppGenerateEntity:
     return AdvancedChatAppGenerateEntity(
         task_id="task-id",
@@ -60,6 +73,46 @@ def _single_event_generator(payload):
     yield payload
 
 
+def _decode_published_payload(payload: bytes) -> dict[str, object] | str:
+    return json.loads(payload.decode())
+
+
+def _published_payloads(topic: MagicMock) -> list[dict[str, object] | str]:
+    return [_decode_published_payload(call.args[0]) for call in topic.publish.call_args_list]
+
+
+@pytest.mark.parametrize(
+    ("event", "expected"),
+    [
+        ({"event": "workflow_started"}, "workflow_started"),
+        ({"event": 123}, "123"),
+        (_StreamEventModel(event="workflow_started"), "workflow_started"),
+        (_StreamEventModel(event=123), "123"),
+        ({}, None),
+        (_StreamEventModel(), None),
+        ("workflow_started", None),
+    ],
+)
+def test_get_event_name(event: object, expected: str | None):
+    assert workflow_execute_task_module._get_event_name(event) == expected
+
+
+@pytest.mark.parametrize(
+    ("event", "expected"),
+    [
+        ({"task_id": "task-id"}, "task-id"),
+        (_StreamEventModel(task_id="task-id"), "task-id"),
+        ({"task_id": 123}, None),
+        (_StreamEventModel(task_id=123), None),
+        ({"task_id": ""}, None),
+        (_StreamEventModel(), None),
+        ("task-id", None),
+    ],
+)
+def test_get_task_id(event: object, expected: str | None):
+    assert workflow_execute_task_module._get_task_id(event) == expected
+
+
 @pytest.fixture
 def mock_topic(monkeypatch: pytest.MonkeyPatch) -> MagicMock:
     topic = MagicMock()
@@ -72,21 +125,413 @@ def mock_topic(monkeypatch: pytest.MonkeyPatch) -> MagicMock:
 
 def test_publish_streaming_response_with_uuid(mock_topic: MagicMock):
     workflow_run_id = uuid.uuid4()
-    response_stream = iter([{"event": "foo"}, "ping"])
+    response_stream = iter(
+        [
+            {"event": "workflow_started", "task_id": "task-id"},
+            {"event": "workflow_finished", "task_id": "task-id", "data": {"status": "succeeded"}},
+        ]
+    )
 
-    _publish_streaming_response(response_stream, workflow_run_id, app_mode=AppMode.ADVANCED_CHAT)
+    _publish_streaming_response(
+        response_stream,
+        workflow_run_id,
+        app_mode=AppMode.ADVANCED_CHAT,
+        workflow_id="workflow-id",
+        inputs={},
+        started_reason=WorkflowStartReason.INITIAL,
+    )
 
-    payloads = [call.args[0] for call in mock_topic.publish.call_args_list]
-    assert payloads == [json.dumps({"event": "foo"}).encode(), json.dumps("ping").encode()]
+    payloads = _published_payloads(mock_topic)
+    assert [payload["event"] for payload in payloads] == ["workflow_started", "workflow_finished"]
 
 
 def test_publish_streaming_response_coerces_string_uuid(mock_topic: MagicMock):
     workflow_run_id = uuid.uuid4()
-    response_stream = iter([{"event": "bar"}])
+    response_stream = iter([{"event": "workflow_paused", "task_id": "task-id"}])
 
-    _publish_streaming_response(response_stream, str(workflow_run_id), app_mode=AppMode.ADVANCED_CHAT)
+    _publish_streaming_response(
+        response_stream,
+        str(workflow_run_id),
+        app_mode=AppMode.ADVANCED_CHAT,
+        workflow_id="workflow-id",
+        inputs={},
+        started_reason=WorkflowStartReason.INITIAL,
+    )
 
-    mock_topic.publish.assert_called_once_with(json.dumps({"event": "bar"}).encode())
+    payloads = _published_payloads(mock_topic)
+    assert [payload["event"] for payload in payloads] == ["workflow_paused"]
+
+
+def test_publish_streaming_response_publishes_started_then_failed_terminal_when_iteration_raises(
+    mock_topic: MagicMock,
+):
+    def _response_stream():
+        if False:
+            yield None
+        raise RuntimeError("stream exploded")
+
+    with pytest.raises(RuntimeError, match="stream exploded"):
+        _publish_streaming_response(
+            _response_stream(),
+            "workflow-run-id",
+            app_mode=AppMode.ADVANCED_CHAT,
+            workflow_id="workflow-id",
+            inputs={"foo": "bar"},
+            started_reason=WorkflowStartReason.INITIAL,
+        )
+
+    payloads = _published_payloads(mock_topic)
+    assert [payload["event"] for payload in payloads] == ["workflow_started", "workflow_finished"]
+    assert payloads[0]["data"]["workflow_id"] == "workflow-id"
+    assert payloads[0]["data"]["inputs"] == {"foo": "bar"}
+    assert payloads[1]["data"]["status"] == WorkflowExecutionStatus.FAILED
+    assert payloads[1]["data"]["error"] == "stream exploded"
+
+
+def test_publish_streaming_response_recovers_when_workflow_started_publish_fails_first(
+    mock_topic: MagicMock,
+    caplog: pytest.LogCaptureFixture,
+):
+    caplog.set_level(logging.ERROR, logger="tasks.app_generate.workflow_execute_task")
+    response_stream = iter([{"event": "workflow_started", "task_id": "task-id"}])
+    successful_payloads: list[dict[str, object] | str] = []
+    started_publish_attempts = 0
+
+    def _publish(payload: bytes) -> None:
+        nonlocal started_publish_attempts
+
+        decoded = _decode_published_payload(payload)
+        if isinstance(decoded, dict) and decoded.get("event") == "workflow_started":
+            started_publish_attempts += 1
+            if started_publish_attempts == 1:
+                raise RuntimeError("started publish failed")
+        successful_payloads.append(decoded)
+
+    mock_topic.publish.side_effect = _publish
+
+    with pytest.raises(RuntimeError, match="started publish failed"):
+        _publish_streaming_response(
+            response_stream,
+            "workflow-run-id",
+            app_mode=AppMode.ADVANCED_CHAT,
+            workflow_id="workflow-id",
+            inputs={"file": object()},
+            started_reason=WorkflowStartReason.INITIAL,
+        )
+
+    assert [payload["event"] for payload in successful_payloads] == ["workflow_started", "workflow_finished"]
+    assert successful_payloads[0]["task_id"] == "task-id"
+    assert isinstance(successful_payloads[0]["data"]["inputs"]["file"], str)
+    assert successful_payloads[1]["task_id"] == "task-id"
+    assert successful_payloads[1]["data"]["status"] == WorkflowExecutionStatus.FAILED
+    assert successful_payloads[1]["data"]["error"] == "started publish failed"
+    assert "workflow-run-id" in caplog.text
+    assert "publishing fallback terminal event" in caplog.text
+
+
+def test_publish_streaming_response_publishes_failed_terminal_without_duplicate_started_on_publish_error(
+    mock_topic: MagicMock,
+    caplog: pytest.LogCaptureFixture,
+):
+    caplog.set_level(logging.ERROR, logger="tasks.app_generate.workflow_execute_task")
+    response_stream = iter(
+        [
+            {
+                "event": "workflow_started",
+                "task_id": "task-id",
+                "workflow_run_id": "workflow-run-id",
+                "data": {"id": "workflow-run-id", "workflow_id": "workflow-id", "inputs": {}, "created_at": 1},
+            },
+            {"event": "node_started", "task_id": "task-id"},
+        ]
+    )
+    successful_payloads: list[dict[str, object] | str] = []
+
+    def _publish(payload: bytes) -> None:
+        decoded = _decode_published_payload(payload)
+        if isinstance(decoded, dict) and decoded.get("event") == "node_started":
+            raise RuntimeError("broker write failed")
+        successful_payloads.append(decoded)
+
+    mock_topic.publish.side_effect = _publish
+
+    with pytest.raises(RuntimeError, match="broker write failed"):
+        _publish_streaming_response(
+            response_stream,
+            "workflow-run-id",
+            app_mode=AppMode.ADVANCED_CHAT,
+            workflow_id="workflow-id",
+            inputs={},
+            started_reason=WorkflowStartReason.INITIAL,
+        )
+
+    assert [payload["event"] for payload in successful_payloads] == ["workflow_started", "workflow_finished"]
+    assert successful_payloads[1]["task_id"] == "task-id"
+    assert successful_payloads[1]["data"]["status"] == WorkflowExecutionStatus.FAILED
+    assert successful_payloads[1]["data"]["error"] == "broker write failed"
+    assert "workflow-run-id" in caplog.text
+    assert "publishing fallback terminal event" in caplog.text
+
+
+def test_publish_streaming_response_recovers_when_workflow_finished_publish_fails_first(
+    mock_topic: MagicMock,
+    caplog: pytest.LogCaptureFixture,
+):
+    caplog.set_level(logging.ERROR, logger="tasks.app_generate.workflow_execute_task")
+    response_stream = iter(
+        [
+            {"event": "workflow_started", "task_id": "task-id"},
+            {"event": "workflow_finished", "task_id": "task-id", "data": {"status": "succeeded"}},
+        ]
+    )
+    successful_payloads: list[dict[str, object] | str] = []
+    finished_publish_attempts = 0
+
+    def _publish(payload: bytes) -> None:
+        nonlocal finished_publish_attempts
+
+        decoded = _decode_published_payload(payload)
+        if isinstance(decoded, dict) and decoded.get("event") == "workflow_finished":
+            finished_publish_attempts += 1
+            if finished_publish_attempts == 1:
+                raise RuntimeError("finished publish failed")
+        successful_payloads.append(decoded)
+
+    mock_topic.publish.side_effect = _publish
+
+    with pytest.raises(RuntimeError, match="finished publish failed"):
+        _publish_streaming_response(
+            response_stream,
+            "workflow-run-id",
+            app_mode=AppMode.ADVANCED_CHAT,
+            workflow_id="workflow-id",
+            inputs={},
+            started_reason=WorkflowStartReason.INITIAL,
+        )
+
+    assert [payload["event"] for payload in successful_payloads] == ["workflow_started", "workflow_finished"]
+    assert successful_payloads[1]["task_id"] == "task-id"
+    assert successful_payloads[1]["data"]["status"] == WorkflowExecutionStatus.FAILED
+    assert successful_payloads[1]["data"]["error"] == "finished publish failed"
+    assert "workflow-run-id" in caplog.text
+    assert "publishing fallback terminal event" in caplog.text
+
+
+def test_publish_streaming_response_publishes_failed_terminal_on_exhaustion_without_terminal_event(
+    mock_topic: MagicMock,
+    caplog: pytest.LogCaptureFixture,
+):
+    caplog.set_level(logging.WARNING, logger="tasks.app_generate.workflow_execute_task")
+    response_stream = iter(
+        [
+            {
+                "event": "workflow_started",
+                "task_id": "task-id",
+                "workflow_run_id": "workflow-run-id",
+                "data": {"id": "workflow-run-id", "workflow_id": "workflow-id", "inputs": {}, "created_at": 1},
+            }
+        ]
+    )
+
+    _publish_streaming_response(
+        response_stream,
+        "workflow-run-id",
+        app_mode=AppMode.ADVANCED_CHAT,
+        workflow_id="workflow-id",
+        inputs={},
+        started_reason=WorkflowStartReason.INITIAL,
+    )
+
+    payloads = _published_payloads(mock_topic)
+    assert [payload["event"] for payload in payloads] == ["workflow_started", "workflow_finished"]
+    assert payloads[1]["task_id"] == "task-id"
+    assert payloads[1]["data"]["status"] == WorkflowExecutionStatus.FAILED
+    assert payloads[1]["data"]["error"] == "Workflow stream ended without a terminal event"
+    assert "workflow-run-id" in caplog.text
+    assert "ended without a terminal event" in caplog.text
+
+
+def test_publish_streaming_response_does_not_publish_synthetic_failure_after_terminal_event(mock_topic: MagicMock):
+    response_stream = iter(
+        [
+            {
+                "event": "workflow_started",
+                "task_id": "task-id",
+                "workflow_run_id": "workflow-run-id",
+                "data": {"id": "workflow-run-id", "workflow_id": "workflow-id", "inputs": {}, "created_at": 1},
+            },
+            {
+                "event": "workflow_finished",
+                "task_id": "task-id",
+                "workflow_run_id": "workflow-run-id",
+                "data": {
+                    "id": "workflow-run-id",
+                    "workflow_id": "workflow-id",
+                    "status": WorkflowExecutionStatus.SUCCEEDED,
+                    "outputs": {},
+                    "error": None,
+                    "elapsed_time": 0.1,
+                    "total_tokens": 1,
+                    "total_steps": 1,
+                    "created_by": {},
+                    "created_at": 1,
+                    "finished_at": 2,
+                    "exceptions_count": 0,
+                    "files": [],
+                },
+            },
+        ]
+    )
+
+    _publish_streaming_response(
+        response_stream,
+        "workflow-run-id",
+        app_mode=AppMode.ADVANCED_CHAT,
+        workflow_id="workflow-id",
+        inputs={},
+        started_reason=WorkflowStartReason.INITIAL,
+    )
+
+    payloads = _published_payloads(mock_topic)
+    assert [payload["event"] for payload in payloads] == ["workflow_started", "workflow_finished"]
+
+
+def test_app_runner_streaming_failure_publishes_started_then_failed_workflow_finished(
+    mock_topic: MagicMock, monkeypatch
+):
+    exec_params = AppExecutionParams(
+        app_id="app-id",
+        workflow_id="workflow-id",
+        tenant_id="tenant-id",
+        app_mode=AppMode.ADVANCED_CHAT,
+        user={"TYPE": "account", "user_id": "user-id"},
+        args={"inputs": {}, "query": "test"},
+        invoke_from=InvokeFrom.EXPLORE,
+        streaming=True,
+        workflow_run_id="workflow-run-id",
+    )
+    runner = _AppRunner(session_factory=MagicMock(), exec_params=exec_params)
+
+    workflow = SimpleNamespace(id="workflow-id", app_id="app-id", created_by="workflow-owner")
+    app = SimpleNamespace(id="app-id")
+    fake_session = MagicMock()
+    fake_session.get.side_effect = [workflow, app]
+
+    monkeypatch.setattr(runner, "_session", lambda: nullcontext(fake_session))
+    monkeypatch.setattr(runner, "_resolve_user", lambda: MagicMock())
+    monkeypatch.setattr(runner, "_setup_flask_context", lambda _user: nullcontext())
+    monkeypatch.setattr(runner, "_run_app", lambda **_kwargs: (_ for _ in ()).throw(ValueError("Invalid upload file")))
+
+    with pytest.raises(ValueError, match="Invalid upload file"):
+        runner.run()
+
+    assert mock_topic.publish.call_count == 2
+    started_payload = json.loads(mock_topic.publish.call_args_list[0].args[0].decode())
+    assert started_payload["event"] == "workflow_started"
+    assert started_payload["workflow_run_id"] == "workflow-run-id"
+    assert started_payload["task_id"] == "workflow-run-id"
+    assert started_payload["data"]["id"] == "workflow-run-id"
+    assert started_payload["data"]["workflow_id"] == "workflow-id"
+    assert started_payload["data"]["reason"] == "initial"
+
+    finished_payload = json.loads(mock_topic.publish.call_args_list[1].args[0].decode())
+    assert finished_payload["event"] == "workflow_finished"
+    assert finished_payload["workflow_run_id"] == "workflow-run-id"
+    assert finished_payload["task_id"] == "workflow-run-id"
+    assert finished_payload["data"]["id"] == "workflow-run-id"
+    assert finished_payload["data"]["workflow_id"] == "workflow-id"
+    assert finished_payload["data"]["status"] == WorkflowExecutionStatus.FAILED
+    assert finished_payload["data"]["error"] == "Invalid upload file"
+    assert finished_payload["data"]["outputs"] is None
+    assert finished_payload["data"]["total_tokens"] == 0
+    assert finished_payload["data"]["total_steps"] == 0
+    assert finished_payload["data"]["exceptions_count"] == 1
+    assert finished_payload["data"]["created_by"] == {}
+    assert finished_payload["data"]["created_at"] == finished_payload["data"]["finished_at"]
+    assert finished_payload["data"]["files"] == []
+
+
+def test_app_runner_streaming_failure_keeps_existing_pre_runtime_helper_behavior(
+    mock_topic: MagicMock,
+    monkeypatch: pytest.MonkeyPatch,
+):
+    exec_params = AppExecutionParams(
+        app_id="app-id",
+        workflow_id="workflow-id",
+        tenant_id="tenant-id",
+        app_mode=AppMode.ADVANCED_CHAT,
+        user={"TYPE": "account", "user_id": "user-id"},
+        args={"inputs": {}, "query": "test"},
+        invoke_from=InvokeFrom.EXPLORE,
+        streaming=True,
+        workflow_run_id="workflow-run-id",
+    )
+    runner = _AppRunner(session_factory=MagicMock(), exec_params=exec_params)
+
+    workflow = SimpleNamespace(id="workflow-id", app_id="app-id", created_by="workflow-owner")
+    app = SimpleNamespace(id="app-id")
+    fake_session = MagicMock()
+    fake_session.get.side_effect = [workflow, app]
+
+    monkeypatch.setattr(runner, "_session", lambda: nullcontext(fake_session))
+    monkeypatch.setattr(runner, "_resolve_user", lambda: MagicMock())
+    monkeypatch.setattr(runner, "_setup_flask_context", lambda _user: nullcontext())
+    monkeypatch.setattr(runner, "_run_app", lambda **_kwargs: (_ for _ in ()).throw(ValueError("Invalid upload file")))
+    monkeypatch.setattr(
+        "core.workflow.workflow_entry.WorkflowEntry.handle_special_values",
+        lambda value: (_ for _ in ()).throw(AssertionError("pre-runtime helper should not normalize inputs")),
+    )
+
+    with pytest.raises(ValueError, match="Invalid upload file"):
+        runner.run()
+
+    payloads = _published_payloads(mock_topic)
+    assert payloads[0]["data"]["inputs"] == {}
+    assert payloads[0]["data"]["reason"] == WorkflowStartReason.INITIAL
+
+
+def test_app_runner_streaming_success_calls_publish_streaming_response_with_full_signature(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    exec_params = AppExecutionParams(
+        app_id="app-id",
+        workflow_id="workflow-id",
+        tenant_id="tenant-id",
+        app_mode=AppMode.ADVANCED_CHAT,
+        user={"TYPE": "account", "user_id": "user-id"},
+        args={"inputs": {"foo": "bar"}, "query": "test"},
+        invoke_from=InvokeFrom.EXPLORE,
+        streaming=True,
+        workflow_run_id="workflow-run-id",
+    )
+    runner = _AppRunner(session_factory=MagicMock(), exec_params=exec_params)
+
+    workflow = SimpleNamespace(id="workflow-id", app_id="app-id", created_by="workflow-owner")
+    app = SimpleNamespace(id="app-id")
+    fake_session = MagicMock()
+    fake_session.get.side_effect = [workflow, app]
+    response_stream = _single_event_generator({"event": "message"})
+    publish_streaming_response = MagicMock()
+
+    monkeypatch.setattr(runner, "_session", lambda: nullcontext(fake_session))
+    monkeypatch.setattr(runner, "_resolve_user", lambda: MagicMock())
+    monkeypatch.setattr(runner, "_setup_flask_context", lambda _user: nullcontext())
+    monkeypatch.setattr(runner, "_run_app", lambda **_kwargs: response_stream)
+    monkeypatch.setattr(
+        "tasks.app_generate.workflow_execute_task._publish_streaming_response",
+        publish_streaming_response,
+    )
+
+    runner.run()
+
+    publish_streaming_response.assert_called_once_with(
+        response_stream,
+        exec_params.workflow_run_id,
+        exec_params.app_mode,
+        exec_params.workflow_id,
+        exec_params.args.get("inputs", {}),
+        WorkflowStartReason.INITIAL,
+    )
 
 
 def test_resume_app_execution_queries_message_by_conversation_and_workflow_run(monkeypatch: pytest.MonkeyPatch):
@@ -247,6 +692,7 @@ def test_resume_app_execution_returns_early_when_advanced_chat_missing_conversat
 def test_resume_advanced_chat_publishes_events_for_originally_blocking_runs(monkeypatch: pytest.MonkeyPatch):
     generate_entity = _build_advanced_chat_generate_entity(conversation_id="conversation-id")
     generate_entity.stream = False
+    workflow = SimpleNamespace(id="workflow-id", created_by="workflow-owner")
 
     generator_instance = MagicMock()
     response_stream = _single_event_generator({"event": "message"})
@@ -271,7 +717,7 @@ def test_resume_advanced_chat_publishes_events_for_originally_blocking_runs(monk
 
     _resume_advanced_chat(
         app_model=SimpleNamespace(id="app-id"),
-        workflow=SimpleNamespace(created_by="workflow-owner"),
+        workflow=workflow,
         user=MagicMock(),
         conversation=SimpleNamespace(id="conversation-id"),
         message=MagicMock(),
@@ -285,11 +731,19 @@ def test_resume_advanced_chat_publishes_events_for_originally_blocking_runs(monk
 
     resumed_entity = generator_instance.resume.call_args.kwargs["application_generate_entity"]
     assert resumed_entity.stream is True
-    publish_streaming_response.assert_called_once_with(response_stream, "workflow-run-id", AppMode.ADVANCED_CHAT)
+    publish_streaming_response.assert_called_once_with(
+        response_stream,
+        "workflow-run-id",
+        AppMode.ADVANCED_CHAT,
+        workflow.id,
+        generate_entity.inputs,
+        WorkflowStartReason.RESUMPTION,
+    )
 
 
 def test_resume_workflow_publishes_events_for_originally_blocking_runs(monkeypatch: pytest.MonkeyPatch):
     generate_entity = _build_workflow_generate_entity(stream=False)
+    workflow = SimpleNamespace(id="workflow-id", created_by="workflow-owner")
 
     generator_instance = MagicMock()
     response_stream = _single_event_generator({"event": "workflow_finished"})
@@ -316,7 +770,7 @@ def test_resume_workflow_publishes_events_for_originally_blocking_runs(monkeypat
 
     _resume_workflow(
         app_model=SimpleNamespace(id="app-id"),
-        workflow=SimpleNamespace(created_by="workflow-owner"),
+        workflow=workflow,
         user=MagicMock(),
         generate_entity=generate_entity,
         graph_runtime_state=MagicMock(),
@@ -330,12 +784,20 @@ def test_resume_workflow_publishes_events_for_originally_blocking_runs(monkeypat
 
     resumed_entity = generator_instance.resume.call_args.kwargs["application_generate_entity"]
     assert resumed_entity.stream is True
-    publish_streaming_response.assert_called_once_with(response_stream, "workflow-run-id", AppMode.WORKFLOW)
+    publish_streaming_response.assert_called_once_with(
+        response_stream,
+        "workflow-run-id",
+        AppMode.WORKFLOW,
+        workflow.id,
+        generate_entity.inputs,
+        WorkflowStartReason.RESUMPTION,
+    )
     workflow_run_repo.delete_workflow_pause.assert_called_once_with(pause_entity)
 
 
 def test_resume_workflow_ignores_missing_old_pause_after_repause(monkeypatch: pytest.MonkeyPatch):
     generate_entity = _build_workflow_generate_entity(stream=False)
+    workflow = SimpleNamespace(id="workflow-id", created_by="workflow-owner")
 
     generator_instance = MagicMock()
     response_stream = _single_event_generator({"event": "workflow_paused"})
@@ -363,7 +825,7 @@ def test_resume_workflow_ignores_missing_old_pause_after_repause(monkeypatch: py
 
     _resume_workflow(
         app_model=SimpleNamespace(id="app-id"),
-        workflow=SimpleNamespace(created_by="workflow-owner"),
+        workflow=workflow,
         user=MagicMock(),
         generate_entity=generate_entity,
         graph_runtime_state=MagicMock(),
@@ -375,5 +837,12 @@ def test_resume_workflow_ignores_missing_old_pause_after_repause(monkeypatch: py
         pause_entity=pause_entity,
     )
 
-    publish_streaming_response.assert_called_once_with(response_stream, "workflow-run-id", AppMode.WORKFLOW)
+    publish_streaming_response.assert_called_once_with(
+        response_stream,
+        "workflow-run-id",
+        AppMode.WORKFLOW,
+        workflow.id,
+        generate_entity.inputs,
+        WorkflowStartReason.RESUMPTION,
+    )
     workflow_run_repo.delete_workflow_pause.assert_called_once_with(pause_entity)

api/uv.lock

@@ -1331,7 +1331,7 @@ docs = [
 
 [[package]]
 name = "dify-api"
-version = "1.14.2"
+version = "1.15.0"
 source = { virtual = "." }
 dependencies = [
     { name = "aliyun-log-python-sdk" },
@@ -1619,7 +1619,7 @@ vdb-xinference = [
 requires-dist = [
     { name = "aliyun-log-python-sdk", specifier = "==0.9.44" },
     { name = "azure-identity", specifier = ">=1.25.3,<2.0.0" },
-    { name = "bleach", specifier = ">=6.3.0,<7.0.0" },
+    { name = "bleach", specifier = ">=6.4.0,<7.0.0" },
     { name = "boto3", specifier = ">=1.43.24,<2.0.0" },
     { name = "celery", specifier = ">=5.6.3,<6.0.0" },
     { name = "croniter", specifier = ">=6.2.2,<7.0.0" },

cli/src/commands/auth/login/device-flow.ts

@@ -89,7 +89,7 @@ async function pollWithRetry(
 
 function expired(): BaseError {
   return new BaseError({
-    code: ErrorCode.ExpiredToken,
+    code: ErrorCode.TokenExpired,
     message: 'code expired before authorization',
   })
 }

cli/src/commands/version/version.test.ts

@@ -158,6 +158,6 @@ describe('Version command', () => {
     if (output?.kind !== 'formatted')
       throw new Error('expected formatted output')
 
-    expect(output.data.text()).toContain('WARNING: This build is a rc release')
+    expect(output.data.text()).toContain('WARNING: This build is a(n) rc release')
   })
 })

cli/src/errors/codes.test.ts

@@ -35,7 +35,6 @@ describe('error codes', () => {
     [ErrorCode.AuthExpired, ExitCode.Auth],
     [ErrorCode.TokenExpired, ExitCode.Auth],
     [ErrorCode.AccessDenied, ExitCode.Auth],
-    [ErrorCode.ExpiredToken, ExitCode.Auth],
     [ErrorCode.VersionSkew, ExitCode.VersionCompat],
     [ErrorCode.UnsupportedEndpoint, ExitCode.VersionCompat],
     [ErrorCode.ConfigSchemaUnsupported, ExitCode.VersionCompat],

cli/src/errors/codes.ts

@@ -3,7 +3,6 @@ export const ErrorCode = {
   AuthExpired: 'auth_expired',
   TokenExpired: 'token_expired',
   AccessDenied: 'access_denied',
-  ExpiredToken: 'expired_token',
   VersionSkew: 'version_skew',
   UnsupportedEndpoint: 'unsupported_endpoint',
   ConfigSchemaUnsupported: 'config_schema_unsupported',
@@ -40,7 +39,6 @@ const CODE_TO_EXIT: Readonly<Record<ErrorCodeValue, ExitCodeValue>> = {
   auth_expired: ExitCode.Auth,
   token_expired: ExitCode.Auth,
   access_denied: ExitCode.Auth,
-  expired_token: ExitCode.Auth,
   version_skew: ExitCode.VersionCompat,
   unsupported_endpoint: ExitCode.VersionCompat,
   config_schema_unsupported: ExitCode.VersionCompat,

cli/src/http/error-mapper.test.ts

@@ -33,7 +33,7 @@ describe('classifyResponse — canonical ErrorBody', () => {
     expect(err.code).toBe(ErrorCode.Server4xxOther)
   })
 
-  it('401 classifies by status as AuthExpired with CLI login hint', async () => {
+  it('401 unauthorized classifies as AuthExpired with CLI login hint', async () => {
     const err = await classified(401, {
       code: 'unauthorized',
       message: 'session expired or revoked',
@@ -44,6 +44,20 @@ describe('classifyResponse — canonical ErrorBody', () => {
     expect(err.hint).toBe('run \'difyctl auth login\' to sign in again')
   })
 
+  it('401 token_expired carries the structured TokenExpired code with the server message', async () => {
+    const err = await classified(401, {
+      code: 'token_expired',
+      message: 'Your session has expired.',
+      status: 401,
+      hint: 'Re-authenticate to continue (e.g. re-run your login command).',
+    })
+
+    expect(err.code).toBe(ErrorCode.TokenExpired)
+    expect(err.exit()).toBe(4)
+    expect(err.message).toBe('Your session has expired.')
+    expect(err.hint).toBe('run \'difyctl auth login\' to sign in again')
+  })
+
   it('unknown future server code is data, not behavior — status bucket decides', async () => {
     const err = await classified(409, {
       code: 'some_future_code',

cli/src/http/error-mapper.ts

@@ -1,6 +1,6 @@
 import type { ErrorBody } from '@dify/contracts/api/openapi/types.gen'
 import type { ErrorCodeValue } from '@/errors/codes'
-import { zErrorBody } from '@dify/contracts/api/openapi/zod.gen'
+import { zErrorBody, zOpenApiErrorCode } from '@dify/contracts/api/openapi/zod.gen'
 import { BaseError, HttpClientError, newError } from '@/errors/base'
 import { ErrorCode } from '@/errors/codes'
 import { redactBearer } from './sanitize'
@@ -24,6 +24,16 @@ const AUTH_EXPIRED_CLASS: StatusClass = {
   includeRaw: false,
 }
 
+// A 401 whose body carries the server's `token_expired` code is a known,
+// behavior-driving signal (not opaque data): the session lapsed rather than the
+// token being unknown/revoked, so wrappers get the distinct structured code.
+const TOKEN_EXPIRED_CLASS: StatusClass = {
+  code: ErrorCode.TokenExpired,
+  fallbackMessage: () => 'session expired',
+  hint: AUTH_LOGIN_HINT,
+  includeRaw: false,
+}
+
 const SERVER_5XX_CLASS: StatusClass = {
   code: ErrorCode.Server5xx,
   fallbackMessage: status => `server error (HTTP ${status})`,
@@ -50,9 +60,9 @@ const ACCESS_DENIED_CLASS: StatusClass = {
   includeRaw: false,
 }
 
-function statusClass(status: number): StatusClass {
+function statusClass(status: number, serverError?: ErrorBody): StatusClass {
   if (status === 401)
-    return AUTH_EXPIRED_CLASS
+    return serverError?.code === zOpenApiErrorCode.enum.token_expired ? TOKEN_EXPIRED_CLASS : AUTH_EXPIRED_CLASS
   if (status === 403)
     return ACCESS_DENIED_CLASS
   if (status === 429)
@@ -87,7 +97,7 @@ export async function classifyResponse(request: Request, response: Response): Pr
 
   const serverError = parseServerError(raw)
   const status = response.status
-  const c = statusClass(status)
+  const c = statusClass(status, serverError)
   return new HttpClientError({
     code: c.code,
     message: serverError?.message ?? c.fallbackMessage(status),

cli/src/version/render.test.ts

@@ -60,7 +60,7 @@ describe('renderVersionText', () => {
     }
     const text = renderVersionText(report)
 
-    expect(text).toContain('WARNING: This build is a rc release')
+    expect(text).toContain('WARNING: This build is a(n) rc release')
     expect(text).toContain('install or wait for the stable channel')
   })
 
@@ -72,7 +72,7 @@ describe('renderVersionText', () => {
     }
     const text = renderVersionText(report)
 
-    expect(text).toContain('WARNING: This build is a alpha release')
+    expect(text).toContain('WARNING: This build is a(n) alpha release')
     expect(text).toContain('install or wait for the stable channel')
   })
 
@@ -84,7 +84,7 @@ describe('renderVersionText', () => {
     }
     const text = renderVersionText(report)
 
-    expect(text).toContain('WARNING: This build is a edge release')
+    expect(text).toContain('WARNING: This build is a(n) edge release')
     expect(text).toContain('install or wait for the stable channel')
   })
 
@@ -140,7 +140,7 @@ describe('renderVersionText', () => {
     // RC warning) ran, yet the output is byte-clean.
     expect(plain).not.toMatch(ANSI_RE)
     expect(plain).toContain('Compatibility: incompatible')
-    expect(plain).toContain('WARNING: This build is a rc release')
+    expect(plain).toContain('WARNING: This build is a(n) rc release')
   })
 
   describe('with picocolors stubbed to always emit ANSI', () => {
@@ -183,7 +183,7 @@ describe('renderVersionText', () => {
       expect(colored).toMatch(ANSI_RE)
       expect(colored).toContain('Compatibility: incompatible')
       // prerelease warning lines also routed through yellow.
-      expect(colored).toContain('WARNING: This build is a rc release')
+      expect(colored).toContain('WARNING: This build is a(n) rc release')
     })
   })
 

cli/src/version/render.ts

@@ -4,7 +4,7 @@ import { colorScheme } from '@/sys/io/color'
 
 function prereleaseWarning(channel: Channel): readonly string[] {
   return [
-    `WARNING: This build is a ${channel} release. It is not stable`,
+    `WARNING: This build is a(n) ${channel} release. It is not stable`,
     '         and may have bugs. For production use, install or wait for the stable channel.',
   ]
 }

dev/pyrefly-check-local

@@ -28,6 +28,10 @@ pyrefly_args=(
   "--project-excludes=tests/"
 )
 
+if [[ "${PYREFLY_OUTPUT_FORMAT:-}" == "github" ]]; then
+  pyrefly_args+=("--output-format=github")
+fi
+
 if [[ -f "$EXCLUDES_FILE" ]]; then
   while IFS= read -r exclude; do
     [[ -z "$exclude" || "${exclude:0:1}" == "#" ]] && continue
@@ -36,6 +40,14 @@ if [[ -f "$EXCLUDES_FILE" ]]; then
 fi
 
 run_pyrefly() {
+  if [[ "${PYREFLY_OUTPUT_FORMAT:-}" == "github" ]]; then
+    set +e
+    "$@"
+    local pyrefly_status=$?
+    set -e
+    return "$pyrefly_status"
+  fi
+
   local tmp_output
   tmp_output="$(mktemp)"
 
@@ -44,7 +56,7 @@ run_pyrefly() {
   local pyrefly_status=$?
   set -e
 
-  uv run --directory api python libs/pyrefly_diagnostics.py < "$tmp_output"
+  uv run --directory api python libs/pyrefly_diagnostics.py --status "$pyrefly_status" < "$tmp_output"
   rm -f "$tmp_output"
   return "$pyrefly_status"
 }
@@ -62,11 +74,17 @@ fi
 run_pyrefly "${pyrefly_command[@]}" || status=$?
 
 if (( ${#target_paths[@]} == 0 )); then
+  test_containers_args=(
+    "--summary=none"
+    "--use-ignore-files=false"
+    "--config=$TEST_CONTAINERS_CONFIG"
+  )
+  if [[ "${PYREFLY_OUTPUT_FORMAT:-}" == "github" ]]; then
+    test_containers_args+=("--output-format=github")
+  fi
   run_pyrefly \
     uv run --directory api --dev pyrefly check \
-    "--summary=none" \
-    "--use-ignore-files=false" \
-    "--config=$TEST_CONTAINERS_CONFIG" \
+    "${test_containers_args[@]}" \
     || status=$?
 fi
 

docker/docker-compose-template.yaml

@@ -220,7 +220,7 @@ services:
   # API service
   api:
     <<: *shared-api-worker-config
-    image: langgenius/dify-api:1.14.2
+    image: langgenius/dify-api:1.15.0
     environment:
       MODE: api
       SENTRY_DSN: ${API_SENTRY_DSN:-}
@@ -264,7 +264,7 @@ services:
   # WebSocket service for workflow collaboration.
   api_websocket:
     <<: *shared-api-worker-config
-    image: langgenius/dify-api:1.14.2
+    image: langgenius/dify-api:1.15.0
     profiles:
       - collaboration
     environment:
@@ -290,7 +290,7 @@ services:
   # The Celery worker for processing all queues (dataset, workflow, mail, etc.)
   worker:
     <<: *shared-worker-config
-    image: langgenius/dify-api:1.14.2
+    image: langgenius/dify-api:1.15.0
     environment:
       MODE: worker
       SENTRY_DSN: ${API_SENTRY_DSN:-}
@@ -333,7 +333,7 @@ services:
   # Celery beat for scheduling periodic tasks.
   worker_beat:
     <<: *shared-worker-beat-config
-    image: langgenius/dify-api:1.14.2
+    image: langgenius/dify-api:1.15.0
     environment:
       MODE: beat
     depends_on:
@@ -366,7 +366,7 @@ services:
 
   # Frontend web application.
   web:
-    image: langgenius/dify-web:1.14.2
+    image: langgenius/dify-web:1.15.0
     restart: always
     env_file:
       - path: ./envs/core-services/web.env
@@ -518,7 +518,7 @@ services:
 
   # plugin daemon
   plugin_daemon:
-    image: langgenius/dify-plugin-daemon:0.6.1-local
+    image: langgenius/dify-plugin-daemon:0.6.3-local
     restart: always
     env_file:
       - path: ./envs/core-services/shared.env

docker/docker-compose.middleware.yaml

@@ -129,7 +129,7 @@ services:
 
   # plugin daemon
   plugin_daemon:
-    image: langgenius/dify-plugin-daemon:0.6.1-local
+    image: langgenius/dify-plugin-daemon:0.6.3-local
     restart: always
     env_file:
       - ./middleware.env

docker/docker-compose.yaml

@@ -226,7 +226,7 @@ services:
   # API service
   api:
     <<: *shared-api-worker-config
-    image: langgenius/dify-api:1.14.2
+    image: langgenius/dify-api:1.15.0
     environment:
       MODE: api
       SENTRY_DSN: ${API_SENTRY_DSN:-}
@@ -270,7 +270,7 @@ services:
   # WebSocket service for workflow collaboration.
   api_websocket:
     <<: *shared-api-worker-config
-    image: langgenius/dify-api:1.14.2
+    image: langgenius/dify-api:1.15.0
     profiles:
       - collaboration
     environment:
@@ -296,7 +296,7 @@ services:
   # The Celery worker for processing all queues (dataset, workflow, mail, etc.)
   worker:
     <<: *shared-worker-config
-    image: langgenius/dify-api:1.14.2
+    image: langgenius/dify-api:1.15.0
     environment:
       MODE: worker
       SENTRY_DSN: ${API_SENTRY_DSN:-}
@@ -339,7 +339,7 @@ services:
   # Celery beat for scheduling periodic tasks.
   worker_beat:
     <<: *shared-worker-beat-config
-    image: langgenius/dify-api:1.14.2
+    image: langgenius/dify-api:1.15.0
     environment:
       MODE: beat
     depends_on:
@@ -372,7 +372,7 @@ services:
 
   # Frontend web application.
   web:
-    image: langgenius/dify-web:1.14.2
+    image: langgenius/dify-web:1.15.0
     restart: always
     env_file:
       - path: ./envs/core-services/web.env
@@ -524,7 +524,7 @@ services:
 
   # plugin daemon
   plugin_daemon:
-    image: langgenius/dify-plugin-daemon:0.6.1-local
+    image: langgenius/dify-plugin-daemon:0.6.3-local
     restart: always
     env_file:
       - path: ./envs/core-services/shared.env

eslint-suppressions.json

@@ -458,6 +458,14 @@
       "count": 1
     }
   },
+  "web/app/components/app/app-access-control/access-control-item.tsx": {
+    "jsx-a11y/click-events-have-key-events": {
+      "count": 1
+    },
+    "jsx-a11y/no-static-element-interactions": {
+      "count": 1
+    }
+  },
   "web/app/components/app/app-publisher/sections.tsx": {
     "jsx-a11y/click-events-have-key-events": {
       "count": 1
@@ -3274,7 +3282,7 @@
   },
   "web/app/components/develop/code.tsx": {
     "ts/no-explicit-any": {
-      "count": 7
+      "count": 6
     }
   },
   "web/app/components/develop/doc.tsx": {
@@ -3286,9 +3294,6 @@
     "jsx-a11y/no-redundant-roles": {
       "count": 1
     },
-    "ts/no-empty-object-type": {
-      "count": 1
-    },
     "ts/no-explicit-any": {
       "count": 2
     }
@@ -4476,11 +4481,6 @@
       "count": 1
     }
   },
-  "web/app/components/tools/provider/detail.tsx": {
-    "jsx-a11y/anchor-has-content": {
-      "count": 1
-    }
-  },
   "web/app/components/tools/provider/tool-item.tsx": {
     "jsx-a11y/click-events-have-key-events": {
       "count": 1
@@ -7128,6 +7128,11 @@
       "count": 1
     }
   },
+  "web/service/access-control/__tests__/index.spec.tsx": {
+    "no-restricted-imports": {
+      "count": 1
+    }
+  },
   "web/service/access-control/__tests__/use-app-access-control.spec.tsx": {
     "no-restricted-imports": {
       "count": 1
@@ -7153,6 +7158,11 @@
       "count": 1
     }
   },
+  "web/service/access-control/index.ts": {
+    "no-restricted-imports": {
+      "count": 1
+    }
+  },
   "web/service/access-control/use-app-access-control.ts": {
     "no-restricted-imports": {
       "count": 1
@@ -7227,7 +7237,7 @@
       "count": 1
     },
     "ts/no-explicit-any": {
-      "count": 13
+      "count": 9
     }
   },
   "web/service/datasets.ts": {

packages/contracts/generated/api/console/agent/types.gen.ts

@@ -269,6 +269,7 @@ export type MessageDetailResponse = {
   agent_thoughts?: Array<AgentThought>
   annotation?: ConversationAnnotation | null
   annotation_hit_history?: ConversationAnnotationHitHistory | null
+  answer: string
   answer_tokens?: number | null
   conversation_id: string
   created_at?: number | null
@@ -284,12 +285,11 @@ export type MessageDetailResponse = {
   }
   message?: JsonValue | null
   message_files?: Array<MessageFile>
-  message_metadata_dict?: JsonValue | null
   message_tokens?: number | null
+  metadata?: JsonValue | null
   parent_message_id?: string | null
   provider_response_latency?: number | null
   query: string
-  re_sign_file_url_answer: string
   status: string
   workflow_run_id?: string | null
 }
@@ -723,7 +723,6 @@ export type AgentThought = {
   created_at?: number | null
   files: Array<string>
   id: string
-  message_chain_id?: string | null
   message_id: string
   observation?: string | null
   position: number
@@ -743,8 +742,8 @@ export type ConversationAnnotation = {
 
 export type ConversationAnnotationHitHistory = {
   annotation_create_account?: SimpleAccount | null
+  annotation_id: string
   created_at?: number | null
-  id: string
 }
 
 export type HumanInputContent = {

packages/contracts/generated/api/console/agent/zod.gen.ts

@@ -570,7 +570,6 @@ export const zAgentThought = z.object({
   created_at: z.int().nullish(),
   files: z.array(z.string()),
   id: z.string(),
-  message_chain_id: z.string().nullish(),
   message_id: z.string(),
   observation: z.string().nullish(),
   position: z.int(),
@@ -1056,8 +1055,8 @@ export const zConversationAnnotation = z.object({
  */
 export const zConversationAnnotationHitHistory = z.object({
   annotation_create_account: zSimpleAccount.nullish(),
+  annotation_id: z.string(),
   created_at: z.int().nullish(),
-  id: z.string(),
 })
 
 /**
@@ -2035,6 +2034,7 @@ export const zMessageDetailResponse = z.object({
   agent_thoughts: z.array(zAgentThought).optional(),
   annotation: zConversationAnnotation.nullish(),
   annotation_hit_history: zConversationAnnotationHitHistory.nullish(),
+  answer: z.string(),
   answer_tokens: z.int().nullish(),
   conversation_id: z.string(),
   created_at: z.int().nullish(),
@@ -2048,12 +2048,11 @@ export const zMessageDetailResponse = z.object({
   inputs: z.record(z.string(), zJsonValue),
   message: zJsonValue.nullish(),
   message_files: z.array(zMessageFile).optional(),
-  message_metadata_dict: zJsonValue.nullish(),
   message_tokens: z.int().nullish(),
+  metadata: zJsonValue.nullish(),
   parent_message_id: z.string().nullish(),
   provider_response_latency: z.number().nullish(),
   query: z.string(),
-  re_sign_file_url_answer: z.string(),
   status: z.string(),
   workflow_run_id: z.string().nullish(),
 })

packages/contracts/generated/api/console/apps/types.gen.ts

@@ -472,6 +472,7 @@ export type MessageDetailResponse = {
   agent_thoughts?: Array<AgentThought>
   annotation?: ConversationAnnotation | null
   annotation_hit_history?: ConversationAnnotationHitHistory | null
+  answer: string
   answer_tokens?: number | null
   conversation_id: string
   created_at?: number | null
@@ -487,12 +488,11 @@ export type MessageDetailResponse = {
   }
   message?: JsonValue | null
   message_files?: Array<MessageFile>
-  message_metadata_dict?: JsonValue | null
   message_tokens?: number | null
+  metadata?: JsonValue | null
   parent_message_id?: string | null
   provider_response_latency?: number | null
   query: string
-  re_sign_file_url_answer: string
   status: string
   workflow_run_id?: string | null
 }
@@ -1498,7 +1498,6 @@ export type AgentThought = {
   created_at?: number | null
   files: Array<string>
   id: string
-  message_chain_id?: string | null
   message_id: string
   observation?: string | null
   position: number
@@ -1518,8 +1517,8 @@ export type ConversationAnnotation = {
 
 export type ConversationAnnotationHitHistory = {
   annotation_create_account?: SimpleAccount | null
+  annotation_id: string
   created_at?: number | null
-  id: string
 }
 
 export type HumanInputContent = {
@@ -4496,6 +4495,7 @@ export type DeleteAppsByAppIdTraceConfigData = {
 
 export type DeleteAppsByAppIdTraceConfigErrors = {
   400: unknown
+  403: unknown
 }
 
 export type DeleteAppsByAppIdTraceConfigResponses = {
@@ -4538,6 +4538,7 @@ export type PatchAppsByAppIdTraceConfigData = {
 
 export type PatchAppsByAppIdTraceConfigErrors = {
   400: unknown
+  403: unknown
 }
 
 export type PatchAppsByAppIdTraceConfigResponses = {
@@ -4558,6 +4559,7 @@ export type PostAppsByAppIdTraceConfigData = {
 
 export type PostAppsByAppIdTraceConfigErrors = {
   400: unknown
+  403: unknown
 }
 
 export type PostAppsByAppIdTraceConfigResponses = {

packages/contracts/generated/api/console/apps/zod.gen.ts

@@ -1150,7 +1150,6 @@ export const zAgentThought = z.object({
   created_at: z.int().nullish(),
   files: z.array(z.string()),
   id: z.string(),
-  message_chain_id: z.string().nullish(),
   message_id: z.string(),
   observation: z.string().nullish(),
   position: z.int(),
@@ -1371,8 +1370,8 @@ export const zConversationAnnotation = z.object({
  */
 export const zConversationAnnotationHitHistory = z.object({
   annotation_create_account: zSimpleAccount.nullish(),
+  annotation_id: z.string(),
   created_at: z.int().nullish(),
-  id: z.string(),
 })
 
 /**
@@ -3455,6 +3454,7 @@ export const zMessageDetailResponse = z.object({
   agent_thoughts: z.array(zAgentThought).optional(),
   annotation: zConversationAnnotation.nullish(),
   annotation_hit_history: zConversationAnnotationHitHistory.nullish(),
+  answer: z.string(),
   answer_tokens: z.int().nullish(),
   conversation_id: z.string(),
   created_at: z.int().nullish(),
@@ -3468,12 +3468,11 @@ export const zMessageDetailResponse = z.object({
   inputs: z.record(z.string(), zJsonValue),
   message: zJsonValue.nullish(),
   message_files: z.array(zMessageFile).optional(),
-  message_metadata_dict: zJsonValue.nullish(),
   message_tokens: z.int().nullish(),
+  metadata: zJsonValue.nullish(),
   parent_message_id: z.string().nullish(),
   provider_response_latency: z.number().nullish(),
   query: z.string(),
-  re_sign_file_url_answer: z.string(),
   status: z.string(),
   workflow_run_id: z.string().nullish(),
 })

packages/contracts/generated/api/console/installed-apps/types.gen.ts

@@ -232,6 +232,7 @@ export type SavedMessageItem = {
 }
 
 export type InstalledAppInfoResponse = {
+  description?: string | null
   icon?: string | null
   icon_background?: string | null
   icon_type?: string | null
@@ -246,7 +247,6 @@ export type AgentThought = {
   created_at?: number | null
   files: Array<string>
   id: string
-  message_chain_id?: string | null
   message_id: string
   observation?: string | null
   position: number

packages/contracts/generated/api/console/installed-apps/zod.gen.ts

@@ -229,6 +229,7 @@ export const zParameters = z.object({
  * InstalledAppInfoResponse
  */
 export const zInstalledAppInfoResponse = z.object({
+  description: z.string().nullish(),
   icon: z.string().nullish(),
   icon_background: z.string().nullish(),
   icon_type: z.string().nullish(),
@@ -266,7 +267,6 @@ export const zAgentThought = z.object({
   created_at: z.int().nullish(),
   files: z.array(z.string()),
   id: z.string(),
-  message_chain_id: z.string().nullish(),
   message_id: z.string(),
   observation: z.string().nullish(),
   position: z.int(),

packages/contracts/generated/api/openapi/types.gen.ts

@@ -340,6 +340,7 @@ export type OpenApiErrorCode
     | 'rate_limit_error'
     | 'recipient_surface_mismatch'
     | 'request_entity_too_large'
+    | 'token_expired'
     | 'too_many_files'
     | 'too_many_requests'
     | 'unauthorized'
@@ -405,6 +406,10 @@ export type SessionRow = {
   prefix: string
 }
 
+export type SimpleResultResponse = {
+  result: string
+}
+
 export type SupportedAppType = 'advanced-chat' | 'agent-chat' | 'chat' | 'completion' | 'workflow'
 
 export type TaskStopResponse = {

packages/contracts/generated/api/openapi/zod.gen.ts

@@ -423,6 +423,7 @@ export const zOpenApiErrorCode = z.enum([
   'rate_limit_error',
   'recipient_surface_mismatch',
   'request_entity_too_large',
+  'token_expired',
   'too_many_files',
   'too_many_requests',
   'unauthorized',
@@ -501,6 +502,13 @@ export const zSessionListResponse = z.object({
   total: z.int(),
 })
 
+/**
+ * SimpleResultResponse
+ */
+export const zSimpleResultResponse = z.object({
+  result: z.string(),
+})
+
 /**
  * SupportedAppType
  *

packages/contracts/generated/enterprise/orpc.gen.ts

@@ -25,8 +25,6 @@ import {
   zAccessServiceUpdateAccessPolicyBody,
   zAccessServiceUpdateAccessPolicyPath,
   zAccessServiceUpdateAccessPolicyResponse,
-  zAccessSubjectServiceListAccessSubjectsQuery,
-  zAccessSubjectServiceListAccessSubjectsResponse,
   zAppInstanceServiceCreateAppInstanceBody,
   zAppInstanceServiceCreateAppInstanceResponse,
   zAppInstanceServiceDeleteAppInstancePath,
@@ -109,21 +107,6 @@ import {
   zWebAppAuthUpdateWebAppWhitelistSubjectsResponse,
 } from './zod.gen'
 
-export const listAccessSubjects = oc
-  .route({
-    inputStructure: 'detailed',
-    method: 'GET',
-    operationId: 'AccessSubjectService_ListAccessSubjects',
-    path: '/enterprise/access-subjects',
-    tags: ['AccessSubjectService'],
-  })
-  .input(z.object({ query: zAccessSubjectServiceListAccessSubjectsQuery.optional() }))
-  .output(zAccessSubjectServiceListAccessSubjectsResponse)
-
-export const accessSubjectService = {
-  listAccessSubjects,
-}
-
 export const listAppInstanceSummaries = oc
   .route({
     inputStructure: 'detailed',
@@ -730,7 +713,6 @@ export const webAppAuth = {
 }
 
 export const contract = {
-  accessSubjectService,
   appInstanceService,
   accessService,
   deploymentService,

packages/contracts/generated/enterprise/types.gen.ts

@@ -1038,6 +1038,7 @@ export type UpdateEnvironmentRequest = {
   environmentId?: string
   displayName: string
   description?: string
+  cpuCount?: number
 }
 
 export type UpdateEnvironmentResponse = {
@@ -1367,6 +1368,7 @@ export type InfoConfigReply = {
   Branding?: BrandingInfo
   WebAppAuth?: WebAppAuthInfo
   PluginInstallationPermission?: PluginInstallationPermissionInfo
+  EnableAppDeploy?: boolean
 }
 
 export type InnerAdmission = {
@@ -1513,11 +1515,6 @@ export type LimitFields = {
   appRunnerEnvCpus?: ResourceQuota
 }
 
-export type ListAccessSubjectsReply = {
-  subjects?: Array<Subject>
-  pagination?: Pagination
-}
-
 export type ListGroupAppsResponse = {
   items?: Array<GroupAppItem>
   total?: string
@@ -2138,25 +2135,6 @@ export type Pagination = {
   totalPages?: number
 }
 
-export type AccessSubjectServiceListAccessSubjectsData = {
-  body?: never
-  path?: never
-  query?: {
-    keyword?: string
-    groupId?: string
-    pageNumber?: number
-    resultsPerPage?: number
-  }
-  url: '/enterprise/access-subjects'
-}
-
-export type AccessSubjectServiceListAccessSubjectsResponses = {
-  200: ListAccessSubjectsReply
-}
-
-export type AccessSubjectServiceListAccessSubjectsResponse
-  = AccessSubjectServiceListAccessSubjectsResponses[keyof AccessSubjectServiceListAccessSubjectsResponses]
-
 export type AppInstanceServiceListAppInstanceSummariesData = {
   body?: never
   path?: never

packages/contracts/generated/enterprise/zod.gen.ts

@@ -912,6 +912,7 @@ export const zUpdateEnvironmentRequest = z.object({
   environmentId: z.string().optional(),
   displayName: z.string(),
   description: z.string().optional(),
+  cpuCount: z.number().optional(),
 })
 
 export const zUpdateEnvironmentResponse = z.object({
@@ -2072,6 +2073,7 @@ export const zInfoConfigReply = z.object({
   Branding: zBrandingInfo.optional(),
   WebAppAuth: zWebAppAuthInfo.optional(),
   PluginInstallationPermission: zPluginInstallationPermissionInfo.optional(),
+  EnableAppDeploy: z.boolean().optional(),
 })
 
 export const zWebOAuth2LoginReply = z.object({
@@ -2243,11 +2245,6 @@ export const zListRollbackTargetsResponse = z.object({
   pagination: zPagination,
 })
 
-export const zListAccessSubjectsReply = z.object({
-  subjects: z.array(zSubject).optional(),
-  pagination: zPagination.optional(),
-})
-
 export const zListMembersReply = z.object({
   data: z.array(zAccountDetail).optional(),
   pagination: zPagination.optional(),
@@ -2268,26 +2265,6 @@ export const zListWorkspacesReply = z.object({
   pagination: zPagination.optional(),
 })
 
-export const zAccessSubjectServiceListAccessSubjectsQuery = z.object({
-  keyword: z.string().optional(),
-  groupId: z.string().optional(),
-  pageNumber: z
-    .int()
-    .min(-2147483648, { error: 'Invalid value: Expected int32 to be >= -2147483648' })
-    .max(2147483647, { error: 'Invalid value: Expected int32 to be <= 2147483647' })
-    .optional(),
-  resultsPerPage: z
-    .int()
-    .min(-2147483648, { error: 'Invalid value: Expected int32 to be >= -2147483648' })
-    .max(2147483647, { error: 'Invalid value: Expected int32 to be <= 2147483647' })
-    .optional(),
-})
-
-/**
- * OK
- */
-export const zAccessSubjectServiceListAccessSubjectsResponse = zListAccessSubjectsReply
-
 export const zAppInstanceServiceListAppInstanceSummariesQuery = z.object({
   pageNumber: z
     .int()