fix(workflow): polish start selector states

fix(workflow): use scroll area for trigger actions
feat(workflow): align start selector design
2026-06-08 20:08:03 -07:00 · 2026-06-08 19:38:14 -07:00 · 2026-06-08 19:13:11 -07:00 · 2026-06-08 19:10:09 -07:00 · 2026-06-08 19:08:33 -07:00 · 2026-06-08 14:19:39 -07:00
1573 changed files with 77668 additions and 14353 deletions
@@ -1,73 +1,93 @@
 ---
 name: frontend-code-review
-description: "Trigger when the user requests a review of frontend files (e.g., `.tsx`, `.ts`, `.js`). Support both pending-change reviews and focused file reviews while applying the checklist rules."
+description: Review Dify frontend code for correctness, accessibility, component design, dify-ui usage, data/query boundaries, performance, and tests. Trigger for `.tsx`, `.ts`, `.js`, UI, React, Next.js, pending-change, or focused frontend review requests.
 ---

 # Frontend Code Review

-## Intent
-Use this skill whenever the user asks to review frontend code (especially `.tsx`, `.ts`, or `.js` files). Support two review modes:
+## When To Use

-1. **Pending-change review** – inspect staged/working-tree files slated for commit and flag checklist violations before submission.
-2. **File-targeted review** – review the specific file(s) the user names and report the relevant checklist findings.
+Use this skill when the user asks to review, audit, analyze, or sanity-check frontend code under `web/`, `packages/dify-ui/`, or frontend-adjacent TypeScript files.

-Stick to the checklist below for every applicable file and mode.
+Supported modes:

-## Checklist
-See [references/code-quality.md](references/code-quality.md), [references/performance.md](references/performance.md), [references/business-logic.md](references/business-logic.md) for the living checklist split by category—treat it as the canonical set of rules to follow.
+- **Pending-change review**: inspect staged and working-tree changes.
+- **File-focused review**: inspect explicitly named files or paths.
+- **Diff/snippet review**: review pasted diffs or snippets using best-effort references.

-Flag each rule violation with urgency metadata so future reviewers can prioritize fixes.
+Do not use this skill for backend-only code under `api/`; use `backend-code-review` instead.
+
+## Required Context
+
+Before reviewing, read the relevant local contracts:
+
+- `web/AGENTS.md` for Dify frontend workflow, overlays, design tokens, state, and tests.
+- `packages/dify-ui/README.md` and `packages/dify-ui/AGENTS.md` when code uses or changes `@langgenius/dify-ui/*`.
+- `web/docs/overlay.md` when reviewing dialogs, drawers, popovers, tooltips, menus, selects, comboboxes, or other floating UI.
+- `web/docs/test.md` and the `frontend-testing` skill when reviewing tests or testability.
+- `how-to-write-component` when reviewing React component structure, ownership, effects, query/mutation contracts, or memoization.
+
+For any UI, UX, or accessibility review, fetch the latest Web Interface Guidelines before finalizing findings. Treat them as a required baseline, not the complete source of accessibility truth:
+
+```text
+https://raw.githubusercontent.com/vercel-labs/web-interface-guidelines/main/command.md
+```
+
+If the review depends on a current framework, SDK, browser API, or accessibility behavior and local code does not settle it, check the current official docs first. For browser compatibility, deprecation, or behavior-sensitive frontend APIs, verify MDN or the relevant standard.
+
+## Rule Packs
+
+Apply every relevant rule pack:
+
+- [references/accessibility-ui.md](references/accessibility-ui.md) — accessibility, semantic HTML, focus, forms, keyboard, disabled states, copy, and long-content behavior. Combines Web Interface Guidelines with Dify UI, Base UI, MDN, and local primitive contracts.
+- [references/dify-ui.md](references/dify-ui.md) — Dify UI primitive usage, Base UI semantics, overlays, forms, tokens, radius mapping, and primitive boundaries.
+- [references/component-architecture.md](references/component-architecture.md) — component ownership, props, state, effects, exports, wrappers, and feature organization.
+- [references/data-query-contracts.md](references/data-query-contracts.md) — generated contracts, TanStack Query, mutations, workspace/auth/SSR boundaries, URL/local storage state.
+- [references/performance.md](references/performance.md) — React/Next performance review rules from Vercel guidance, scoped to real risk.
+- [references/testing.md](references/testing.md) — frontend test review rules.
+- [references/dify-invariants.md](references/dify-invariants.md) — stable Dify-specific runtime invariants that generic React/a11y rules will not catch.
+- [references/code-quality.md](references/code-quality.md) — general TypeScript, styling, naming, and maintainability rules.

 ## Review Process
-1. Open the relevant component/module. Gather lines that relate to class names, React Flow hooks, prop memoization, and styling.
-2. For each rule in the review point, note where the code deviates and capture a representative snippet.
-3. Compose the review section per the template below. Group violations first by **Urgent** flag, then by category order (Code Quality, Performance, Business Logic).

-## Required output
-When invoked, the response must exactly follow one of the two templates:
+1. Identify the review scope. For pending changes, inspect `git diff --stat`, `git diff`, and staged diff if relevant. For file-focused reviews, stay within the named files unless a referenced owner/contract must be read.
+2. Read code around the changed lines and the owning module. Do not review by isolated snippets when nearby ownership, labels, query inputs, or overlay structure decide correctness.
+3. Check user-visible regressions first: accessibility, broken interaction, auth/permission leaks, query/hydration errors, data loss, navigation mistakes, and impossible states.
+4. Then check maintainability and performance: ownership, effects, wrappers, memoization, bundle/waterfall risks, tests, and design-system drift.
+5. Report only actionable findings. Do not list speculative risks, style preferences, or broad refactors unless they are directly tied to a reproducible issue in scope.

-### Template A (any findings)
-```
-# Code review
-Found <N> urgent issues need to be fixed:
+## Severity

-## 1 <brief description of bug>
-FilePath: <path> line <line>
-<relevant code snippet or pointer>
+- **P0**: security/privacy/auth leak, data loss, production crash, inaccessible critical flow, or broken primary workflow.
+- **P1**: user-visible regression, hydration/SSR failure, invalid API/query contract, broken keyboard/focus behavior, or serious design-system/a11y violation.
+- **P2**: maintainability or performance issue likely to cause bugs, duplicated state, incorrect ownership, missing tests for risky behavior, or non-critical a11y issue.
+- **P3**: minor cleanup with clear value. Omit unless the user asked for a thorough audit.

+## Output Format

-### Suggested fix
-<brief description of suggested fix>
+Lead with findings, ordered by severity. Use this structure:

---
-... (repeat for each urgent issue) ...
+```markdown
+## Findings

-Found <M> suggestions for improvement:
+- [P1] Short issue title
+  File: `path/to/file.tsx:123`
+  Why it matters and how to reproduce or reason about it.
+  Suggested fix: concrete fix direction.

-## 1 <brief description of suggestion>
-FilePath: <path> line <line>
-<relevant code snippet or pointer>
+## Open Questions

+- Question or assumption, if any.

-### Suggested fix
-<brief description of suggested fix>
+## Summary

---
-
-... (repeat for each suggestion) ...
+Brief secondary context. Mention tests not run or residual risk.
 ```

-If there are no urgent issues, omit that section. If there are no suggestions, omit that section.
-
-If the issue number is more than 10, summarize as "10+ urgent issues" or "10+ suggestions" and just output the first 10 issues.
-
-Don't compress the blank lines between sections; keep them as-is for readability.
-
-If you use Template A (i.e., there are issues to fix) and at least one issue requires code changes, append a brief follow-up question after the structured output asking whether the user wants you to apply the suggested fix(es). For example: "Would you like me to use the Suggested fix section to address these issues?"
-
-### Template B (no issues)
-```
-## Code review
-No issues found.
-```
+Rules:

+- If there are no findings, say `No issues found.` and mention any test gaps or residual risk.
+- Always include file and line when available.
+- Keep findings concrete and reproducible.
+- Do not include praise sections by default.
+- Do not ask to apply fixes unless the user explicitly wants review plus implementation.
@@ -0,0 +1,109 @@
+# Accessibility And UI Rules
+
+Accessibility findings are first-class review findings. Treat broken keyboard access, missing accessible names, focus loss, and unreachable popup content as correctness bugs, not polish.
+
+Before finalizing UI or accessibility findings, fetch the latest Web Interface Guidelines as a required baseline:
+
+```text
+https://raw.githubusercontent.com/vercel-labs/web-interface-guidelines/main/command.md
+```
+
+Do not treat that document as the complete accessibility rule set. Combine it with:
+
+- `packages/dify-ui/README.md`, `packages/dify-ui/AGENTS.md`, and the relevant primitive implementation when code uses `@langgenius/dify-ui/*`.
+- Base UI docs and local `.d.ts` contracts when primitive semantics, focus target, labels, or popup reachability are unclear.
+- MDN or relevant WAI-ARIA/browser standards when behavior, compatibility, or deprecation status matters.
+- The current feature's product semantics, because an accessible primitive can still be used in an inaccessible workflow.
+
+## Semantic HTML
+
+Flag:
+
+- Clickable `div` or `span` used for actions.
+- Router navigation implemented with button or `onClick` when a `Link` / `<a>` is the real semantic element.
+- Icon-only buttons without `aria-label` or `aria-labelledby`.
+- Decorative icons missing `aria-hidden="true"`.
+- Images without `alt`; use `alt=""` only when truly decorative.
+- Heading levels that skip hierarchy in page-level content.
+
+Prefer semantic HTML before ARIA.
+
+## Keyboard And Focus
+
+Flag:
+
+- Interactive elements without visible `focus-visible` treatment.
+- `outline-none` / `outline-hidden` without an equivalent focus-visible ring or state.
+- Custom interactive elements missing keyboard handling.
+- Focus trapped, lost, or sent to the wrong surface after dialog/popover/menu close.
+- Focus ring applied to the wrong DOM node. Verify the actual focus target, especially with Base UI controls such as Slider.
+
+Use `focus-visible` for keyboard focus. Use `focus-within` or `has-[:focus-visible]` when the visual wrapper is not the focused element.
+
+## Forms
+
+Flag:
+
+- Inputs, selects, switches, checkboxes, radios, comboboxes, or sliders without a label relationship.
+- Missing stable `name` on form fields that submit or validate.
+- Incorrect input `type`, `inputMode`, `autoComplete`, or `spellCheck` for email, token, URL, number, search, code, or username fields.
+- Labels that are not clickable.
+- Submit buttons disabled before a request starts, preventing normal submit behavior.
+- Non-submit buttons inside forms missing `type="button"`.
+- Errors not associated with fields or not reachable by screen readers.
+- Error recovery that does not focus or expose the first invalid field.
+- `onPaste` blocking paste.
+- Placeholder text used as the only label.
+- Password managers accidentally triggered on non-auth fields because autocomplete is missing or wrong.
+
+Prefer visible labels. If visible surrounding text already labels the control, use a visually hidden label or a precise `aria-label`.
+
+## Disabled, Loading, And Async States
+
+Flag:
+
+- Loading state without `aria-busy`, `role="status"`, or another accessible update path when it changes user interaction.
+- Spinner or decorative loading icon exposed to screen readers.
+- Disabled controls that hide the reason users cannot proceed.
+- `aria-disabled` used without manually blocking click, Space, and Enter.
+- Toasts, inline validation, or async status changes that are not announced when users need the update to continue.
+- Icon-only loading/error affordances without text or accessible status where the state matters.
+
+Use native `disabled` when the control must not be interactive. Use `aria-disabled` only when the element must remain focusable and the code handles all blocked interactions.
+
+For repeated shared disabled reasons, prefer a visible group message or badge plus native disabled controls. Use per-control popover/info only when the reason is item-specific.
+
+## Overlays And Popup Reachability
+
+Flag:
+
+- Tooltip used for long, structured, interactive, or unique information.
+- Tooltip content required to understand or complete a flow.
+- PreviewCard content that touch or screen-reader users cannot reach through the trigger's click destination.
+- Popover/dialog/menu triggers without accessible names.
+- Popup content without title/description where the primitive requires them.
+
+Use Popover for explanatory content, rich help, and infotips. Use Tooltip only as a short visual label for a trigger that already has an accessible name.
+
+## Long Content And Layout
+
+Flag:
+
+- Text in flex/grid children without `min-w-0` when it can overflow.
+- Names, labels, file names, model names, workspace names, or user content lacking `truncate`, `line-clamp`, or `break-words`.
+- Right-side icons, badges, checks, or actions that shrink before the text area.
+- Empty arrays or empty strings rendering broken layout instead of an empty state.
+- Button, tab, badge, chip, menu item, or card text that can overlap sibling controls at common viewport widths.
+
+The usual Dify layout chain is: container has width constraints, text region uses `min-w-0 flex-1 truncate`, adornments use `shrink-0`.
+
+## Motion, Images, And Copy
+
+Flag:
+
+- `transition-all`.
+- Animations that do not respect reduced motion.
+- Layout-affecting animation where transform/opacity would work.
+- Images without dimensions.
+- Loading copy using `...` instead of `…`.
+- Hardcoded dates, times, numbers, or currency formats instead of `Intl.*`.
@@ -1,15 +0,0 @@
-# Rule Catalog — Business Logic
-
-## Can't use workflowStore in Node components
-
-IsUrgent: True
-
-### Description
-
-File path pattern of node components: `web/app/components/workflow/nodes/[nodeName]/node.tsx`
-
-Node components are also used when creating a RAG Pipe from a template, but in that context there is no workflowStore Provider, which results in a blank screen. [This Issue](https://github.com/langgenius/dify/issues/29168) was caused by exactly this reason.
-
-### Suggested Fix
-
-Use `import { useNodes } from 'reactflow'` instead of `import useNodes from '@/app/components/workflow/store/workflow/use-nodes'`.
@@ -1,44 +1,66 @@
-# Rule Catalog — Code Quality
+# Code Quality Rules

-## Conditional class names use utility function
+## Scope Control

-IsUrgent: True
-Category: Code Quality
+Flag changes that expand beyond the requested feature or review scope:

-### Description
+- Repo-wide cleanup mixed into a targeted fix.
+- Compatibility exports, aliases, shims, or wrapper layers added without an explicit migration requirement.
+- Shared abstractions created before there is stable cross-feature reuse.
+- Business components moved into generic shared locations without a clear ownership boundary.

-Ensure conditional CSS is handled via the shared `classNames` instead of custom ternaries, string concatenation, or template strings. Centralizing class logic keeps components consistent and easier to maintain.
+## TypeScript

-### Suggested Fix
+Flag:

-```ts
-import { cn } from '@/utils/classnames'
-const classNames = cn(isActive ? 'text-primary-600' : 'text-gray-500')
-```
+- `any` or broad `Record<string, any>` where generated/API types or local domain types exist.
+- Re-declared API shapes instead of importing generated or returned types.
+- Weak route/query param typing that leaks `string | string[] | undefined` deep into components.
+- Runtime wrappers added only to satisfy TypeScript when a narrower type boundary would preserve the existing runtime shape.

-## Tailwind-first styling
+Prefer:

-IsUrgent: True
-Category: Code Quality
+- Explicit domain names that match the API contract.
+- Type narrowing at route/API boundaries.
+- Small conversion helpers colocated with the component that needs them.

-### Description
+## Styling

-Favor Tailwind CSS utility classes instead of adding new `.module.css` files unless a Tailwind combination cannot achieve the required styling. Keeping styles in Tailwind improves consistency and reduces maintenance overhead.
+Flag:

-Update this file when adding, editing, or removing Code Quality rules so the catalog remains accurate.
+- New CSS modules or ad hoc CSS when Tailwind utilities and Dify tokens cover the need.
+- Generic color utilities where Dify semantic tokens exist.
+- Hardcoded magic class values for colors, spacing, radius, shadow, z-index, or typography when Dify tokens, component variants, or documented radius mappings exist.
+- `!` important modifiers or important CSS overrides without a narrow, documented reason.
+- Manual string concatenation for conditional classes.
+- JS conditional class branches for primitive visual states already exposed by Dify UI/Base UI `data-*` selectors.
+- Incoming `className` placed before default classes in `cn(...)`, preventing call-site overrides.
+- Arbitrary z-index or one-off layering fixes on overlays.

-## Classname ordering for easy overrides
+Use:

-### Description
+- `cn(...)` from the local package or utility already used by the file.
+- Dify semantic tokens and Tailwind v4 utilities.
+- Existing component variants before one-off class forks.
+- Primitive selectors such as `data-disabled:*`, `data-checked:*`, `data-highlighted:*`, `group-data-*`, `peer-data-*`, and `has-[:focus-visible]` before adding React state or boolean props solely for styling.
+- Component-level variants, semantic tokens, and normal cascade/order before `!` overrides. Use `!` only for a contained compatibility override that cannot be expressed through the component API or local selector structure.

-When writing components, always place the incoming `className` prop after the component’s own class values so that downstream consumers can override or extend the styling. This keeps your component’s defaults but still lets external callers change or remove specific styles.
+## Imports

-Example:
+Flag:

-```tsx
-import { cn } from '@/utils/classnames'
+- Barrel imports from `@langgenius/dify-ui`; consumers must use subpath exports.
+- New overlay imports from legacy `@/app/components/base/modal`, `dialog`, or `drawer`.
+- Cross-feature imports that bypass explicit top-level public files.
+- Direct imports from generated/internal implementation files when a feature contract already exposes the intended surface.

-const Button = ({ className }) => {
-  return <div className={cn('bg-primary-600', className)}></div>
-}
-```
+## Copy And i18n
+
+Flag:
+
+- User-facing hardcoded strings in `web/`.
+- Translation namespace drift, especially using unrelated module namespaces for local feature copy.
+- Generic button labels like `Continue` where the action is specific.
+- Error messages that state only the failure and not the next step.
+
+Use feature-local translation keys by default. Alias only when crossing namespaces.
@@ -0,0 +1,85 @@
+# Component Architecture Rules
+
+Use these rules for React component structure, ownership, state, props, effects, and module organization.
+
+## Ownership
+
+Flag:
+
+- State, query, mutation, or handlers hoisted above the lowest component that actually uses them.
+- Parent components owning row/item actions that do not coordinate a workflow.
+- Prop drilling through multiple pass-through layers.
+- A page/tab-level section component becoming the data owner without needing a shared snapshot or shared loading/error/empty UI.
+- Feature code promoted to shared only because it appears once or might be reused later.
+
+Accept repeated TanStack Query calls in siblings when each component independently consumes the data. Cache deduplication is not a reason to hoist by itself.
+
+## Component Boundaries
+
+Flag:
+
+- Shallow wrappers that only rename props or hide the real primitive.
+- Extra DOM wrappers that do not provide layout, semantics, accessibility, state ownership, or library integration.
+- Dialog/dropdown/popover hidden surfaces that obscure the parent flow when they should be extracted into a small local component.
+- Business forms, menu bodies, or one-off helpers moved away from their owner without reuse or semantic value.
+
+Prefer colocated components split by actual data and state needs.
+
+## Bad Component Design Patterns
+
+Flag:
+
+- Components that mix data fetching, mutation side effects, popup state, form validation, layout, and row rendering without a clear owner.
+- Generic components with many boolean props that encode one feature's workflow.
+- A shared component that imports feature-specific copy, routes, or API contracts.
+- A feature component that accepts pre-rendered fragments only to avoid placing ownership correctly.
+- A child component that receives both raw server data and separately derived flags for the same concept.
+- A wrapper that changes accessible semantics of the primitive it wraps.
+- A component that exposes controlled props but still keeps a competing private state for the same value.
+- A component that cannot render empty, loading, or missing optional API fields without caller-side preprocessing.
+
+## Props And Types
+
+Flag:
+
+- `React.FC` / `FC`.
+- Default exports outside framework-required files.
+- Named `Props` types for trivial one-off props where inline typing is clearer.
+- Props named by UI implementation instead of domain/API role.
+- API data converted too early or under a generic name that breaks traceability.
+- Callers duplicating fallback checks that the lowest rendering component already handles.
+
+Prefer top-level `function` declarations for components and module helpers. Use arrow functions for callbacks and local lambdas.
+
+## Effects
+
+Flag effects that:
+
+- Transform props/state for rendering.
+- Copy one state value into another representing the same concept.
+- Handle user actions that belong in event handlers.
+- Reset state from props when a keyed reset, stable ID, or render-time derivation would work.
+- Fetch data that belongs in framework APIs or TanStack Query.
+
+If an effect remains, it must synchronize with a named external system: browser API, subscription, timer, analytics-on-visibility, non-React widget, or imperative DOM integration.
+
+## State Modeling
+
+Flag:
+
+- Storing derived booleans, disabled flags, default tabs, or loading labels that can be calculated from current query/feature state.
+- Local state used to fake server data or generated contract fields.
+- UI state persisted to localStorage when it is live app state.
+- Feature-local mock shells wired to unrelated existing APIs before the real API is confirmed.
+
+Prefer render-time derivation. Keep true local state for user choices, transient input, controlled popups, and feature UI state that has no server source.
+
+## Navigation
+
+Flag:
+
+- Imperative router navigation for ordinary links.
+- Button semantics used for navigation.
+- Navigation state hidden in component state when URL state is required for shareable filters, tabs, or pagination.
+
+Use `Link` for normal navigation. Use router APIs for mutation success, guarded redirects, command flows, or form submission side effects.
@@ -0,0 +1,74 @@
+# Data, Query, And Contract Rules
+
+Use these rules for generated contracts, TanStack Query, mutations, auth/SSR boundaries, URL state, and client persistence.
+
+## Generated Contracts
+
+Flag:
+
+- New legacy service/helper wrappers around generated `queryOptions()` or `mutationOptions()`.
+- Continuing to use deprecated contract operations when a ready generated contract exists.
+- Assuming a generated file means an operation is ready without checking deprecated markers, schema shape, and the actual UI consumer.
+- Re-declaring API DTOs in components.
+- Adding compatibility layers instead of migrating the pointed line and deleting the old layer.
+
+Use `web/contract/*` as the API shape source of truth. Follow existing `{ params, query?, body? }` input shape.
+
+## Queries
+
+Flag:
+
+- `enabled` used to hide missing required input instead of `input: skipToken`.
+- Fake fallback IDs or placeholder inputs used to force a query to run.
+- Query results copied into local state for rendering.
+- Shared query behavior such as invalidation, stale defaults, or retry rules reimplemented at call sites.
+- `prefetchQuery` treated as a hard gate or as returning data/errors to the caller.
+
+Use `useQuery(consoleQuery.xxx.queryOptions(...))` or `useQuery(marketplaceQuery.xxx.queryOptions(...))` directly unless a feature hook performs real orchestration.
+
+## Mutations
+
+Flag:
+
+- Deprecated `useInvalid` or `useReset`.
+- `mutateAsync` used without a need for Promise semantics.
+- Awaited mutations without `try/catch`.
+- Components owning shared cache invalidation that belongs in query defaults.
+- Optimistic updates that do not match current list/detail ownership.
+
+Use generated `mutationOptions()` directly when possible. Put shared cache behavior in `createTanstackQueryUtils(...experimental_defaults...)`.
+
+## SSR, Auth, And Route Boundaries
+
+Flag:
+
+- Request-time auth, setup, workspace role, or tenant decisions moved into static `next.config redirects()`.
+- Dynamic role gates depending on `workspaces.current` implemented as static path redirects.
+- Authorization logic depending on soft `prefetchQuery`.
+- Removing a client fallback before server API unavailable behavior is defined.
+- Global placeholder query contracts introduced to solve a route-local Suspense issue.
+- Branding-sensitive UI reading placeholder defaults without checking pending/placeholder state.
+
+Separate hard gates from soft prefetches. `fetchQuery` can be a server decision boundary; `prefetchQuery` is cache warmup.
+
+## Workspace And Tenant
+
+Flag:
+
+- Treating workspace switch as ordinary CRUD invalidation when the current app flow performs server switch plus full reload.
+- Query keys that omit workspace/tenant identity when the query truly varies by workspace and no full reload boundary applies.
+- Mixing `workspace_id` and `tenant_id` without tracing the current backend/API contract.
+
+Current Dify workspace switch should be reviewed as a tenant cache boundary first.
+
+## URL State And Local Storage
+
+Flag:
+
+- Shareable filters, tabs, pagination, selected panels, or search state hidden only in component state.
+- One-shot navigation signals modeled as subscribed persistent state.
+- Live app state stored in localStorage.
+- Direct `window.localStorage`, `globalThis.localStorage`, or raw storage calls in app code.
+- High-frequency interaction state persisted on every change instead of on commit/settle.
+
+Use URL state for shareable UI state, feature/Jotai/store state for live UI state, and `@/hooks/use-local-storage` only for low-frequency client-only preferences, dismissed notices, and UI defaults.
@@ -0,0 +1,22 @@
+# Dify Invariants
+
+Use these stable Dify-specific runtime rules in addition to the generic review packs.
+
+This file is not a place for active feature notes. Do not add rules for one branch, one PR, or a short-lived product decision such as a specific agent-v2, plugin, model-provider, or onboarding task. Keep a rule here only when all of these are true:
+
+- It is a stable Dify runtime invariant.
+- Generic React, TypeScript, accessibility, dify-ui, query, or performance rules would not catch it.
+- The failure mode is concrete enough to produce a file-line review finding.
+- The rule is likely to remain valid across normal feature work.
+
+## Workflow Nodes And RAG Pipe
+
+Flag:
+
+- Node components under `web/app/components/workflow/nodes/[nodeName]/node.tsx` importing workflow store hooks that are unavailable in RAG Pipe template rendering.
+- Node UI relying on provider context that is not mounted in every rendering surface.
+- Store reads in render where React Flow `useNodes` / `useEdges` provide the actual node/edge source.
+
+Known failure mode: workflow node components can also render while creating a RAG Pipe from a template. In that context there may be no workflowStore provider, causing a blank screen.
+
+Prefer React Flow hooks for node/edge UI consumption. Use store APIs only where the provider is guaranteed and the code path is workflow-only.
@@ -0,0 +1,123 @@
+# Dify UI Rules
+
+Use these rules whenever a review touches `packages/dify-ui/` or code consuming `@langgenius/dify-ui/*`.
+
+Before finalizing findings for those files, read the current local docs that apply:
+
+- `packages/dify-ui/README.md`
+- `packages/dify-ui/AGENTS.md`
+- `web/docs/overlay.md` for floating UI
+- `packages/dify-ui/src/<primitive>/index.tsx` for the primitive being changed or consumed
+
+## Package Boundary
+
+Flag in `packages/dify-ui`:
+
+- Imports from `web/`.
+- Dependencies on Next.js, i18n, ky, Jotai, Zustand, TanStack Query, oRPC, or business APIs.
+- Business-specific component behavior that belongs in `web/`.
+- Multiple unrelated primitives in one component folder.
+
+`packages/dify-ui` is a primitive layer: Base UI headless components + `cva` + `cn` + Dify design tokens.
+
+## Imports And Exports
+
+Flag:
+
+- Consumer imports from `@langgenius/dify-ui` without a subpath.
+- Missing `package.json#exports` entry for a new primitive.
+- Internal package imports using workspace subpaths instead of relative paths.
+- Exported props using internal-only types that consumers cannot import from the component subpath.
+
+Consumers use subpath exports such as `@langgenius/dify-ui/button`.
+
+## Props And State
+
+Flag:
+
+- Flattened props where related values need a discriminated union, such as `value` / `defaultValue`, `multiple` / `value`, or `clearable` / `onChange`.
+- React state used only to mirror Base UI state for class names.
+- JavaScript conditional class logic for visual states that the Dify UI/Base UI primitive already exposes through `data-*` attributes or CSS variables.
+- Controlled props added when uncontrolled DOM state or CSS variables would be enough.
+- Thin wrappers that rename Base UI parts without adding semantics.
+
+Prefer Base UI/Dify UI data attributes and CSS variables for visual state: `data-open`, `data-checked`, `data-disabled`, `data-highlighted`, `data-popup-open`, `group-data-*`, `peer-data-*`, `has-[:focus-visible]`, and primitive CSS variables such as anchor width or transform origin. Use JS conditional classes for product/business state that the primitive does not expose.
+
+## Forms
+
+Flag:
+
+- Form-like UI using unrelated `Input` and `Button` pieces without a submit boundary.
+- Text-like fields not composed through `FieldRoot`, `FieldLabel`, and `FieldControl` when using Dify UI form semantics.
+- Select fields using `FieldLabel` instead of `SelectLabel`.
+- Slider fields using a generic label instead of `SliderLabel`.
+- Checkbox/radio groups missing `FieldsetRoot` and `FieldsetLegend`.
+- Field errors or descriptions rendered without `FieldDescription` / `FieldError` relationships.
+
+`Form` is the submit boundary. Dify UI form primitives are not a form state-management framework; business validation and schema-driven behavior belong in `web/`.
+
+## Overlay Contract
+
+Flag:
+
+- Legacy web overlay imports in new or modified code.
+- Manual portals around Dify UI overlay primitives.
+- Call-site `z-*` overrides on overlays.
+- Missing root `isolation: isolate` assumptions when debugging overlay stacking.
+- Repeated backdrop, z-index, or portal chrome at call sites.
+- Tooltip used for infotips, long text, or interactive content.
+
+All Dify UI body-portalled overlays use `z-50`. Toast uses `z-60`. DOM order handles stacking between overlays.
+
+## Primitive Selection
+
+Flag:
+
+- `Tabs` used for simple mode/filter/view selection where `SegmentedControl` is the semantic primitive.
+- `SegmentedControl` used where `tablist` / `tabpanel` semantics are required.
+- `Select` used for searchable or free-form input.
+- `Combobox` used for unrestricted search text where no selected option is remembered.
+- `Autocomplete` used for closed-list selection.
+- Tooltip or PreviewCard used for content that must be reachable on touch or by screen readers.
+
+Use:
+
+- `Autocomplete` for free-form text with optional suggestions.
+- `Combobox` for searchable selected values from a collection.
+- `Select` for closed, scannable option sets.
+- `Popover` for infotips, help text, rich content, or interactions.
+
+## Bad Usage Patterns To Flag
+
+Flag:
+
+- Styling a raw Base UI primitive directly in `web/` when a Dify UI primitive exists.
+- Wrapping a Dify UI primitive in a feature component that hides its label, error, disabled, or focus contract.
+- Replacing a semantic primitive with a generic `div` plus classes to match a screenshot.
+- Using `Tooltip` because it is visually convenient when the content is actually help text or needs touch access.
+- Adding a `z-*` override to make a child popup appear over a parent dialog.
+- Adding a new app-level wrapper around Dialog, Drawer, Popover, Select, or Combobox that repeats portal/backdrop/positioner logic.
+- Using dify-ui `Input` as a drop-in replacement for legacy inputs that include search, clear, copy, unit, localized placeholder, or number normalization behavior.
+- Building a form row from loose text and controls instead of the matching Field/Form primitives.
+- Adding component state only to style `data-open`, `data-checked`, `data-disabled`, or highlighted states that Base UI already exposes.
+- Passing booleans down only so children can toggle classes already expressible with primitive `data-*` selectors.
+
+## Tokens, Radius, And Styling
+
+Flag:
+
+- `radius-*` class names.
+- Custom Tailwind `borderRadius` extension for Figma radius values.
+- Generic colors where semantic Dify tokens exist.
+- Hardcoded design values where Dify tokens, component variants, or documented Figma radius mappings exist.
+- `!` important modifiers used to fight primitive styles instead of fixing the variant, selector, or component composition.
+- Manual class strings that duplicate primitive variants.
+- `min-w-(--anchor-width)` on picker popups when it defeats viewport clamping.
+
+Use the Figma radius mapping from `packages/dify-ui/AGENTS.md`; for example `--radius/sm` maps to `rounded-md`, and `--radius/md` maps to `rounded-lg`.
+
+Use `!` only for a tightly scoped compatibility override after confirming the primitive API, data attributes, and selector structure cannot express the state.
+
+## Focus Details
+
+Flag focus rings attached to the wrong element. For example, Base UI `Slider.Thumb` focuses an internal `input[type=range]`, so the visible thumb wrapper needs `has-[:focus-visible]` rather than direct wrapper `focus-visible`.
@@ -1,45 +1,78 @@
-# Rule Catalog — Performance
+# Performance Rules

-## React Flow data usage
+Review performance only where there is realistic impact. Do not request `memo`, `useMemo`, `useCallback`, virtualization, or caching as style preferences.

-IsUrgent: True
-Category: Performance
+## Async Waterfalls

-### Description
+Flag:

-When rendering React Flow, prefer `useNodes`/`useEdges` for UI consumption and rely on `useStoreApi` inside callbacks that mutate or read node/edge state. Avoid manually pulling Flow data outside of these hooks.
+- Awaiting remote feature flags or fetches before checking cheap synchronous conditions.
+- Sequential awaits for independent operations.
+- API routes or server components starting requests late when they could start early.
+- Nested per-item fetches running serially when each item can fetch in parallel.
+- Suspense boundaries that force the whole page to wait when a lower boundary could stream or isolate loading.

-## Complex prop stability
+Prefer `Promise.all` for independent work and branch-local awaits for conditionally needed data.

-IsUrgent: False
-Category: Performance
+## Bundle Size

-### Description
+Flag:

-Only require stable object, array, or map props when there is a clear reason: the child is memoized, the value participates in effect/query dependencies, the value is part of a stable-reference API contract, or profiling/local behavior shows avoidable re-renders. Do not request `useMemo` for every inline object by default; `how-to-write-component` treats memoization as a targeted optimization.
+- Barrel imports from heavy libraries or `@langgenius/dify-ui`.
+- Dynamic paths that prevent static trace analysis.
+- Heavy components loaded eagerly when hidden behind a dialog, tab, command, or feature activation.
+- Analytics, logging, editor, visualization, or third-party SDK code loaded before it is needed.
+- Feature-local optional modules imported at top level only for rare flows.

-Update this file when adding, editing, or removing Performance rules so the catalog remains accurate.
+Use direct imports and `next/dynamic` where the user-visible path benefits.

-Risky:
+## Server Rendering

-```tsx
-<HeavyComp
-    config={{
-        provider: ...,
-        detail: ...
-    }}
-/>
-```
+Flag:

-Better when stable identity matters:
+- Request-specific mutable state stored at module scope in SSR/RSC paths.
+- Large duplicate data serialized across RSC/client boundaries.
+- Static I/O repeated per request when it could be hoisted safely.
+- Cross-request cache without a bounded invalidation strategy.
+- Server actions lacking API-route-equivalent auth checks.

-```tsx
-const config = useMemo(() => ({
-    provider: ...,
-    detail: ...
-}), [provider, detail]);
+Use request-scoped deduplication such as `React.cache()` when repeated server reads in one request are the problem.

-<HeavyComp
-    config={config}
-/>
-```
+## Re-rendering
+
+Flag:
+
+- Effects or subscriptions reading broad state when a derived boolean or narrower selector is enough.
+- Components defined inside components.
+- Derived rendering state stored in state/effects.
+- Non-primitive default props recreated for memoized children.
+- Expensive work recalculated on every render where it affects real interaction cost.
+- High-frequency transient values stored in state when refs or CSS variables would avoid render loops.
+
+Do not flag simple primitive expressions wrapped or not wrapped in `useMemo`; prefer no memo for simple work.
+
+Require stable object/array/function identity only when:
+
+- The child is memoized and identity affects renders.
+- The value is an effect/query dependency.
+- A library API requires stable references.
+- Profiling or local behavior shows avoidable re-rendering.
+
+## DOM, Lists, And Rendering
+
+Flag:
+
+- Layout reads in render (`getBoundingClientRect`, `offset*`, `scrollTop`).
+- Interleaved DOM reads/writes that can cause layout thrashing.
+- Large lists rendering without virtualization, pagination, or `content-visibility`.
+- SVG/animation code animating expensive properties when transform/opacity would work.
+- `transition-all`.
+- Long-running non-critical browser work performed immediately instead of idle/deferred scheduling.
+
+## React Flow
+
+For workflow React Flow components, keep this Dify-specific rule:
+
+- UI consumption should use React Flow hooks such as `useNodes` / `useEdges`.
+- Callback-only reads or mutations can use `useStoreApi`.
+- Node components under `web/app/components/workflow/nodes/[nodeName]/node.tsx` must not depend on workflow stores that are absent in RAG Pipe template rendering.
@@ -0,0 +1,72 @@
+# Testing Review Rules
+
+Use these rules when reviewing test files, testability of changed code, or risky frontend changes that should have tests.
+
+## Missing Coverage
+
+Flag missing tests when the change affects:
+
+- User-visible behavior, navigation, form submission, validation, permissions, or loading/error/empty states.
+- Query/mutation cache behavior.
+- Accessibility-critical behavior such as labels, keyboard flow, focus, disabled state, or popup reachability.
+- URL state parsing/serialization.
+- Storage persistence or one-shot signals.
+- Regression-prone workflow or generated contract migration paths.
+
+Do not request tests for purely mechanical renames or styling-only changes unless the styling affects layout, focus, or interaction.
+
+## Selectors
+
+Flag:
+
+- `getByTestId` used where role, label, text, placeholder, landmark, or scoped dialog/menu queries are available.
+- Production `data-testid` added only to satisfy tests.
+- Assertions against decorative icons rather than the named control.
+- Tests that cannot find controls semantically but leave broken markup unchanged.
+
+Prefer `getByRole` with accessible name, then `getByLabelText`, `getByPlaceholderText`, `getByText`, and `within(...)`.
+
+## Mocking
+
+Flag:
+
+- Mocking `@langgenius/dify-ui/*` primitives.
+- Mocking `@/app/components/base/*` components when the real component is practical.
+- Mocking sibling or child components in the same directory for integration behavior.
+- Mocks that do not match the real component's conditional rendering.
+- Module-level mock state not reset in `beforeEach`.
+- `vi.clearAllMocks()` in `afterEach` instead of `beforeEach`.
+
+Use real project components for integration behavior. Mock APIs, `next/navigation`, browser shims, or complex providers only when setup would dominate the test.
+
+## Behavior
+
+Flag:
+
+- Tests inspecting implementation details instead of user-observable behavior.
+- Assertions that hardcode brittle copy when pattern matching or semantic roles would express behavior better.
+- Fake timers used without real timing behavior.
+- Async assertions missing `await`, `findBy*`, or `waitFor`.
+- Test data missing required fields because inline partial objects bypass real types.
+
+Use typed factory functions with complete defaults and partial overrides.
+
+## URL State
+
+For `nuqs` or query-state hooks, flag tests that:
+
+- Mock URL state when URL synchronization is the behavior under review.
+- Do not test parser serialize/parse round trips for custom parsers.
+- Do not assert default-clearing behavior when defaults should be removed from the URL.
+
+Prefer shared `NuqsTestingAdapter` helpers when available.
+
+## Organization
+
+Flag:
+
+- Component/hook/util tests outside sibling `__tests__/` directories.
+- Directory-level reviews that test only `index.tsx` while other files in scope contain behavior.
+- Large test files with repeated setup that should use local builders.
+
+When a component is very complex, prefer a refactor finding before asking for exhaustive tests.
@@ -0,0 +1,33 @@
+---
+name: karpathy-guidelines
+description: Lightweight coding guardrails for making focused, simple, and verifiable changes in this repo. Use for all coding work.
+---
+
+# Karpathy Guidelines
+
+Use this skill whenever you touch code in this repository.
+
+## Principles
+
+- Keep the change small and directly tied to the user request.
+- Prefer the simplest implementation that fits the existing codebase.
+- Read the nearby code first, then match its patterns.
+- Avoid unrelated refactors, broad rewrites, or style churn.
+- Preserve existing behavior unless the user explicitly asked to change it.
+- Treat regressions as a signal to narrow the change, not to add workaround layers.
+
+## Workflow
+
+1. Inspect the current implementation and tests around the change.
+2. Make the smallest coherent edit.
+3. Add or update focused tests when the behavior changes or the risk is non-trivial.
+4. Run the narrowest relevant verification first.
+5. Report exactly what was verified and anything left unverified.
+
+## Review Checklist
+
+- Does this change solve the stated problem without expanding scope?
+- Did it preserve existing route/component/data-flow semantics?
+- Are new abstractions justified by real complexity?
+- Are tests focused on the behavior that could regress?
+- Are unrelated files and generated artifacts left alone?
@@ -15,6 +15,10 @@
 # Agents
 /.agents/skills/ @hyoban

+# Packages
+/packages/ @lyzno1
+/packages/contracts/ @crazywoola @laipz8200
+
 # Docs
 /docs/ @crazywoola

@@ -143,6 +147,14 @@
 # Frontend
 /web/ @iamjoel

+# Frontend - Platform and Features
+/web/config/ @lyzno1
+/web/contract/ @lyzno1
+/web/env.ts @lyzno1
+/web/features/ @lyzno1
+/web/hooks/ @lyzno1
+/web/scripts/gen-icons.mjs @lyzno1
+
 # Frontend - Web Tests
 /.github/workflows/web-tests.yml @iamjoel

@@ -253,7 +265,6 @@
 /web/utils/time.ts @iamjoel @zxhlyh
 /web/utils/format.ts @iamjoel @zxhlyh
 /web/utils/clipboard.ts @iamjoel @zxhlyh
-/web/hooks/use-document-title.ts @iamjoel @zxhlyh

 # Frontend - Billing and Education
 /web/app/components/billing/ @iamjoel @zxhlyh
@@ -2,87 +2,165 @@ name: CLI Release

 on:
  workflow_dispatch:
-  push:
-    tags:
-      - 'difyctl-v*'
+    inputs:
+      release_tag:
+        description: Dify release tag to attach difyctl assets to (blank = latest stable)
+        required: false
+        type: string
+  workflow_call:
+    inputs:
+      release_tag:
+        description: Dify release tag to attach difyctl assets to (blank = latest stable)
+        required: false
+        type: string
+  release:
+    types: [released]

 concurrency:
-  group: cli-release-${{ github.ref }}
-  cancel-in-progress: true
+  group: difyctl-release
+  cancel-in-progress: false

 jobs:
-  release:
-    name: build standalone binaries (all targets)
+  validate:
+    name: validate manifest + resolve target Dify release
    runs-on: depot-ubuntu-24.04
    if: github.repository == 'langgenius/dify'
+    permissions:
+      contents: read
+    defaults:
+      run:
+        shell: bash
+        working-directory: ./cli
+    outputs:
+      dify_tag: ${{ steps.resolve.outputs.dify_tag }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+
+      - name: Export manifest to env
+        run: node scripts/release-naming.mjs github-env >> "$GITHUB_ENV"
+
+      - name: Validate manifest
+        run: scripts/release-validate-manifest.sh
+
+      - name: Resolve target Dify release
+        id: resolve
+        env:
+          GH_TOKEN: ${{ github.token }}
+          EVENT_TAG: ${{ github.event.release.tag_name }}
+          INPUT_TAG: ${{ inputs.release_tag }}
+        run: |
+          if [ -n "$EVENT_TAG" ]; then
+              tag="$EVENT_TAG"
+          elif [ -n "$INPUT_TAG" ]; then
+              tag="$INPUT_TAG"
+          else
+              tag="$(gh api "repos/${GITHUB_REPOSITORY}/releases/latest" --jq .tag_name)"
+          fi
+          if [ -z "$tag" ]; then
+              echo "::error::could not resolve a target Dify release tag"
+              exit 1
+          fi
+          if ! gh release view "$tag" --repo "$GITHUB_REPOSITORY" >/dev/null 2>&1; then
+              echo "::error::target Dify release ${tag} not found"
+              exit 1
+          fi
+          echo "dify_tag=${tag}" >> "$GITHUB_OUTPUT"
+          echo "::notice::target Dify release ${tag}"
+
+      - name: Compatibility check
+        env:
+          DIFY_TAG: ${{ steps.resolve.outputs.dify_tag }}
+        run: node scripts/release-naming.mjs compat-check "$DIFY_TAG"
+
+      - name: Reject duplicate difyctl version
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          if gh api "repos/${GITHUB_REPOSITORY}/git/ref/tags/${difyctlTag}" >/dev/null 2>&1; then
+              echo "::error::difyctl ${version} already released (tag ${difyctlTag} exists); bump cli/package.json version"
+              exit 1
+          fi
+
+  release:
+    name: build + attach standalone binaries (all targets)
+    needs: validate
+    runs-on: depot-ubuntu-24.04
    permissions:
      contents: write
    defaults:
      run:
        shell: bash
        working-directory: ./cli
-
+    env:
+      DIFY_TAG: ${{ needs.validate.outputs.dify_tag }}
    steps:
      - name: Checkout
        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
        with:
          persist-credentials: false
-          fetch-depth: 0
+          fetch-depth: 1
+
+      - name: Enable cross-arch native prebuilds
+        working-directory: ./
+        run: cat cli/scripts/cross-arch.pnpm.yaml >> pnpm-workspace.yaml

      - name: Setup web environment
        uses: ./.github/actions/setup-web

+      - name: Export manifest to env
+        run: node scripts/release-naming.mjs github-env >> "$GITHUB_ENV"
+
      - name: Setup Bun
        uses: oven-sh/setup-bun@4bc047ad259df6fc24a6c9b0f9a0cb08cf17fbe5 # v2.0.2
        with:
-          bun-version: latest
-
-      - name: Read cli/package.json
-        id: manifest
-        run: |
-          version=$(node -p "require('./package.json').version")
-          channel=$(node -p "require('./package.json').difyctl.channel")
-          minDify=$(node -p "require('./package.json').difyctl.compat.minDify")
-          maxDify=$(node -p "require('./package.json').difyctl.compat.maxDify")
-          {
-              echo "version=$version"
-              echo "channel=$channel"
-              echo "minDify=$minDify"
-              echo "maxDify=$maxDify"
-          } >> "$GITHUB_OUTPUT"
-
-      - name: Validate manifest
-        run: scripts/release-validate-manifest.sh
-
-      - name: Install cross-arch native prebuilds
-        # Re-installs node_modules with every @napi-rs/keyring platform variant
-        # so `bun build --compile` can embed the right .node into each target.
-        working-directory: ./
-        run: NPM_CONFIG_USERCONFIG="$PWD/cli/scripts/cross-arch.npmrc" pnpm install --frozen-lockfile
+          bun-version-file: cli/.bun-version

      - name: Compile standalone binaries (all targets)
-        env:
-          CLI_VERSION: ${{ steps.manifest.outputs.version }}
-          DIFYCTL_CHANNEL: ${{ steps.manifest.outputs.channel }}
-          DIFYCTL_MIN_DIFY: ${{ steps.manifest.outputs.minDify }}
-          DIFYCTL_MAX_DIFY: ${{ steps.manifest.outputs.maxDify }}
        run: |
          DIFYCTL_COMMIT="$(git rev-parse HEAD)" \
          DIFYCTL_BUILD_DATE="$(git log -1 --format=%cI HEAD)" \
            pnpm build:bin

      - name: Generate sha256 checksum file
-        env:
-          CLI_VERSION: ${{ steps.manifest.outputs.version }}
        run: scripts/release-write-checksums.sh

-      - name: Publish GitHub Release
-        uses: softprops/action-gh-release@72f2c25fcb47643c292f7107632f7a47c1df5cd8 # v2.3.2
-        with:
-          tag_name: difyctl-v${{ steps.manifest.outputs.version }}
-          name: difyctl ${{ steps.manifest.outputs.version }}
-          prerelease: ${{ steps.manifest.outputs.channel != 'stable' }}
-          generate_release_notes: true
-          fail_on_unmatched_files: true
-          files: |
-            cli/dist/bin/difyctl-v*
+      - name: Attach difyctl assets to Dify release
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          gh release upload "$DIFY_TAG" dist/bin/${tagPrefix}* \
+              --repo "$GITHUB_REPOSITORY" --clobber
+
+      - name: Prune stale difyctl assets
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          new_set="$(cd dist/bin && ls ${tagPrefix}*)"
+          gh release view "$DIFY_TAG" --repo "$GITHUB_REPOSITORY" \
+              --json assets --jq '.assets[].name' \
+            | { grep -E "^${tagPrefix}" || true; } \
+            | while IFS= read -r name; do
+                if ! printf '%s\n' "$new_set" | grep -qxF -- "$name"; then
+                    echo "::notice::pruning stale asset ${name}"
+                    gh release delete-asset "$DIFY_TAG" "$name" \
+                        --repo "$GITHUB_REPOSITORY" --yes
+                fi
+              done
+
+      - name: Create provenance tag
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          ref="refs/tags/${difyctlTag}"
+          sha="$(git rev-parse HEAD)"
+          status="$(gh api -X POST "repos/${GITHUB_REPOSITORY}/git/refs" \
+                -f ref="$ref" -f sha="$sha" --silent --include 2>/dev/null \
+                | awk 'NR==1 {print $2; exit}' || true)"
+          case "$status" in
+              201) echo "::notice::created ${ref}" ;;
+              422) echo "::notice::tag ${ref} already exists; skipping (immutable)" ;;
+              *)   echo "::error::provenance tag ${ref} not created (HTTP ${status:-unknown})"; exit 1 ;;
+          esac
@@ -37,6 +37,10 @@ jobs:
      - name: Setup web environment
        uses: ./.github/actions/setup-web

+      - name: Validate release manifest
+        if: matrix.os == 'depot-ubuntu-24.04'
+        run: scripts/release-validate-manifest.sh
+
      - name: CI pipeline (typecheck, lint, coverage, build)
        run: pnpm ci

@@ -11,6 +11,7 @@ from .data_migration import (
    migration_data_wizard,
 )
 from .plugin import (
+    backfill_plugin_auto_upgrade,
    extract_plugins,
    extract_unique_plugins,
    install_plugins,
@@ -49,6 +50,7 @@ from .vector import (
 __all__ = [
    "add_qdrant_index",
    "archive_workflow_runs",
+    "backfill_plugin_auto_upgrade",
    "clean_expired_messages",
    "clean_workflow_runs",
    "cleanup_orphaned_draft_variables",
@@ -1,10 +1,11 @@
 import json
 import logging
+import time
 from typing import Any, cast

 import click
 from pydantic import TypeAdapter
-from sqlalchemy import delete, select
+from sqlalchemy import delete, func, select
 from sqlalchemy.engine import CursorResult

 from configs import dify_config
@@ -15,11 +16,13 @@ from core.plugin.plugin_service import PluginService
 from core.tools.utils.system_encryption import encrypt_system_params
 from extensions.ext_database import db
 from models import Tenant
+from models.account import TenantPluginAutoUpgradeStrategy
 from models.oauth import DatasourceOauthParamConfig, DatasourceProvider
 from models.provider_ids import DatasourceProviderID, ToolProviderID
 from models.source import DataSourceApiKeyAuthBinding, DataSourceOauthBinding
 from models.tools import ToolOAuthSystemClient
 from services.plugin.data_migration import PluginDataMigration
+from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService
 from services.plugin.plugin_migration import PluginMigration

 logger = logging.getLogger(__name__)
@@ -402,6 +405,110 @@ def migrate_data_for_plugin():
    click.echo(click.style("Migrate data for plugin completed.", fg="green"))


+def _candidate_auto_upgrade_strategy_tenant_ids_stmt(limit: int | None = None):
+    category_count = len(TenantPluginAutoUpgradeStrategy.PluginCategory)
+    stmt = (
+        select(TenantPluginAutoUpgradeStrategy.tenant_id)
+        .group_by(TenantPluginAutoUpgradeStrategy.tenant_id)
+        .having(func.count(func.distinct(TenantPluginAutoUpgradeStrategy.category)) < category_count)
+        .order_by(TenantPluginAutoUpgradeStrategy.tenant_id)
+    )
+
+    if limit is not None:
+        stmt = stmt.limit(limit)
+
+    return stmt
+
+
+def _count_auto_upgrade_strategy_tenant_ids(limit: int | None) -> int:
+    candidate_stmt = _candidate_auto_upgrade_strategy_tenant_ids_stmt(limit).subquery()
+    return db.session.scalar(select(func.count()).select_from(candidate_stmt)) or 0
+
+
+def _iter_auto_upgrade_strategy_tenant_ids(limit: int | None):
+    stmt = _candidate_auto_upgrade_strategy_tenant_ids_stmt(limit).execution_options(yield_per=1000)
+    yield from db.session.scalars(stmt)
+
+
+@click.command(
+    "backfill-plugin-auto-upgrade",
+    help="Backfill category-scoped plugin auto-upgrade strategies and normalize plugin lists.",
+)
+@click.option("--tenant-id", multiple=True, help="Tenant ID to backfill. Can be passed multiple times.")
+@click.option("--limit", type=int, default=None, help="Maximum number of candidate tenants to process.")
+@click.option("--batch-size", type=int, default=500, show_default=True, help="Progress reporting batch size.")
+@click.option("--dry-run", is_flag=True, help="Only print candidate tenant count.")
+def backfill_plugin_auto_upgrade(
+    tenant_id: tuple[str, ...],
+    limit: int | None,
+    batch_size: int,
+    dry_run: bool,
+):
+    """
+    Backfill historical auto-upgrade strategies after the category column exists.
+
+    Missing category rows are created from the tenant's tool/default row. Pure default
+    strategies become latest for model plugins and fix-only for all other categories.
+    Tenants with include/exclude plugin IDs are split
+    by installed plugin category using plugin daemon metadata.
+    """
+    start_at = time.perf_counter()
+    candidate_count = len(tenant_id) if tenant_id else _count_auto_upgrade_strategy_tenant_ids(limit)
+    click.echo(click.style(f"Found {candidate_count} candidate tenants.", fg="yellow"))
+
+    if dry_run:
+        elapsed = time.perf_counter() - start_at
+        click.echo(click.style(f"Dry run completed. elapsed={elapsed:.2f}s", fg="green"))
+        return
+
+    tenant_ids = list(tenant_id) if tenant_id else _iter_auto_upgrade_strategy_tenant_ids(limit)
+
+    backfilled_count = 0
+    created_count = 0
+    normalized_count = 0
+    skipped_count = 0
+    failed_count = 0
+    for index, current_tenant_id in enumerate(tenant_ids, start=1):
+        try:
+            result = PluginAutoUpgradeService.backfill_strategy_categories(
+                current_tenant_id,
+            )
+        except Exception as e:
+            failed_count += 1
+            click.echo(click.style(f"Failed tenant {current_tenant_id}: {str(e)}", fg="red"))
+            continue
+
+        if result.created_count > 0:
+            backfilled_count += 1
+            created_count += result.created_count
+        elif not result.normalized:
+            skipped_count += 1
+        if result.normalized:
+            normalized_count += 1
+
+        if batch_size > 0 and index % batch_size == 0:
+            click.echo(
+                click.style(
+                    f"Processed {index}/{candidate_count} tenants. "
+                    f"backfilled={backfilled_count}, created_rows={created_count}, "
+                    f"normalized={normalized_count}, skipped={skipped_count}, failed={failed_count}, "
+                    f"elapsed={time.perf_counter() - start_at:.2f}s",
+                    fg="yellow",
+                )
+            )
+
+    elapsed = time.perf_counter() - start_at
+    click.echo(
+        click.style(
+            f"Backfill plugin auto-upgrade strategy categories completed. "
+            f"backfilled={backfilled_count}, created_rows={created_count}, "
+            f"normalized={normalized_count}, skipped={skipped_count}, failed={failed_count}, "
+            f"elapsed={elapsed:.2f}s",
+            fg="green",
+        )
+    )
+
+
@click.command("extract-plugins", help="Extract plugins.")
@click.option("--output_file", prompt=True, help="The file to store the extracted plugins.", default="plugins.jsonl")
@click.option("--workers", prompt=True, help="The number of workers to extract plugins.", default=10)
@@ -122,6 +122,7 @@ from .explore import (
    saved_message,
    trial,
 )
+from .snippets import snippet_workflow, snippet_workflow_draft_variable
 from .socketio import workflow as socketio_workflow

 # Import tag controllers
@@ -137,6 +138,7 @@ from .workspace import (
    model_providers,
    models,
    plugin,
+    snippets,
    tool_providers,
    trigger_providers,
    workspace,
@@ -212,6 +214,9 @@ __all__ = [
    "saved_message",
    "setup",
    "site",
+    "snippet_workflow",
+    "snippet_workflow_draft_variable",
+    "snippets",
    "socketio_workflow",
    "spec",
    "statistic",
@@ -1,8 +1,9 @@
 import logging
 import re
 import uuid
+from collections.abc import Sequence
 from datetime import datetime
-from typing import Any, Literal
+from typing import Any, Literal, cast

 from flask import request
 from flask_restx import Resource
@@ -14,7 +15,12 @@ from werkzeug.exceptions import BadRequest

 from controllers.common.fields import RedirectUrlResponse, SimpleResultResponse
 from controllers.common.helpers import FileInfo
-from controllers.common.schema import register_enum_models, register_response_schema_models, register_schema_models
+from controllers.common.schema import (
+    query_params_from_model,
+    register_enum_models,
+    register_response_schema_models,
+    register_schema_models,
+)
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model, with_session
 from controllers.console.workspace.models import LoadBalancingPayload
@@ -36,12 +42,12 @@ from core.trigger.constants import TRIGGER_NODE_TYPES
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from graphon.enums import WorkflowExecutionStatus
-from libs.helper import build_icon_url, to_timestamp
+from libs.helper import build_icon_url, dump_response, to_timestamp
 from libs.login import login_required
 from models import Account, App, DatasetPermissionEnum, Workflow
 from models.model import IconType
 from services.app_dsl_service import AppDslService
-from services.app_service import AppListParams, AppService, CreateAppParams
+from services.app_service import AppListParams, AppListSortBy, AppService, CreateAppParams, StarredAppListParams
 from services.enterprise.enterprise_service import EnterpriseService
 from services.entities.dsl_entities import ImportMode, ImportStatus
 from services.entities.knowledge_entities.knowledge_entities import (
@@ -64,16 +70,21 @@ register_enum_models(console_ns, IconType)

 _logger = logging.getLogger(__name__)
 _TAG_IDS_BRACKET_PATTERN = re.compile(r"^tag_ids\[(\d+)\]$")
+_CREATOR_IDS_BRACKET_PATTERN = re.compile(r"^creator_ids\[(\d+)\]$")
+AppListMode = Literal["completion", "chat", "advanced-chat", "workflow", "agent-chat", "agent", "channel", "all"]


-class AppListQuery(BaseModel):
+class AppListBaseQuery(BaseModel):
    page: int = Field(default=1, ge=1, le=99999, description="Page number (1-99999)")
    limit: int = Field(default=20, ge=1, le=100, description="Page size (1-100)")
-    mode: Literal["completion", "chat", "advanced-chat", "workflow", "agent-chat", "agent", "channel", "all"] = Field(
-        default="all", description="App mode filter"
+    mode: AppListMode = Field(default=cast(AppListMode, "all"), description="App mode filter")
+    sort_by: AppListSortBy = Field(
+        default="last_modified",
+        description="Sort apps by last modified, recently created, or earliest created",
    )
    name: str | None = Field(default=None, description="Filter by app name")
    tag_ids: list[str] | None = Field(default=None, description="Filter by tag IDs")
+    creator_ids: list[str] | None = Field(default=None, description="Filter by creator account IDs")
    is_created_by_me: bool | None = Field(default=None, description="Filter by creator")

    @field_validator("tag_ids", mode="before")
@@ -94,10 +105,37 @@ class AppListQuery(BaseModel):
        except ValueError as exc:
            raise ValueError("Invalid UUID format in tag_ids.") from exc

+    @field_validator("creator_ids", mode="before")
+    @classmethod
+    def validate_creator_ids(cls, value: list[str] | None) -> list[str] | None:
+        if not value:
+            return None
+
+        if not isinstance(value, list):
+            raise ValueError("Unsupported creator_ids type.")
+
+        items = [str(item).strip() for item in value if item and str(item).strip()]
+        if not items:
+            return None
+
+        try:
+            return [str(uuid.UUID(item)) for item in items]
+        except ValueError as exc:
+            raise ValueError("Invalid UUID format in creator_ids.") from exc
+
+
+class AppListQuery(AppListBaseQuery):
+    pass
+
+
+class StarredAppListQuery(AppListBaseQuery):
+    pass
+

 def _normalize_app_list_query_args(query_args: MultiDict[str, str]) -> dict[str, str | list[str]]:
    normalized: dict[str, str | list[str]] = {}
    indexed_tag_ids: list[tuple[int, str]] = []
+    indexed_creator_ids: list[tuple[int, str]] = []

    for key in query_args:
        match = _TAG_IDS_BRACKET_PATTERN.fullmatch(key)
@@ -105,12 +143,19 @@ def _normalize_app_list_query_args(query_args: MultiDict[str, str]) -> dict[str,
            indexed_tag_ids.extend((int(match.group(1)), value) for value in query_args.getlist(key))
            continue

+        match = _CREATOR_IDS_BRACKET_PATTERN.fullmatch(key)
+        if match:
+            indexed_creator_ids.extend((int(match.group(1)), value) for value in query_args.getlist(key))
+            continue
+
        value = query_args.get(key)
        if value is not None:
            normalized[key] = value

    if indexed_tag_ids:
        normalized["tag_ids"] = [value for _, value in sorted(indexed_tag_ids)]
+    if indexed_creator_ids:
+        normalized["creator_ids"] = [value for _, value in sorted(indexed_creator_ids)]

    return normalized

@@ -350,6 +395,7 @@ class AppPartial(ResponseModel):
    create_user_name: str | None = None
    author_name: str | None = None
    has_draft_trigger: bool | None = None
+    is_starred: bool = False

    @computed_field(return_type=str | None)  # type: ignore
    @property
@@ -419,12 +465,54 @@ class AppExportResponse(ResponseModel):
    data: str


+def _enrich_app_list_items(session: Session, *, apps: Sequence[App], tenant_id: str) -> None:
+    if FeatureService.get_system_features().webapp_auth.enabled:
+        app_ids = [str(app.id) for app in apps]
+        res = EnterpriseService.WebAppAuth.batch_get_app_access_mode_by_id(app_ids=app_ids)
+        if len(res) != len(app_ids):
+            raise BadRequest("Invalid app id in webapp auth")
+
+        for app in apps:
+            if str(app.id) in res:
+                app.access_mode = res[str(app.id)].access_mode
+
+    workflow_capable_app_ids = [str(app.id) for app in apps if app.mode in {"workflow", "advanced-chat"}]
+    draft_trigger_app_ids: set[str] = set()
+    if workflow_capable_app_ids:
+        draft_workflows = (
+            session.execute(
+                select(Workflow).where(
+                    Workflow.version == Workflow.VERSION_DRAFT,
+                    Workflow.app_id.in_(workflow_capable_app_ids),
+                    Workflow.tenant_id == tenant_id,
+                )
+            )
+            .scalars()
+            .all()
+        )
+        trigger_node_types = TRIGGER_NODE_TYPES
+        for workflow in draft_workflows:
+            node_id = None
+            try:
+                for node_id, node_data in workflow.walk_nodes():
+                    if node_data.get("type") in trigger_node_types:
+                        draft_trigger_app_ids.add(str(workflow.app_id))
+                        break
+            except Exception:
+                _logger.exception("error while walking nodes, workflow_id=%s, node_id=%s", workflow.id, node_id)
+                continue
+
+    for app in apps:
+        app.has_draft_trigger = str(app.id) in draft_trigger_app_ids
+
+
 register_enum_models(console_ns, RetrievalMethod, WorkflowExecutionStatus, DatasetPermissionEnum)
 register_response_schema_models(console_ns, RedirectUrlResponse, SimpleResultResponse)

 register_schema_models(
    console_ns,
    AppListQuery,
+    StarredAppListQuery,
    CreateAppPayload,
    UpdateAppPayload,
    CopyAppPayload,
@@ -468,7 +556,7 @@ register_schema_models(
 class AppListApi(Resource):
    @console_ns.doc("list_apps")
    @console_ns.doc(description="Get list of applications with pagination and filtering")
-    @console_ns.expect(console_ns.models[AppListQuery.__name__])
+    @console_ns.doc(params=query_params_from_model(AppListQuery))
    @console_ns.response(200, "Success", console_ns.models[AppPagination.__name__])
    @setup_required
    @login_required
@@ -484,8 +572,10 @@ class AppListApi(Resource):
            page=args.page,
            limit=args.limit,
            mode=args.mode,
+            sort_by=args.sort_by,
            name=args.name,
            tag_ids=args.tag_ids,
+            creator_ids=args.creator_ids,
            is_created_by_me=args.is_created_by_me,
        )

@@ -496,46 +586,7 @@ class AppListApi(Resource):
            empty = AppPagination(page=args.page, limit=args.limit, total=0, has_more=False, data=[])
            return empty.model_dump(mode="json"), 200

-        if FeatureService.get_system_features().webapp_auth.enabled:
-            app_ids = [str(app.id) for app in app_pagination.items]
-            res = EnterpriseService.WebAppAuth.batch_get_app_access_mode_by_id(app_ids=app_ids)
-            if len(res) != len(app_ids):
-                raise BadRequest("Invalid app id in webapp auth")
-
-            for app in app_pagination.items:
-                if str(app.id) in res:
-                    app.access_mode = res[str(app.id)].access_mode
-
-        workflow_capable_app_ids = [
-            str(app.id) for app in app_pagination.items if app.mode in {"workflow", "advanced-chat"}
-        ]
-        draft_trigger_app_ids: set[str] = set()
-        if workflow_capable_app_ids:
-            draft_workflows = (
-                session.execute(
-                    select(Workflow).where(
-                        Workflow.version == Workflow.VERSION_DRAFT,
-                        Workflow.app_id.in_(workflow_capable_app_ids),
-                        Workflow.tenant_id == current_tenant_id,
-                    )
-                )
-                .scalars()
-                .all()
-            )
-            trigger_node_types = TRIGGER_NODE_TYPES
-            for workflow in draft_workflows:
-                node_id = None
-                try:
-                    for node_id, node_data in workflow.walk_nodes():
-                        if node_data.get("type") in trigger_node_types:
-                            draft_trigger_app_ids.add(str(workflow.app_id))
-                            break
-                except Exception:
-                    _logger.exception("error while walking nodes, workflow_id=%s, node_id=%s", workflow.id, node_id)
-                    continue
-
-        for app in app_pagination.items:
-            app.has_draft_trigger = str(app.id) in draft_trigger_app_ids
+        _enrich_app_list_items(session, apps=app_pagination.items, tenant_id=current_tenant_id)

        pagination_model = AppPagination.model_validate(app_pagination, from_attributes=True)
        return pagination_model.model_dump(mode="json"), 200
@@ -571,6 +622,78 @@ class AppListApi(Resource):
        return app_detail.model_dump(mode="json"), 201


+@console_ns.route("/apps/starred")
+class StarredAppListApi(Resource):
+    @console_ns.doc("list_starred_apps")
+    @console_ns.doc(description="Get applications starred by the current account")
+    @console_ns.doc(params=query_params_from_model(StarredAppListQuery))
+    @console_ns.response(200, "Success", console_ns.models[AppPagination.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    @with_session(write=False)
+    @with_current_user_id
+    @with_current_tenant_id
+    def get(self, current_tenant_id: str, current_user_id: str, session: Session):
+        args = StarredAppListQuery.model_validate(_normalize_app_list_query_args(request.args))
+        params = StarredAppListParams(
+            page=args.page,
+            limit=args.limit,
+            mode=args.mode,
+            sort_by=args.sort_by,
+            name=args.name,
+            tag_ids=args.tag_ids,
+            creator_ids=args.creator_ids,
+            is_created_by_me=args.is_created_by_me,
+        )
+
+        app_pagination = AppService().get_paginate_starred_apps(current_user_id, current_tenant_id, params)
+        if not app_pagination:
+            empty = AppPagination(page=args.page, limit=args.limit, total=0, has_more=False, data=[])
+            return empty.model_dump(mode="json"), 200
+
+        _enrich_app_list_items(session, apps=app_pagination.items, tenant_id=current_tenant_id)
+
+        pagination_model = AppPagination.model_validate(app_pagination, from_attributes=True)
+        return pagination_model.model_dump(mode="json"), 200
+
+
+@console_ns.route("/apps/<uuid:app_id>/star")
+class AppStarApi(Resource):
+    @console_ns.doc("star_app")
+    @console_ns.doc(description="Star an application for the current account")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
+    @console_ns.response(404, "App not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    @with_current_user_id
+    @with_session
+    @get_app_model(mode=None)
+    def post(self, session: Session, current_user_id: str, app_model: App):
+        AppService.star_app(session, app=app_model, account_id=current_user_id)
+        return dump_response(SimpleResultResponse, {"result": "success"})
+
+    @console_ns.doc("unstar_app")
+    @console_ns.doc(description="Remove the current account's star from an application")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
+    @console_ns.response(404, "App not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    @with_current_user_id
+    @with_session
+    @get_app_model(mode=None)
+    def delete(self, session: Session, current_user_id: str, app_model: App):
+        AppService.unstar_app(session, app=app_model, account_id=current_user_id)
+        return dump_response(SimpleResultResponse, {"result": "success"})
+
+
@console_ns.route("/apps/<uuid:app_id>")
 class AppApi(Resource):
    @console_ns.doc("get_app_detail")
@@ -23,6 +23,7 @@ from controllers.console.wraps import (
    account_initialization_required,
    edit_permission_required,
    setup_required,
+    with_current_user,
    with_current_user_id,
 )
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
@@ -36,7 +37,7 @@ from core.helper.trace_id_helper import get_external_trace_id
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
-from libs.login import current_user, login_required
+from libs.login import login_required
 from models import Account
 from models.model import App, AppMode
 from services.app_generate_service import AppGenerateService
@@ -104,7 +105,8 @@ class CompletionMessageApi(Resource):
    @login_required
    @account_initialization_required
    @get_app_model(mode=AppMode.COMPLETION)
-    def post(self, app_model: App):
+    @with_current_user
+    def post(self, current_user: Account, app_model: App):
        args_model = CompletionMessagePayload.model_validate(console_ns.payload)
        args = args_model.model_dump(exclude_none=True, by_alias=True)

@@ -112,8 +114,6 @@ class CompletionMessageApi(Resource):
        args["auto_generate_name"] = False

        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account or EndUser instance")
            response = AppGenerateService.generate(
                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=streaming
            )
@@ -178,7 +178,8 @@ class ChatMessageApi(Resource):
    @account_initialization_required
    @get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.AGENT])
    @edit_permission_required
-    def post(self, app_model: App):
+    @with_current_user
+    def post(self, current_user: Account, app_model: App):
        raw_payload = console_ns.payload or {}
        args_model = ChatMessagePayload.model_validate(raw_payload)
        args = args_model.model_dump(exclude_none=True, by_alias=True)
@@ -197,8 +198,6 @@ class ChatMessageApi(Resource):
            args["external_trace_id"] = external_trace_id

        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account or EndUser instance")
            response = AppGenerateService.generate(
                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=streaming
            )
@@ -7,12 +7,18 @@ from pydantic import BaseModel, Field, TypeAdapter, computed_field, field_valida
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
-from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+    with_current_tenant_id,
+    with_current_user,
+)
 from fields.base import ResponseModel
 from fields.member_fields import AccountWithRole
 from libs.helper import build_avatar_url, dump_response, to_timestamp
-from libs.login import current_user, login_required
-from models import App
+from libs.login import login_required
+from models import Account, App
 from services.account_service import TenantService
 from services.workflow_comment_service import WorkflowCommentService

@@ -213,9 +219,10 @@ class WorkflowCommentListApi(Resource):
    @setup_required
    @account_initialization_required
    @get_app_model()
-    def get(self, app_model: App):
+    @with_current_tenant_id
+    def get(self, current_tenant_id: str, app_model: App):
        """Get all comments for a workflow."""
-        comments = WorkflowCommentService.get_comments(tenant_id=current_user.current_tenant_id, app_id=app_model.id)
+        comments = WorkflowCommentService.get_comments(tenant_id=current_tenant_id, app_id=app_model.id)

        return WorkflowCommentBasicList.model_validate({"data": comments}).model_dump(mode="json")

@@ -229,12 +236,14 @@ class WorkflowCommentListApi(Resource):
    @account_initialization_required
    @get_app_model()
    @edit_permission_required
-    def post(self, app_model: App):
+    @with_current_user
+    @with_current_tenant_id
+    def post(self, current_tenant_id: str, current_user: Account, app_model: App):
        """Create a new workflow comment."""
        payload = WorkflowCommentCreatePayload.model_validate(console_ns.payload or {})

        result = WorkflowCommentService.create_comment(
-            tenant_id=current_user.current_tenant_id,
+            tenant_id=current_tenant_id,
            app_id=app_model.id,
            created_by=current_user.id,
            content=payload.content,
@@ -258,10 +267,11 @@ class WorkflowCommentDetailApi(Resource):
    @setup_required
    @account_initialization_required
    @get_app_model()
-    def get(self, app_model: App, comment_id: str):
+    @with_current_tenant_id
+    def get(self, current_tenant_id: str, app_model: App, comment_id: str):
        """Get a specific workflow comment."""
        comment = WorkflowCommentService.get_comment(
-            tenant_id=current_user.current_tenant_id, app_id=app_model.id, comment_id=comment_id
+            tenant_id=current_tenant_id, app_id=app_model.id, comment_id=comment_id
        )

        return dump_response(WorkflowCommentDetail, comment)
@@ -276,12 +286,14 @@ class WorkflowCommentDetailApi(Resource):
    @account_initialization_required
    @get_app_model()
    @edit_permission_required
-    def put(self, app_model: App, comment_id: str):
+    @with_current_user
+    @with_current_tenant_id
+    def put(self, current_tenant_id: str, current_user: Account, app_model: App, comment_id: str):
        """Update a workflow comment."""
        payload = WorkflowCommentUpdatePayload.model_validate(console_ns.payload or {})

        result = WorkflowCommentService.update_comment(
-            tenant_id=current_user.current_tenant_id,
+            tenant_id=current_tenant_id,
            app_id=app_model.id,
            comment_id=comment_id,
            user_id=current_user.id,
@@ -302,10 +314,12 @@ class WorkflowCommentDetailApi(Resource):
    @account_initialization_required
    @get_app_model()
    @edit_permission_required
-    def delete(self, app_model: App, comment_id: str):
+    @with_current_user
+    @with_current_tenant_id
+    def delete(self, current_tenant_id: str, current_user: Account, app_model: App, comment_id: str):
        """Delete a workflow comment."""
        WorkflowCommentService.delete_comment(
-            tenant_id=current_user.current_tenant_id,
+            tenant_id=current_tenant_id,
            app_id=app_model.id,
            comment_id=comment_id,
            user_id=current_user.id,
@@ -327,10 +341,12 @@ class WorkflowCommentResolveApi(Resource):
    @account_initialization_required
    @get_app_model()
    @edit_permission_required
-    def post(self, app_model: App, comment_id: str):
+    @with_current_user
+    @with_current_tenant_id
+    def post(self, current_tenant_id: str, current_user: Account, app_model: App, comment_id: str):
        """Resolve a workflow comment."""
        comment = WorkflowCommentService.resolve_comment(
-            tenant_id=current_user.current_tenant_id,
+            tenant_id=current_tenant_id,
            app_id=app_model.id,
            comment_id=comment_id,
            user_id=current_user.id,
@@ -353,11 +369,13 @@ class WorkflowCommentReplyApi(Resource):
    @account_initialization_required
    @get_app_model()
    @edit_permission_required
-    def post(self, app_model: App, comment_id: str):
+    @with_current_user
+    @with_current_tenant_id
+    def post(self, current_tenant_id: str, current_user: Account, app_model: App, comment_id: str):
        """Add a reply to a workflow comment."""
        # Validate comment access first
        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
+            comment_id=comment_id, tenant_id=current_tenant_id, app_id=app_model.id
        )

        payload = WorkflowCommentReplyPayload.model_validate(console_ns.payload or {})
@@ -386,17 +404,19 @@ class WorkflowCommentReplyDetailApi(Resource):
    @account_initialization_required
    @get_app_model()
    @edit_permission_required
-    def put(self, app_model: App, comment_id: str, reply_id: str):
+    @with_current_user
+    @with_current_tenant_id
+    def put(self, current_tenant_id: str, current_user: Account, app_model: App, comment_id: str, reply_id: str):
        """Update a comment reply."""
        # Validate comment access first
        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
+            comment_id=comment_id, tenant_id=current_tenant_id, app_id=app_model.id
        )

        payload = WorkflowCommentReplyPayload.model_validate(console_ns.payload or {})

        reply = WorkflowCommentService.update_reply(
-            tenant_id=current_user.current_tenant_id,
+            tenant_id=current_tenant_id,
            app_id=app_model.id,
            comment_id=comment_id,
            reply_id=reply_id,
@@ -416,15 +436,17 @@ class WorkflowCommentReplyDetailApi(Resource):
    @account_initialization_required
    @get_app_model()
    @edit_permission_required
-    def delete(self, app_model: App, comment_id: str, reply_id: str):
+    @with_current_user
+    @with_current_tenant_id
+    def delete(self, current_tenant_id: str, current_user: Account, app_model: App, comment_id: str, reply_id: str):
        """Delete a comment reply."""
        # Validate comment access first
        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
+            comment_id=comment_id, tenant_id=current_tenant_id, app_id=app_model.id
        )

        WorkflowCommentService.delete_reply(
-            tenant_id=current_user.current_tenant_id,
+            tenant_id=current_tenant_id,
            app_id=app_model.id,
            comment_id=comment_id,
            reply_id=reply_id,
@@ -448,9 +470,13 @@ class WorkflowCommentMentionUsersApi(Resource):
    @setup_required
    @account_initialization_required
    @get_app_model()
-    def get(self, app_model: App):
+    @with_current_user
+    def get(self, current_user: Account, app_model: App):
        """Get all users in current tenant for mentions."""
-        members = TenantService.get_tenant_members(current_user.current_tenant)
+        current_tenant = current_user.current_tenant  # need the tenant object here
+        if current_tenant is None:
+            raise ValueError("current tenant is required")
+        members = TenantService.get_tenant_members(current_tenant)
        users = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
        response = WorkflowCommentMentionUsersPayload(users=users)
        return response.model_dump(mode="json"), 200
@@ -15,7 +15,12 @@ from controllers.console.app.error import (
    DraftWorkflowNotExist,
 )
 from controllers.console.app.wraps import get_app_model
-from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+    with_current_user,
+)
 from controllers.web.error import InvalidArgumentError, NotFoundError
 from core.app.file_access import DatabaseFileAccessController
 from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
@@ -27,8 +32,8 @@ from graphon.file import helpers as file_helpers
 from graphon.variables.segment_group import SegmentGroup
 from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment
 from graphon.variables.types import SegmentType
-from libs.login import current_user, login_required
-from models import App, AppMode
+from libs.login import login_required
+from models import Account, App, AppMode
 from models.workflow import WorkflowDraftVariable
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
 from services.workflow_service import WorkflowService
@@ -123,14 +128,15 @@ def _serialize_full_content(variable: WorkflowDraftVariable) -> FullContentDict
    return result


-def _ensure_variable_access(
+def ensure_variable_access(
    variable: WorkflowDraftVariable | None,
    app_id: str,
    variable_id: str,
+    current_user_id: str,
 ) -> WorkflowDraftVariable:
    if variable is None:
        raise NotFoundError(description=f"variable not found, id={variable_id}")
-    if variable.app_id != app_id or variable.user_id != current_user.id:
+    if variable.app_id != app_id or variable.user_id != current_user_id:
        raise NotFoundError(description=f"variable not found, id={variable_id}")
    return variable

@@ -215,7 +221,7 @@ workflow_draft_variable_list_model = console_ns.model(


 def _api_prerequisite[T, **P, R](
-    f: Callable[Concatenate[T, P], R],
+    f: Callable[Concatenate[T, Account, P], R],
 ) -> Callable[Concatenate[T, P], R | Response]:
    """Common prerequisites for all draft workflow variable APIs.

@@ -232,9 +238,10 @@ def _api_prerequisite[T, **P, R](
    @account_initialization_required
    @edit_permission_required
    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    @with_current_user
    @wraps(f)
-    def wrapper(self: T, *args: P.args, **kwargs: P.kwargs) -> R | Response:
-        return f(self, *args, **kwargs)
+    def wrapper(self: T, current_user: Account, *args: P.args, **kwargs: P.kwargs) -> R | Response:
+        return f(self, current_user, *args, **kwargs)

    return wrapper

@@ -251,7 +258,7 @@ class WorkflowVariableCollectionApi(Resource):
    )
    @_api_prerequisite
    @marshal_with(workflow_draft_variable_list_without_value_model)
-    def get(self, app_model: App):
+    def get(self, current_user: Account, app_model: App):
        """
        Get draft workflow
        """
@@ -281,7 +288,7 @@ class WorkflowVariableCollectionApi(Resource):
    @console_ns.doc(description="Delete all draft workflow variables")
    @console_ns.response(204, "Workflow variables deleted successfully")
    @_api_prerequisite
-    def delete(self, app_model: App):
+    def delete(self, current_user: Account, app_model: App):
        draft_var_srv = WorkflowDraftVariableService(
            session=db.session(),
        )
@@ -315,7 +322,7 @@ class NodeVariableCollectionApi(Resource):
    @console_ns.response(200, "Node variables retrieved successfully", workflow_draft_variable_list_model)
    @_api_prerequisite
    @marshal_with(workflow_draft_variable_list_model)
-    def get(self, app_model: App, node_id: str):
+    def get(self, current_user: Account, app_model: App, node_id: str):
        validate_node_id(node_id)
        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
            draft_var_srv = WorkflowDraftVariableService(
@@ -329,7 +336,7 @@ class NodeVariableCollectionApi(Resource):
    @console_ns.doc(description="Delete all variables for a specific node")
    @console_ns.response(204, "Node variables deleted successfully")
    @_api_prerequisite
-    def delete(self, app_model: App, node_id: str):
+    def delete(self, current_user: Account, app_model: App, node_id: str):
        validate_node_id(node_id)
        srv = WorkflowDraftVariableService(db.session())
        srv.delete_node_variables(app_model.id, node_id, user_id=current_user.id)
@@ -349,15 +356,16 @@ class VariableApi(Resource):
    @console_ns.response(404, "Variable not found")
    @_api_prerequisite
    @marshal_with(workflow_draft_variable_model)
-    def get(self, app_model: App, variable_id: UUID):
+    def get(self, current_user: Account, app_model: App, variable_id: UUID):
        draft_var_srv = WorkflowDraftVariableService(
            session=db.session(),
        )
        variable_id_str = str(variable_id)
-        variable = _ensure_variable_access(
+        variable = ensure_variable_access(
            variable=draft_var_srv.get_variable(variable_id=variable_id_str),
            app_id=app_model.id,
            variable_id=variable_id_str,
+            current_user_id=current_user.id,
        )
        return variable

@@ -368,7 +376,7 @@ class VariableApi(Resource):
    @console_ns.response(404, "Variable not found")
    @_api_prerequisite
    @marshal_with(workflow_draft_variable_model)
-    def patch(self, app_model: App, variable_id: UUID):
+    def patch(self, current_user: Account, app_model: App, variable_id: UUID):
        # Request payload for file types:
        #
        # Local File:
@@ -396,10 +404,11 @@ class VariableApi(Resource):
        args_model = WorkflowDraftVariableUpdatePayload.model_validate(console_ns.payload or {})

        variable_id_str = str(variable_id)
-        variable = _ensure_variable_access(
+        variable = ensure_variable_access(
            variable=draft_var_srv.get_variable(variable_id=variable_id_str),
            app_id=app_model.id,
            variable_id=variable_id_str,
+            current_user_id=current_user.id,
        )

        new_name = args_model.name
@@ -440,15 +449,16 @@ class VariableApi(Resource):
    @console_ns.response(204, "Variable deleted successfully")
    @console_ns.response(404, "Variable not found")
    @_api_prerequisite
-    def delete(self, app_model: App, variable_id: UUID):
+    def delete(self, current_user: Account, app_model: App, variable_id: UUID):
        draft_var_srv = WorkflowDraftVariableService(
            session=db.session(),
        )
        variable_id_str = str(variable_id)
-        variable = _ensure_variable_access(
+        variable = ensure_variable_access(
            variable=draft_var_srv.get_variable(variable_id=variable_id_str),
            app_id=app_model.id,
            variable_id=variable_id_str,
+            current_user_id=current_user.id,
        )
        draft_var_srv.delete_variable(variable)
        db.session.commit()
@@ -464,7 +474,7 @@ class VariableResetApi(Resource):
    @console_ns.response(204, "Variable reset (no content)")
    @console_ns.response(404, "Variable not found")
    @_api_prerequisite
-    def put(self, app_model: App, variable_id: UUID):
+    def put(self, current_user: Account, app_model: App, variable_id: UUID):
        draft_var_srv = WorkflowDraftVariableService(
            session=db.session(),
        )
@@ -476,10 +486,11 @@ class VariableResetApi(Resource):
                f"Draft workflow not found, app_id={app_model.id}",
            )
        variable_id_str = str(variable_id)
-        variable = _ensure_variable_access(
+        variable = ensure_variable_access(
            variable=draft_var_srv.get_variable(variable_id=variable_id_str),
            app_id=app_model.id,
            variable_id=variable_id_str,
+            current_user_id=current_user.id,
        )

        resetted = draft_var_srv.reset_variable(draft_workflow, variable)
@@ -490,20 +501,20 @@ class VariableResetApi(Resource):
            return marshal(resetted, workflow_draft_variable_model)


-def _get_variable_list(app_model: App, node_id) -> WorkflowDraftVariableList:
+def _get_variable_list(app_model: App, node_id: str, current_user_id: str) -> WorkflowDraftVariableList:
    with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
        draft_var_srv = WorkflowDraftVariableService(
            session=session,
        )
        if node_id == CONVERSATION_VARIABLE_NODE_ID:
-            draft_vars = draft_var_srv.list_conversation_variables(app_model.id, user_id=current_user.id)
+            draft_vars = draft_var_srv.list_conversation_variables(app_model.id, user_id=current_user_id)
        elif node_id == SYSTEM_VARIABLE_NODE_ID:
-            draft_vars = draft_var_srv.list_system_variables(app_model.id, user_id=current_user.id)
+            draft_vars = draft_var_srv.list_system_variables(app_model.id, user_id=current_user_id)
        else:
            draft_vars = draft_var_srv.list_node_variables(
                app_id=app_model.id,
                node_id=node_id,
-                user_id=current_user.id,
+                user_id=current_user_id,
            )
    return draft_vars

@@ -517,7 +528,7 @@ class ConversationVariableCollectionApi(Resource):
    @console_ns.response(404, "Draft workflow not found")
    @_api_prerequisite
    @marshal_with(workflow_draft_variable_list_model)
-    def get(self, app_model: App):
+    def get(self, current_user: Account, app_model: App):
        # NOTE(QuantumGhost): Prefill conversation variables into the draft variables table
        # so their IDs can be returned to the caller.
        workflow_srv = WorkflowService()
@@ -527,7 +538,7 @@ class ConversationVariableCollectionApi(Resource):
        draft_var_srv = WorkflowDraftVariableService(db.session())
        draft_var_srv.prefill_conversation_variable_default_values(draft_workflow, user_id=current_user.id)
        db.session.commit()
-        return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
+        return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID, current_user.id)

    @console_ns.expect(console_ns.models[ConversationVariableUpdatePayload.__name__])
    @console_ns.doc("update_conversation_variables")
@@ -539,7 +550,8 @@ class ConversationVariableCollectionApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @get_app_model(mode=AppMode.ADVANCED_CHAT)
-    def post(self, app_model: App):
+    @with_current_user
+    def post(self, current_user: Account, app_model: App):
        payload = ConversationVariableUpdatePayload.model_validate(console_ns.payload or {})

        workflow_service = WorkflowService()
@@ -566,8 +578,8 @@ class SystemVariableCollectionApi(Resource):
    @console_ns.response(200, "System variables retrieved successfully", workflow_draft_variable_list_model)
    @_api_prerequisite
    @marshal_with(workflow_draft_variable_list_model)
-    def get(self, app_model: App):
-        return _get_variable_list(app_model, SYSTEM_VARIABLE_NODE_ID)
+    def get(self, current_user: Account, app_model: App):
+        return _get_variable_list(app_model, SYSTEM_VARIABLE_NODE_ID, current_user.id)


@console_ns.route("/apps/<uuid:app_id>/workflows/draft/environment-variables")
@@ -578,7 +590,7 @@ class EnvironmentVariableCollectionApi(Resource):
    @console_ns.response(200, "Environment variables retrieved successfully")
    @console_ns.response(404, "Draft workflow not found")
    @_api_prerequisite
-    def get(self, app_model: App):
+    def get(self, _current_user: Account, app_model: App):
        """
        Get draft workflow
        """
@@ -619,7 +631,8 @@ class EnvironmentVariableCollectionApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
+    @with_current_user
+    def post(self, current_user: Account, app_model: App):
        payload = EnvironmentVariableUpdatePayload.model_validate(console_ns.payload or {})

        workflow_service = WorkflowService()
@@ -1,5 +1,5 @@
 from datetime import UTC, datetime, timedelta
-from typing import Literal, cast
+from typing import Literal
 from uuid import UUID

 from flask import request
@@ -12,7 +12,12 @@ from configs import dify_config
 from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
-from controllers.console.wraps import account_initialization_required, setup_required
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+    with_current_tenant_id,
+    with_current_user,
+)
 from controllers.web.error import NotFoundError
 from core.workflow.human_input_forms import load_form_tokens_by_form_id as _load_form_tokens_by_form_id
 from extensions.ext_database import db
@@ -30,8 +35,8 @@ from graphon.enums import WorkflowExecutionStatus
 from libs.archive_storage import ArchiveStorageNotConfiguredError, get_archive_storage
 from libs.custom_inputs import time_duration
 from libs.helper import uuid_value
-from libs.login import current_user, login_required
-from models import Account, App, AppMode, EndUser, WorkflowArchiveLog, WorkflowRunTriggeredFrom
+from libs.login import login_required
+from models import Account, App, AppMode, WorkflowArchiveLog, WorkflowRunTriggeredFrom
 from models.workflow import WorkflowRun
 from repositories.factory import DifyAPIRepositoryFactory
 from services.retention.workflow_run.constants import ARCHIVE_BUNDLE_NAME
@@ -190,8 +195,8 @@ class WorkflowRunExportApi(Resource):
    @account_initialization_required
    @get_app_model()
    def get(self, app_model: App, run_id: UUID):
-        tenant_id = str(app_model.tenant_id)
-        app_id = str(app_model.id)
+        tenant_id = app_model.tenant_id
+        app_id = app_model.id
        run_id_str = str(run_id)

        run_created_at = db.session.scalar(
@@ -397,18 +402,18 @@ class WorkflowRunNodeExecutionListApi(Resource):
    @login_required
    @account_initialization_required
    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App, run_id: UUID):
+    @with_current_user
+    def get(self, current_user: Account, app_model: App, run_id: UUID):
        """
        Get workflow run node execution list
        """
        run_id_str = str(run_id)

        workflow_run_service = WorkflowRunService()
-        user = cast("Account | EndUser", current_user)
        node_executions = workflow_run_service.get_workflow_run_node_executions(
            app_model=app_model,
            run_id=run_id_str,
-            user=user,
+            user=current_user,
        )

        return WorkflowRunNodeExecutionListResponse.model_validate(
@@ -432,7 +437,8 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    def get(self, workflow_run_id: str):
+    @with_current_tenant_id
+    def get(self, current_tenant_id: str, workflow_run_id: str):
        """
        Get workflow pause details.

@@ -449,7 +455,7 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
        if not workflow_run:
            raise NotFoundError("Workflow run not found")

-        if workflow_run.tenant_id != current_user.current_tenant_id:
+        if workflow_run.tenant_id != current_tenant_id:
            raise NotFoundError("Workflow run not found")

        # Check if workflow is suspended
@@ -12,14 +12,14 @@ from configs import dify_config
 from controllers.common.schema import register_schema_models
 from extensions.ext_database import db
 from fields.base import ResponseModel
-from libs.login import current_user, login_required
+from libs.login import login_required
 from models.enums import AppTriggerStatus
-from models.model import Account, App, AppMode
+from models.model import App, AppMode
 from models.trigger import AppTrigger, WorkflowWebhookTrigger

 from .. import console_ns
 from ..app.wraps import get_app_model
-from ..wraps import account_initialization_required, edit_permission_required, setup_required
+from ..wraps import account_initialization_required, edit_permission_required, setup_required, with_current_tenant_id

 logger = logging.getLogger(__name__)

@@ -124,18 +124,16 @@ class AppTriggersApi(Resource):
    @account_initialization_required
    @get_app_model(mode=AppMode.WORKFLOW)
    @console_ns.response(200, "Success", console_ns.models[WorkflowTriggerListResponse.__name__])
-    def get(self, app_model: App):
+    @with_current_tenant_id
+    def get(self, current_tenant_id: str, app_model: App):
        """Get app triggers list"""
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
-
        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            # Get all triggers for this app using select API
            triggers = (
                session.execute(
                    select(AppTrigger)
                    .where(
-                        AppTrigger.tenant_id == current_user.current_tenant_id,
+                        AppTrigger.tenant_id == current_tenant_id,
                        AppTrigger.app_id == app_model.id,
                    )
                    .order_by(AppTrigger.created_at.desc(), AppTrigger.id.desc())
@@ -166,19 +164,18 @@ class AppTriggerEnableApi(Resource):
    @edit_permission_required
    @get_app_model(mode=AppMode.WORKFLOW)
    @console_ns.response(200, "Success", console_ns.models[WorkflowTriggerResponse.__name__])
-    def post(self, app_model: App):
+    @with_current_tenant_id
+    def post(self, current_tenant_id: str, app_model: App):
        """Update app trigger (enable/disable)"""
        args = ParserEnable.model_validate(console_ns.payload)

-        assert current_user.current_tenant_id is not None
-
        trigger_id = args.trigger_id
        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            # Find the trigger using select
            trigger = session.execute(
                select(AppTrigger).where(
                    AppTrigger.id == trigger_id,
-                    AppTrigger.tenant_id == current_user.current_tenant_id,
+                    AppTrigger.tenant_id == current_tenant_id,
                    AppTrigger.app_id == app_model.id,
                )
            ).scalar_one_or_none()
@@ -76,7 +76,7 @@ class EmailRegisterSendEmailApi(Resource):
        if AccountService.is_email_send_ip_limit(ip_address):
            raise EmailSendIpLimitError()
        language = "en-US"
-        if args.language in languages:
+        if args.language is not None and args.language in languages:
            language = args.language

        if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(normalized_email):
@@ -2,13 +2,12 @@ from flask_restx import (  # type: ignore
    Resource,  # type: ignore
 )
 from pydantic import BaseModel
-from werkzeug.exceptions import Forbidden

 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.datasets.wraps import get_rag_pipeline
-from controllers.console.wraps import account_initialization_required, setup_required
-from libs.login import current_user, login_required
+from controllers.console.wraps import account_initialization_required, setup_required, with_current_user
+from libs.login import login_required
 from models import Account
 from models.dataset import Pipeline
 from services.rag_pipeline.rag_pipeline import RagPipelineService
@@ -30,13 +29,11 @@ class DataSourceContentPreviewApi(Resource):
    @login_required
    @account_initialization_required
    @get_rag_pipeline
-    def post(self, pipeline: Pipeline, node_id: str):
+    @with_current_user
+    def post(self, current_user: Account, pipeline: Pipeline, node_id: str):
        """
        Run datasource content preview
        """
-        if not isinstance(current_user, Account):
-            raise Forbidden()
-
        args = Parser.model_validate(console_ns.payload)

        inputs = args.inputs
@@ -1,5 +1,6 @@
 import logging
 from collections.abc import Callable
+from functools import wraps
 from typing import Any, Concatenate, NoReturn
 from uuid import UUID

@@ -21,7 +22,7 @@ from controllers.console.app.workflow_draft_variable import (
    workflow_draft_variable_model,
 )
 from controllers.console.datasets.wraps import get_rag_pipeline
-from controllers.console.wraps import account_initialization_required, setup_required
+from controllers.console.wraps import account_initialization_required, setup_required, with_current_user
 from controllers.web.error import InvalidArgumentError, NotFoundError
 from core.app.file_access import DatabaseFileAccessController
 from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
@@ -29,7 +30,7 @@ from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type
 from graphon.variables.types import SegmentType
-from libs.login import current_user, login_required
+from libs.login import login_required
 from models import Account
 from models.dataset import Pipeline
 from services.rag_pipeline.rag_pipeline import RagPipelineService
@@ -58,7 +59,7 @@ register_schema_models(console_ns, WorkflowDraftVariablePatchPayload)


 def _api_prerequisite[T, **P, R](
-    f: Callable[Concatenate[T, P], R],
+    f: Callable[Concatenate[T, Account, P], R],
 ) -> Callable[Concatenate[T, P], R | Response]:
    """Common prerequisites for all draft workflow variable APIs.

@@ -74,10 +75,12 @@ def _api_prerequisite[T, **P, R](
    @login_required
    @account_initialization_required
    @get_rag_pipeline
-    def wrapper(self: T, *args: P.args, **kwargs: P.kwargs) -> R | Response:
-        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+    @with_current_user
+    @wraps(f)
+    def wrapper(self: T, current_user: Account, *args: P.args, **kwargs: P.kwargs) -> R | Response:
+        if not current_user.has_edit_permission:
            raise Forbidden()
-        return f(self, *args, **kwargs)
+        return f(self, current_user, *args, **kwargs)

    return wrapper

@@ -86,7 +89,7 @@ def _api_prerequisite[T, **P, R](
 class RagPipelineVariableCollectionApi(Resource):
    @_api_prerequisite
    @marshal_with(workflow_draft_variable_list_without_value_model)
-    def get(self, pipeline: Pipeline):
+    def get(self, current_user: Account, pipeline: Pipeline):
        """
        Get draft workflow
        """
@@ -114,7 +117,7 @@ class RagPipelineVariableCollectionApi(Resource):
        return workflow_vars

    @_api_prerequisite
-    def delete(self, pipeline: Pipeline):
+    def delete(self, current_user: Account, pipeline: Pipeline):
        draft_var_srv = WorkflowDraftVariableService(
            session=db.session(),
        )
@@ -145,7 +148,7 @@ def validate_node_id(node_id: str) -> NoReturn | None:
 class RagPipelineNodeVariableCollectionApi(Resource):
    @_api_prerequisite
    @marshal_with(workflow_draft_variable_list_model)
-    def get(self, pipeline: Pipeline, node_id: str):
+    def get(self, current_user: Account, pipeline: Pipeline, node_id: str):
        validate_node_id(node_id)
        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
            draft_var_srv = WorkflowDraftVariableService(
@@ -156,7 +159,7 @@ class RagPipelineNodeVariableCollectionApi(Resource):
        return node_vars

    @_api_prerequisite
-    def delete(self, pipeline: Pipeline, node_id: str):
+    def delete(self, current_user: Account, pipeline: Pipeline, node_id: str):
        validate_node_id(node_id)
        srv = WorkflowDraftVariableService(db.session())
        srv.delete_node_variables(pipeline.id, node_id, user_id=current_user.id)
@@ -171,7 +174,7 @@ class RagPipelineVariableApi(Resource):

    @_api_prerequisite
    @marshal_with(workflow_draft_variable_model)
-    def get(self, pipeline: Pipeline, variable_id: UUID):
+    def get(self, _current_user: Account, pipeline: Pipeline, variable_id: UUID):
        draft_var_srv = WorkflowDraftVariableService(
            session=db.session(),
        )
@@ -186,7 +189,7 @@ class RagPipelineVariableApi(Resource):
    @_api_prerequisite
    @marshal_with(workflow_draft_variable_model)
    @console_ns.expect(console_ns.models[WorkflowDraftVariablePatchPayload.__name__])
-    def patch(self, pipeline: Pipeline, variable_id: UUID):
+    def patch(self, _current_user: Account, pipeline: Pipeline, variable_id: UUID):
        # Request payload for file types:
        #
        # Local File:
@@ -255,7 +258,7 @@ class RagPipelineVariableApi(Resource):
        return variable

    @_api_prerequisite
-    def delete(self, pipeline: Pipeline, variable_id: UUID):
+    def delete(self, _current_user: Account, pipeline: Pipeline, variable_id: UUID):
        draft_var_srv = WorkflowDraftVariableService(
            session=db.session(),
        )
@@ -273,7 +276,7 @@ class RagPipelineVariableApi(Resource):
@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/variables/<uuid:variable_id>/reset")
 class RagPipelineVariableResetApi(Resource):
    @_api_prerequisite
-    def put(self, pipeline: Pipeline, variable_id: UUID):
+    def put(self, _current_user: Account, pipeline: Pipeline, variable_id: UUID):
        draft_var_srv = WorkflowDraftVariableService(
            session=db.session(),
        )
@@ -299,17 +302,17 @@ class RagPipelineVariableResetApi(Resource):
            return marshal(resetted, _WORKFLOW_DRAFT_VARIABLE_FIELDS)


-def _get_variable_list(pipeline: Pipeline, node_id) -> WorkflowDraftVariableList:
+def _get_variable_list(pipeline: Pipeline, node_id: str, current_user_id: str) -> WorkflowDraftVariableList:
    with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
        draft_var_srv = WorkflowDraftVariableService(
            session=session,
        )
        if node_id == CONVERSATION_VARIABLE_NODE_ID:
-            draft_vars = draft_var_srv.list_conversation_variables(pipeline.id, user_id=current_user.id)
+            draft_vars = draft_var_srv.list_conversation_variables(pipeline.id, user_id=current_user_id)
        elif node_id == SYSTEM_VARIABLE_NODE_ID:
-            draft_vars = draft_var_srv.list_system_variables(pipeline.id, user_id=current_user.id)
+            draft_vars = draft_var_srv.list_system_variables(pipeline.id, user_id=current_user_id)
        else:
-            draft_vars = draft_var_srv.list_node_variables(app_id=pipeline.id, node_id=node_id, user_id=current_user.id)
+            draft_vars = draft_var_srv.list_node_variables(app_id=pipeline.id, node_id=node_id, user_id=current_user_id)
    return draft_vars


@@ -317,14 +320,14 @@ def _get_variable_list(pipeline: Pipeline, node_id) -> WorkflowDraftVariableList
 class RagPipelineSystemVariableCollectionApi(Resource):
    @_api_prerequisite
    @marshal_with(workflow_draft_variable_list_model)
-    def get(self, pipeline: Pipeline):
-        return _get_variable_list(pipeline, SYSTEM_VARIABLE_NODE_ID)
+    def get(self, current_user: Account, pipeline: Pipeline):
+        return _get_variable_list(pipeline, SYSTEM_VARIABLE_NODE_ID, current_user.id)


@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/environment-variables")
 class RagPipelineEnvironmentVariableCollectionApi(Resource):
    @_api_prerequisite
-    def get(self, pipeline: Pipeline):
+    def get(self, _current_user: Account, pipeline: Pipeline):
        """
        Get draft workflow
        """
@@ -155,19 +155,28 @@ class InstalledAppsListApi(Resource):
        if current_user.current_tenant is None:
            raise ValueError("current_user.current_tenant must not be None")
        current_user.role = TenantService.get_user_role(current_user, current_user.current_tenant)
-        installed_app_list: list[dict[str, Any]] = [
-            {
-                "id": installed_app.id,
-                "app": installed_app.app,
-                "app_owner_tenant_id": installed_app.app_owner_tenant_id,
-                "is_pinned": installed_app.is_pinned,
-                "last_used_at": installed_app.last_used_at,
-                "editable": current_user.role in {"owner", "admin"},
-                "uninstallable": current_tenant_id == installed_app.app_owner_tenant_id,
-            }
-            for installed_app in installed_apps
-            if installed_app.app is not None
-        ]
+
+        app_ids = [installed_app.app_id for installed_app in installed_apps]
+        apps = db.session.scalars(select(App).where(App.id.in_(app_ids))).all() if app_ids else []
+        apps_by_id = {app.id: app for app in apps}
+
+        installed_app_list: list[dict[str, Any]] = []
+        for installed_app in installed_apps:
+            app_model = apps_by_id.get(installed_app.app_id)
+            if app_model is None:
+                continue
+
+            installed_app_list.append(
+                {
+                    "id": installed_app.id,
+                    "app": app_model,
+                    "app_owner_tenant_id": installed_app.app_owner_tenant_id,
+                    "is_pinned": installed_app.is_pinned,
+                    "last_used_at": installed_app.last_used_at,
+                    "editable": current_user.role in {"owner", "admin"},
+                    "uninstallable": current_tenant_id == installed_app.app_owner_tenant_id,
+                }
+            )

        # filter out apps that user doesn't have access to
        if FeatureService.get_system_features().webapp_auth.enabled:
@@ -8,10 +8,11 @@ from pydantic import BaseModel, Field, computed_field, field_validator
 from constants.languages import languages
 from controllers.common.schema import query_params_from_model, register_schema_models
 from controllers.console import console_ns
-from controllers.console.wraps import account_initialization_required
+from controllers.console.wraps import account_initialization_required, with_current_user
 from fields.base import ResponseModel
 from libs.helper import build_icon_url
-from libs.login import current_user, login_required
+from libs.login import login_required
+from models import Account
 from services.recommended_app_service import RecommendedAppService


@@ -64,31 +65,39 @@ class RecommendedAppListResponse(ResponseModel):
    categories: list[str]


+class LearnDifyAppListResponse(ResponseModel):
+    recommended_apps: list[RecommendedAppResponse]
+
+
 register_schema_models(
    console_ns,
    RecommendedAppsQuery,
    RecommendedAppInfoResponse,
    RecommendedAppResponse,
    RecommendedAppListResponse,
+    LearnDifyAppListResponse,
 )


+def _resolve_language(language: str | None, user: Account) -> str:
+    if language and language in languages:
+        return language
+    if user.interface_language:
+        return user.interface_language
+    return languages[0]
+
+
@console_ns.route("/explore/apps")
 class RecommendedAppListApi(Resource):
    @console_ns.doc(params=query_params_from_model(RecommendedAppsQuery))
    @console_ns.response(200, "Success", console_ns.models[RecommendedAppListResponse.__name__])
    @login_required
    @account_initialization_required
-    def get(self):
+    @with_current_user
+    def get(self, current_user: Account):
        # language args
        args = RecommendedAppsQuery.model_validate(request.args.to_dict(flat=True))
-        language = args.language
-        if language and language in languages:
-            language_prefix = language
-        elif current_user and current_user.interface_language:
-            language_prefix = current_user.interface_language
-        else:
-            language_prefix = languages[0]
+        language_prefix = _resolve_language(args.language, current_user)

        return RecommendedAppListResponse.model_validate(
            RecommendedAppService.get_recommended_apps_and_categories(language_prefix),
@@ -96,6 +105,23 @@ class RecommendedAppListApi(Resource):
        ).model_dump(mode="json")


+@console_ns.route("/explore/apps/learn-dify")
+class LearnDifyAppListApi(Resource):
+    @console_ns.doc(params=query_params_from_model(RecommendedAppsQuery))
+    @console_ns.response(200, "Success", console_ns.models[LearnDifyAppListResponse.__name__])
+    @login_required
+    @account_initialization_required
+    @with_current_user
+    def get(self, current_user: Account):
+        args = RecommendedAppsQuery.model_validate(request.args.to_dict(flat=True))
+        language_prefix = _resolve_language(args.language, current_user)
+
+        return LearnDifyAppListResponse.model_validate(
+            RecommendedAppService.get_learn_dify_apps(language_prefix),
+            from_attributes=True,
+        ).model_dump(mode="json")
+
+
@console_ns.route("/explore/apps/<uuid:app_id>")
 class RecommendedAppApi(Resource):
    @login_required
@@ -0,0 +1,164 @@
+import uuid
+from typing import Any, Literal
+
+from pydantic import AliasChoices, BaseModel, Field, field_validator
+
+
+class SnippetListQuery(BaseModel):
+    """Query parameters for listing snippets."""
+
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=20, ge=1, le=100)
+    keyword: str | None = None
+    is_published: bool | None = Field(default=None, description="Filter by published status")
+    creators: list[str] | None = Field(
+        default=None,
+        description="Filter by creator account IDs",
+        validation_alias=AliasChoices("creators", "creator_id"),
+    )
+    tag_ids: list[str] | None = Field(default=None, description="Filter by tag IDs")
+
+    @field_validator("creators", mode="before")
+    @classmethod
+    def parse_creators(cls, value: object) -> list[str] | None:
+        """Normalize creators filter from query string or list input."""
+        return cls._normalize_string_list(value)
+
+    @field_validator("tag_ids", mode="before")
+    @classmethod
+    def parse_tag_ids(cls, value: object) -> list[str] | None:
+        """Normalize and validate tag IDs from query string or list input."""
+        items = cls._normalize_string_list(value)
+        if not items:
+            return None
+        try:
+            return [str(uuid.UUID(item)) for item in items]
+        except ValueError as exc:
+            raise ValueError("Invalid UUID format in tag_ids.") from exc
+
+    @staticmethod
+    def _normalize_string_list(value: object) -> list[str] | None:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return [item.strip() for item in value.split(",") if item.strip()] or None
+        if isinstance(value, list):
+            return [str(item).strip() for item in value if str(item).strip()] or None
+        return None
+
+
+class IconInfo(BaseModel):
+    """Icon information model."""
+
+    icon: str | None = None
+    icon_type: Literal["emoji", "image"] | None = None
+    icon_background: str | None = None
+    icon_url: str | None = None
+
+
+class InputFieldDefinition(BaseModel):
+    """Input field definition for snippet parameters."""
+
+    default: str | None = None
+    hint: bool | None = None
+    label: str | None = None
+    max_length: int | None = None
+    options: list[str] | None = None
+    placeholder: str | None = None
+    required: bool | None = None
+    type: str | None = None  # e.g., "text-input"
+
+
+class CreateSnippetPayload(BaseModel):
+    """Payload for creating a new snippet."""
+
+    name: str = Field(..., min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=2000)
+    type: Literal["node", "group"] = "node"
+    icon_info: IconInfo | None = None
+    graph: dict[str, Any] | None = None
+    input_fields: list[InputFieldDefinition] | None = Field(default_factory=list)
+
+
+class UpdateSnippetPayload(BaseModel):
+    """Payload for updating a snippet."""
+
+    name: str | None = Field(default=None, min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=2000)
+    icon_info: IconInfo | None = None
+
+
+class SnippetDraftSyncPayload(BaseModel):
+    """Payload for syncing snippet draft workflow."""
+
+    graph: dict[str, Any]
+    hash: str | None = None
+    conversation_variables: list[dict[str, Any]] | None = Field(
+        default=None,
+        description="Ignored. Snippet workflows do not persist conversation variables.",
+    )
+    input_fields: list[dict[str, Any]] | None = None
+
+
+class SnippetWorkflowListQuery(BaseModel):
+    """Query parameters for listing snippet published workflows."""
+
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=10, ge=1, le=100)
+
+
+class WorkflowRunQuery(BaseModel):
+    """Query parameters for workflow runs."""
+
+    last_id: str | None = None
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+class SnippetDraftRunPayload(BaseModel):
+    """Payload for running snippet draft workflow."""
+
+    inputs: dict[str, Any]
+    files: list[dict[str, Any]] | None = None
+
+
+class SnippetDraftNodeRunPayload(BaseModel):
+    """Payload for running a single node in snippet draft workflow."""
+
+    inputs: dict[str, Any]
+    query: str = ""
+    files: list[dict[str, Any]] | None = None
+
+
+class SnippetIterationNodeRunPayload(BaseModel):
+    """Payload for running an iteration node in snippet draft workflow."""
+
+    inputs: dict[str, Any] | None = None
+
+
+class SnippetLoopNodeRunPayload(BaseModel):
+    """Payload for running a loop node in snippet draft workflow."""
+
+    inputs: dict[str, Any] | None = None
+
+
+class PublishWorkflowPayload(BaseModel):
+    """Payload for publishing snippet workflow."""
+
+    knowledge_base_setting: dict[str, Any] | None = None
+
+
+class SnippetImportPayload(BaseModel):
+    """Payload for importing snippet from DSL."""
+
+    mode: str = Field(..., description="Import mode: yaml-content or yaml-url")
+    yaml_content: str | None = Field(default=None, description="YAML content (required for yaml-content mode)")
+    yaml_url: str | None = Field(default=None, description="YAML URL (required for yaml-url mode)")
+    name: str | None = Field(default=None, description="Override snippet name")
+    description: str | None = Field(default=None, description="Override snippet description")
+    snippet_id: str | None = Field(default=None, description="Snippet ID to update (optional)")
+
+
+class IncludeSecretQuery(BaseModel):
+    """Query parameter for including secret variables in export."""
+
+    include_secret: str = Field(default="false", description="Whether to include secret variables")
@@ -0,0 +1,678 @@
+import logging
+from collections.abc import Callable
+from functools import wraps
+
+from flask import request
+from flask_restx import Resource
+from pydantic import Field
+from sqlalchemy.orm import Session, sessionmaker
+from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
+
+from controllers.common.schema import register_response_schema_models, register_schema_models
+from controllers.console import console_ns
+from controllers.console.app.error import DraftWorkflowNotExist, DraftWorkflowNotSync
+from controllers.console.app.workflow import (
+    RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE,
+    WorkflowPaginationResponse,
+    WorkflowResponse,
+)
+from controllers.console.snippets.payloads import (
+    PublishWorkflowPayload,
+    SnippetDraftNodeRunPayload,
+    SnippetDraftRunPayload,
+    SnippetDraftSyncPayload,
+    SnippetIterationNodeRunPayload,
+    SnippetLoopNodeRunPayload,
+    SnippetWorkflowListQuery,
+    WorkflowRunQuery,
+)
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.entities.app_invoke_entities import InvokeFrom
+from extensions.ext_database import db
+from extensions.ext_redis import redis_client
+from fields.workflow_run_fields import (
+    WorkflowRunDetailResponse,
+    WorkflowRunNodeExecutionListResponse,
+    WorkflowRunNodeExecutionResponse,
+    WorkflowRunPaginationResponse,
+)
+from graphon.graph_engine.manager import GraphEngineManager
+from libs import helper
+from libs.helper import TimestampField
+from libs.login import current_account_with_tenant, login_required
+from models.snippet import CustomizedSnippet
+from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
+from services.snippet_generate_service import SnippetGenerateService
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+
+# Register Pydantic models with Swagger
+
+
+def _snippet_session_maker() -> sessionmaker[Session]:
+    return sessionmaker(bind=db.engine, expire_on_commit=False)
+
+
+def _snippet_service() -> SnippetService:
+    return SnippetService(_snippet_session_maker())
+
+
+class SnippetWorkflowResponse(WorkflowResponse):
+    input_fields: list[dict] = Field(default_factory=list)
+
+
+register_schema_models(
+    console_ns,
+    SnippetDraftSyncPayload,
+    SnippetDraftNodeRunPayload,
+    SnippetDraftRunPayload,
+    SnippetIterationNodeRunPayload,
+    SnippetLoopNodeRunPayload,
+    SnippetWorkflowListQuery,
+    WorkflowRunQuery,
+    PublishWorkflowPayload,
+)
+register_response_schema_models(
+    console_ns,
+    SnippetWorkflowResponse,
+    WorkflowPaginationResponse,
+    WorkflowRunPaginationResponse,
+    WorkflowRunDetailResponse,
+    WorkflowRunNodeExecutionListResponse,
+    WorkflowRunNodeExecutionResponse,
+)
+
+
+class SnippetNotFoundError(Exception):
+    """Snippet not found error."""
+
+    pass
+
+
+def get_snippet[**P, R](view_func: Callable[P, R]) -> Callable[P, R]:
+    """Decorator to fetch and validate snippet access."""
+
+    @wraps(view_func)
+    def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
+        if not kwargs.get("snippet_id"):
+            raise ValueError("missing snippet_id in path parameters")
+
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet_id = str(kwargs.get("snippet_id"))
+        del kwargs["snippet_id"]
+
+        snippet_service = _snippet_service()
+        snippet = snippet_service.get_snippet_by_id(
+            snippet_id=snippet_id,
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        kwargs["snippet"] = snippet
+
+        return view_func(*args, **kwargs)
+
+    return decorated_view
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft")
+class SnippetDraftWorkflowApi(Resource):
+    @console_ns.doc("get_snippet_draft_workflow")
+    @console_ns.response(
+        200,
+        "Draft workflow retrieved successfully",
+        console_ns.models[SnippetWorkflowResponse.__name__],
+    )
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get draft workflow for snippet."""
+        snippet_service = _snippet_service()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+
+        if not workflow:
+            raise DraftWorkflowNotExist()
+
+        workflow.conversation_variables = []
+        response = SnippetWorkflowResponse.model_validate(workflow, from_attributes=True).model_dump(mode="json")
+        response["input_fields"] = snippet.input_fields_list
+        return response
+
+    @console_ns.doc("sync_snippet_draft_workflow")
+    @console_ns.expect(console_ns.models.get(SnippetDraftSyncPayload.__name__))
+    @console_ns.response(200, "Draft workflow synced successfully")
+    @console_ns.response(400, "Hash mismatch")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """Sync draft workflow for snippet."""
+        current_user, _ = current_account_with_tenant()
+
+        payload = SnippetDraftSyncPayload.model_validate(console_ns.payload or {})
+
+        try:
+            snippet_service = _snippet_service()
+            workflow = snippet_service.sync_draft_workflow(
+                snippet=snippet,
+                graph=payload.graph,
+                unique_hash=payload.hash,
+                account=current_user,
+                input_fields=payload.input_fields,
+            )
+        except WorkflowHashNotEqualError:
+            raise DraftWorkflowNotSync()
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return {
+            "result": "success",
+            "hash": workflow.unique_hash,
+            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/config")
+class SnippetDraftConfigApi(Resource):
+    @console_ns.doc("get_snippet_draft_config")
+    @console_ns.response(200, "Draft config retrieved successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get snippet draft workflow configuration limits."""
+        return {
+            "parallel_depth_limit": 3,
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/publish")
+class SnippetPublishedWorkflowApi(Resource):
+    @console_ns.doc("get_snippet_published_workflow")
+    @console_ns.response(
+        200,
+        "Published workflow retrieved successfully",
+        console_ns.models[SnippetWorkflowResponse.__name__],
+    )
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get published workflow for snippet."""
+        if not snippet.is_published:
+            return None
+
+        snippet_service = _snippet_service()
+        workflow = snippet_service.get_published_workflow(snippet=snippet)
+
+        if not workflow:
+            return None
+
+        response = SnippetWorkflowResponse.model_validate(workflow, from_attributes=True).model_dump(mode="json")
+        response["input_fields"] = snippet.input_fields_list
+        return response
+
+    @console_ns.doc("publish_snippet_workflow")
+    @console_ns.expect(console_ns.models.get(PublishWorkflowPayload.__name__))
+    @console_ns.response(200, "Workflow published successfully")
+    @console_ns.response(400, "No draft workflow found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """Publish snippet workflow."""
+        current_user, _ = current_account_with_tenant()
+        snippet_service = _snippet_service()
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            try:
+                workflow = snippet_service.publish_workflow(
+                    session=session,
+                    snippet=snippet,
+                    account=current_user,
+                )
+                workflow_created_at = TimestampField().format(workflow.created_at)
+                session.commit()
+            except ValueError as e:
+                return {"message": str(e)}, 400
+
+        return {
+            "result": "success",
+            "created_at": workflow_created_at,
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/default-workflow-block-configs")
+class SnippetDefaultBlockConfigsApi(Resource):
+    @console_ns.doc("get_snippet_default_block_configs")
+    @console_ns.response(200, "Default block configs retrieved successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get default block configurations for snippet workflow."""
+        snippet_service = _snippet_service()
+        return snippet_service.get_default_block_configs()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows")
+class SnippetPublishedAllWorkflowApi(Resource):
+    @console_ns.expect(console_ns.models[SnippetWorkflowListQuery.__name__])
+    @console_ns.doc("get_all_snippet_published_workflows")
+    @console_ns.doc(description="Get all published workflows for a snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID"})
+    @console_ns.response(
+        200,
+        "Published workflows retrieved successfully",
+        console_ns.models[WorkflowPaginationResponse.__name__],
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get all published workflow versions for snippet."""
+        args = SnippetWorkflowListQuery.model_validate(request.args.to_dict(flat=True))
+
+        snippet_service = _snippet_service()
+        with Session(db.engine) as session:
+            workflows, has_more = snippet_service.get_all_published_workflows(
+                session=session,
+                snippet=snippet,
+                page=args.page,
+                limit=args.limit,
+            )
+
+        return WorkflowPaginationResponse.model_validate(
+            {
+                "items": workflows,
+                "page": args.page,
+                "limit": args.limit,
+                "has_more": has_more,
+            },
+            from_attributes=True,
+        ).model_dump(mode="json")
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/<string:workflow_id>/restore")
+class SnippetDraftWorkflowRestoreApi(Resource):
+    @console_ns.doc("restore_snippet_workflow_to_draft")
+    @console_ns.doc(description="Restore a published snippet workflow version into the draft workflow")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "workflow_id": "Published workflow ID"})
+    @console_ns.response(200, "Workflow restored successfully")
+    @console_ns.response(400, "Source workflow must be published")
+    @console_ns.response(404, "Workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, workflow_id: str):
+        """Restore a published snippet workflow version into the draft workflow."""
+        current_user, _ = current_account_with_tenant()
+        snippet_service = _snippet_service()
+
+        try:
+            workflow = snippet_service.restore_published_workflow_to_draft(
+                snippet=snippet,
+                workflow_id=workflow_id,
+                account=current_user,
+            )
+        except IsDraftWorkflowError as exc:
+            raise BadRequest(RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE) from exc
+        except WorkflowNotFoundError as exc:
+            raise NotFound(str(exc)) from exc
+        except ValueError as exc:
+            raise BadRequest(str(exc)) from exc
+
+        return {
+            "result": "success",
+            "hash": workflow.unique_hash,
+            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs")
+class SnippetWorkflowRunsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_runs")
+    @console_ns.response(
+        200,
+        "Workflow runs retrieved successfully",
+        console_ns.models[WorkflowRunPaginationResponse.__name__],
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    def get(self, snippet: CustomizedSnippet):
+        """List workflow runs for snippet."""
+        query = WorkflowRunQuery.model_validate(
+            {
+                "last_id": request.args.get("last_id"),
+                "limit": request.args.get("limit", type=int, default=20),
+            }
+        )
+        args = {
+            "last_id": query.last_id,
+            "limit": query.limit,
+        }
+
+        snippet_service = _snippet_service()
+        result = snippet_service.get_snippet_workflow_runs(snippet=snippet, args=args)
+
+        return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>")
+class SnippetWorkflowRunDetailApi(Resource):
+    @console_ns.doc("get_snippet_workflow_run_detail")
+    @console_ns.response(
+        200,
+        "Workflow run detail retrieved successfully",
+        console_ns.models[WorkflowRunDetailResponse.__name__],
+    )
+    @console_ns.response(404, "Workflow run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    def get(self, snippet: CustomizedSnippet, run_id):
+        """Get workflow run detail for snippet."""
+        run_id = str(run_id)
+
+        snippet_service = _snippet_service()
+        workflow_run = snippet_service.get_snippet_workflow_run(snippet=snippet, run_id=run_id)
+
+        if not workflow_run:
+            raise NotFound("Workflow run not found")
+
+        return WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json")
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>/node-executions")
+class SnippetWorkflowRunNodeExecutionsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_run_node_executions")
+    @console_ns.response(
+        200,
+        "Node executions retrieved successfully",
+        console_ns.models[WorkflowRunNodeExecutionListResponse.__name__],
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    def get(self, snippet: CustomizedSnippet, run_id):
+        """List node executions for a workflow run."""
+        run_id = str(run_id)
+
+        snippet_service = _snippet_service()
+        node_executions = snippet_service.get_snippet_workflow_run_node_executions(
+            snippet=snippet,
+            run_id=run_id,
+        )
+
+        return WorkflowRunNodeExecutionListResponse.model_validate(
+            {"data": node_executions}, from_attributes=True
+        ).model_dump(mode="json")
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/run")
+class SnippetDraftNodeRunApi(Resource):
+    @console_ns.doc("run_snippet_draft_node")
+    @console_ns.doc(description="Run a single node in snippet draft workflow (single-step debugging)")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetDraftNodeRunPayload.__name__))
+    @console_ns.response(
+        200, "Node run completed successfully", console_ns.models[WorkflowRunNodeExecutionResponse.__name__]
+    )
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a single node in snippet draft workflow.
+
+        Executes a specific node with provided inputs for single-step debugging.
+        Returns the node execution result including status, outputs, and timing.
+        """
+        current_user, _ = current_account_with_tenant()
+        payload = SnippetDraftNodeRunPayload.model_validate(console_ns.payload or {})
+
+        user_inputs = payload.inputs
+
+        # Get draft workflow for file parsing
+        snippet_service = _snippet_service()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not draft_workflow:
+            raise NotFound("Draft workflow not found")
+
+        files = SnippetGenerateService.parse_files(draft_workflow, payload.files)
+
+        workflow_node_execution = SnippetGenerateService.run_draft_node(
+            snippet=snippet,
+            node_id=node_id,
+            user_inputs=user_inputs,
+            account=current_user,
+            query=payload.query,
+            files=files,
+            session_maker=_snippet_session_maker(),
+        )
+
+        return WorkflowRunNodeExecutionResponse.model_validate(
+            workflow_node_execution, from_attributes=True
+        ).model_dump(mode="json")
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/last-run")
+class SnippetDraftNodeLastRunApi(Resource):
+    @console_ns.doc("get_snippet_draft_node_last_run")
+    @console_ns.doc(description="Get last run result for a node in snippet draft workflow")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.response(
+        200, "Node last run retrieved successfully", console_ns.models[WorkflowRunNodeExecutionResponse.__name__]
+    )
+    @console_ns.response(404, "Snippet, draft workflow, or node last run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    def get(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Get the last run result for a specific node in snippet draft workflow.
+
+        Returns the most recent execution record for the given node,
+        including status, inputs, outputs, and timing information.
+        """
+        snippet_service = _snippet_service()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not draft_workflow:
+            raise NotFound("Draft workflow not found")
+
+        node_exec = snippet_service.get_snippet_node_last_run(
+            snippet=snippet,
+            workflow=draft_workflow,
+            node_id=node_id,
+        )
+        if node_exec is None:
+            raise NotFound("Node last run not found")
+
+        return WorkflowRunNodeExecutionResponse.model_validate(node_exec, from_attributes=True).model_dump(mode="json")
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/iteration/nodes/<string:node_id>/run")
+class SnippetDraftRunIterationNodeApi(Resource):
+    @console_ns.doc("run_snippet_draft_iteration_node")
+    @console_ns.doc(description="Run draft workflow iteration node for snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetIterationNodeRunPayload.__name__))
+    @console_ns.response(200, "Iteration node run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a draft workflow iteration node for snippet.
+
+        Iteration nodes execute their internal sub-graph multiple times over an input list.
+        Returns an SSE event stream with iteration progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+        args = SnippetIterationNodeRunPayload.model_validate(console_ns.payload or {}).model_dump(exclude_none=True)
+
+        try:
+            response = SnippetGenerateService.generate_single_iteration(
+                snippet=snippet,
+                user=current_user,
+                node_id=node_id,
+                args=args,
+                streaming=True,
+                session_maker=_snippet_session_maker(),
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/loop/nodes/<string:node_id>/run")
+class SnippetDraftRunLoopNodeApi(Resource):
+    @console_ns.doc("run_snippet_draft_loop_node")
+    @console_ns.doc(description="Run draft workflow loop node for snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetLoopNodeRunPayload.__name__))
+    @console_ns.response(200, "Loop node run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a draft workflow loop node for snippet.
+
+        Loop nodes execute their internal sub-graph repeatedly until a condition is met.
+        Returns an SSE event stream with loop progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+        args = SnippetLoopNodeRunPayload.model_validate(console_ns.payload or {})
+
+        try:
+            response = SnippetGenerateService.generate_single_loop(
+                snippet=snippet,
+                user=current_user,
+                node_id=node_id,
+                args=args,
+                streaming=True,
+                session_maker=_snippet_session_maker(),
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/run")
+class SnippetDraftWorkflowRunApi(Resource):
+    @console_ns.doc("run_snippet_draft_workflow")
+    @console_ns.expect(console_ns.models.get(SnippetDraftRunPayload.__name__))
+    @console_ns.response(200, "Draft workflow run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """
+        Run draft workflow for snippet.
+
+        Executes the snippet's draft workflow with the provided inputs
+        and returns an SSE event stream with execution progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+
+        payload = SnippetDraftRunPayload.model_validate(console_ns.payload or {})
+        args = payload.model_dump(exclude_none=True)
+
+        try:
+            response = SnippetGenerateService.generate(
+                snippet=snippet,
+                user=current_user,
+                args=args,
+                invoke_from=InvokeFrom.DEBUGGER,
+                streaming=True,
+                session_maker=_snippet_session_maker(),
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/tasks/<string:task_id>/stop")
+class SnippetWorkflowTaskStopApi(Resource):
+    @console_ns.doc("stop_snippet_workflow_task")
+    @console_ns.response(200, "Task stopped successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, task_id: str):
+        """
+        Stop a running snippet workflow task.
+
+        Uses both the legacy stop flag mechanism and the graph engine
+        command channel for backward compatibility.
+        """
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager(redis_client).send_stop_command(task_id)
+
+        return {"result": "success"}
@@ -0,0 +1,334 @@
+"""
+Snippet draft workflow variable APIs.
+
+Mirrors console app routes under /apps/.../workflows/draft/variables for snippet scope,
+using CustomizedSnippet.id as WorkflowDraftVariable.app_id (same invariant as snippet execution).
+
+Snippet workflows do not expose system variables (`node_id == sys`) or conversation variables
+(`node_id == conversation`): paginated list queries exclude those rows; single-variable GET/PATCH/DELETE/reset
+reject them; `GET .../system-variables` and `GET .../conversation-variables` return empty lists for API parity.
+Other routes mirror `workflow_draft_variable` app APIs under `/snippets/...`.
+"""
+
+from collections.abc import Callable
+from functools import wraps
+from typing import Any, Concatenate
+
+from flask import Response, request
+from flask_restx import Resource, marshal, marshal_with
+from sqlalchemy.orm import Session, sessionmaker
+
+from controllers.console import console_ns
+from controllers.console.app.error import DraftWorkflowNotExist
+from controllers.console.app.workflow_draft_variable import (
+    WorkflowDraftVariableListQuery,
+    WorkflowDraftVariableUpdatePayload,
+    ensure_variable_access,
+    validate_node_id,
+    workflow_draft_variable_list_model,
+    workflow_draft_variable_list_without_value_model,
+    workflow_draft_variable_model,
+)
+from controllers.console.snippets.snippet_workflow import get_snippet
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+    with_current_user,
+)
+from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.app.file_access import DatabaseFileAccessController
+from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from extensions.ext_database import db
+from factories.file_factory import build_from_mapping, build_from_mappings
+from factories.variable_factory import build_segment_with_type
+from graphon.variables.types import SegmentType
+from libs.login import login_required
+from models import Account
+from models.snippet import CustomizedSnippet
+from models.workflow import WorkflowDraftVariable
+from services.snippet_service import SnippetService
+from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
+
+_SNIPPET_EXCLUDED_DRAFT_VARIABLE_NODE_IDS: frozenset[str] = frozenset(
+    {SYSTEM_VARIABLE_NODE_ID, CONVERSATION_VARIABLE_NODE_ID}
+)
+_file_access_controller = DatabaseFileAccessController()
+
+
+def _snippet_service() -> SnippetService:
+    return SnippetService(sessionmaker(bind=db.engine, expire_on_commit=False))
+
+
+def _ensure_snippet_draft_variable_row_allowed(
+    *,
+    variable: WorkflowDraftVariable,
+    variable_id: str,
+) -> None:
+    """Snippet scope only supports canvas-node draft variables; treat sys/conversation rows as not found."""
+    if variable.node_id in _SNIPPET_EXCLUDED_DRAFT_VARIABLE_NODE_IDS:
+        raise NotFoundError(description=f"variable not found, id={variable_id}")
+
+
+def _snippet_draft_var_prerequisite[T, **P, R](
+    f: Callable[Concatenate[T, Account, P], R],
+) -> Callable[Concatenate[T, P], R | Response]:
+    """Setup, auth, snippet resolution, and tenant edit permission (same stack as snippet workflow APIs)."""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @with_current_user
+    @wraps(f)
+    def wrapper(self: T, current_user: Account, *args: P.args, **kwargs: P.kwargs) -> R | Response:
+        return f(self, current_user, *args, **kwargs)
+
+    return wrapper
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/variables")
+class SnippetWorkflowVariableCollectionApi(Resource):
+    @console_ns.expect(console_ns.models[WorkflowDraftVariableListQuery.__name__])
+    @console_ns.doc("get_snippet_workflow_variables")
+    @console_ns.doc(description="List draft workflow variables without values (paginated, snippet scope)")
+    @console_ns.response(
+        200,
+        "Workflow variables retrieved successfully",
+        workflow_draft_variable_list_without_value_model,
+    )
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_without_value_model)
+    def get(self, current_user: Account, snippet: CustomizedSnippet) -> WorkflowDraftVariableList:
+        args = WorkflowDraftVariableListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+
+        snippet_service = _snippet_service()
+        if snippet_service.get_draft_workflow(snippet=snippet) is None:
+            raise DraftWorkflowNotExist()
+
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(session=session)
+            workflow_vars = draft_var_srv.list_variables_without_values(
+                app_id=snippet.id,
+                page=args.page,
+                limit=args.limit,
+                user_id=current_user.id,
+                exclude_node_ids=_SNIPPET_EXCLUDED_DRAFT_VARIABLE_NODE_IDS,
+            )
+
+        return workflow_vars
+
+    @console_ns.doc("delete_snippet_workflow_variables")
+    @console_ns.doc(description="Delete all draft workflow variables for the current user (snippet scope)")
+    @console_ns.response(204, "Workflow variables deleted successfully")
+    @_snippet_draft_var_prerequisite
+    def delete(self, current_user: Account, snippet: CustomizedSnippet) -> Response:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        draft_var_srv.delete_user_workflow_variables(snippet.id, user_id=current_user.id)
+        db.session.commit()
+        return Response("", 204)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/variables")
+class SnippetNodeVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_node_variables")
+    @console_ns.doc(description="Get variables for a specific node (snippet draft workflow)")
+    @console_ns.response(200, "Node variables retrieved successfully", workflow_draft_variable_list_model)
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_model)
+    def get(self, current_user: Account, snippet: CustomizedSnippet, node_id: str) -> WorkflowDraftVariableList:
+        validate_node_id(node_id)
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(session=session)
+            node_vars = draft_var_srv.list_node_variables(snippet.id, node_id, user_id=current_user.id)
+
+        return node_vars
+
+    @console_ns.doc("delete_snippet_node_variables")
+    @console_ns.doc(description="Delete all variables for a specific node (snippet draft workflow)")
+    @console_ns.response(204, "Node variables deleted successfully")
+    @_snippet_draft_var_prerequisite
+    def delete(self, current_user: Account, snippet: CustomizedSnippet, node_id: str) -> Response:
+        validate_node_id(node_id)
+        srv = WorkflowDraftVariableService(db.session())
+        srv.delete_node_variables(snippet.id, node_id, user_id=current_user.id)
+        db.session.commit()
+        return Response("", 204)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/variables/<uuid:variable_id>")
+class SnippetVariableApi(Resource):
+    @console_ns.doc("get_snippet_workflow_variable")
+    @console_ns.doc(description="Get a specific draft workflow variable (snippet scope)")
+    @console_ns.response(200, "Variable retrieved successfully", workflow_draft_variable_model)
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_model)
+    def get(self, current_user: Account, snippet: CustomizedSnippet, variable_id: str) -> WorkflowDraftVariable:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        variable = ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+            current_user_id=current_user.id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+        return variable
+
+    @console_ns.doc("update_snippet_workflow_variable")
+    @console_ns.doc(description="Update a draft workflow variable (snippet scope)")
+    @console_ns.expect(console_ns.models[WorkflowDraftVariableUpdatePayload.__name__])
+    @console_ns.response(200, "Variable updated successfully", workflow_draft_variable_model)
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_model)
+    def patch(self, current_user: Account, snippet: CustomizedSnippet, variable_id: str) -> WorkflowDraftVariable:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        args_model = WorkflowDraftVariableUpdatePayload.model_validate(console_ns.payload or {})
+
+        variable = ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+            current_user_id=current_user.id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+
+        new_name = args_model.name
+        raw_value = args_model.value
+        if new_name is None and raw_value is None:
+            return variable
+
+        new_value = None
+        if raw_value is not None:
+            if variable.value_type == SegmentType.FILE:
+                if not isinstance(raw_value, dict):
+                    raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
+                    raw_value = build_from_mapping(
+                        mapping=raw_value,
+                        tenant_id=snippet.tenant_id,
+                        access_controller=_file_access_controller,
+                    )
+            elif variable.value_type == SegmentType.ARRAY_FILE:
+                if not isinstance(raw_value, list):
+                    raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
+                if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
+                    raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
+                    raw_value = build_from_mappings(
+                        mappings=raw_value,
+                        tenant_id=snippet.tenant_id,
+                        access_controller=_file_access_controller,
+                    )
+            new_value = build_segment_with_type(variable.value_type, raw_value)
+        draft_var_srv.update_variable(variable, name=new_name, value=new_value)
+        db.session.commit()
+        return variable
+
+    @console_ns.doc("delete_snippet_workflow_variable")
+    @console_ns.doc(description="Delete a draft workflow variable (snippet scope)")
+    @console_ns.response(204, "Variable deleted successfully")
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    def delete(self, current_user: Account, snippet: CustomizedSnippet, variable_id: str) -> Response:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        variable = ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+            current_user_id=current_user.id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+        draft_var_srv.delete_variable(variable)
+        db.session.commit()
+        return Response("", 204)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/variables/<uuid:variable_id>/reset")
+class SnippetVariableResetApi(Resource):
+    @console_ns.doc("reset_snippet_workflow_variable")
+    @console_ns.doc(description="Reset a draft workflow variable to its default value (snippet scope)")
+    @console_ns.response(200, "Variable reset successfully", workflow_draft_variable_model)
+    @console_ns.response(204, "Variable reset (no content)")
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    def put(self, current_user: Account, snippet: CustomizedSnippet, variable_id: str) -> Response | Any:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        snippet_service = _snippet_service()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if draft_workflow is None:
+            raise NotFoundError(
+                f"Draft workflow not found, snippet_id={snippet.id}",
+            )
+        variable = ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+            current_user_id=current_user.id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+
+        resetted = draft_var_srv.reset_variable(draft_workflow, variable)
+        db.session.commit()
+        if resetted is None:
+            return Response("", 204)
+        return marshal(resetted, workflow_draft_variable_model)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/conversation-variables")
+class SnippetConversationVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_conversation_variables")
+    @console_ns.doc(
+        description="Conversation variables are not used in snippet workflows; returns an empty list for API parity"
+    )
+    @console_ns.response(200, "Conversation variables retrieved successfully", workflow_draft_variable_list_model)
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_model)
+    def get(self, _current_user: Account, snippet: CustomizedSnippet) -> WorkflowDraftVariableList:
+        return WorkflowDraftVariableList(variables=[])
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/system-variables")
+class SnippetSystemVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_system_variables")
+    @console_ns.doc(
+        description="System variables are not used in snippet workflows; returns an empty list for API parity"
+    )
+    @console_ns.response(200, "System variables retrieved successfully", workflow_draft_variable_list_model)
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_model)
+    def get(self, _current_user: Account, snippet: CustomizedSnippet) -> WorkflowDraftVariableList:
+        return WorkflowDraftVariableList(variables=[])
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/environment-variables")
+class SnippetEnvironmentVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_environment_variables")
+    @console_ns.doc(description="Get environment variables from snippet draft workflow graph")
+    @console_ns.response(200, "Environment variables retrieved successfully")
+    @console_ns.response(404, "Draft workflow not found")
+    @_snippet_draft_var_prerequisite
+    def get(self, _current_user: Account, snippet: CustomizedSnippet) -> dict[str, list[dict[str, Any]]]:
+        snippet_service = _snippet_service()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if workflow is None:
+            raise DraftWorkflowNotExist()
+
+        env_vars_list: list[dict[str, Any]] = []
+        for v in workflow.environment_variables:
+            env_vars_list.append(
+                {
+                    "id": v.id,
+                    "type": "env",
+                    "name": v.name,
+                    "description": v.description,
+                    "selector": v.selector,
+                    "value_type": v.value_type.exposed_type().value,
+                    "value": v.value,
+                    "edited": False,
+                    "visible": True,
+                    "editable": True,
+                }
+            )
+
+        return {"items": env_vars_list}
@@ -51,7 +51,7 @@ class TagBindingRemovePayload(BaseModel):


 class TagListQueryParam(BaseModel):
-    type: Literal["knowledge", "app", ""] = Field("", description="Tag type filter")
+    type: Literal["knowledge", "app", "snippet", ""] = Field("", description="Tag type filter")
    keyword: str | None = Field(None, description="Search keyword")


@@ -96,7 +96,10 @@ class TagListApi(Resource):
    @login_required
    @account_initialization_required
    @console_ns.doc(
-        params={"type": 'Tag type filter. Can be "knowledge" or "app".', "keyword": "Search keyword for tag name."}
+        params={
+            "type": 'Tag type filter. Can be "knowledge", "app", or "snippet".',
+            "keyword": "Search keyword for tag name.",
+        }
    )
    @console_ns.doc(responses={200: ("Success", [console_ns.models[TagResponse.__name__]])})
    @with_current_tenant_id
@@ -1,28 +1,51 @@
 import io
 from collections.abc import Mapping
-from typing import Any, Literal
+from datetime import datetime
+from typing import Any, Literal, TypedDict

 from flask import request, send_file
 from flask_restx import Resource
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 from werkzeug.datastructures import FileStorage
 from werkzeug.exceptions import Forbidden

 from configs import dify_config
 from controllers.common.fields import SuccessResponse
-from controllers.common.schema import register_enum_models, register_response_schema_models, register_schema_models
+from controllers.common.schema import (
+    query_params_from_model,
+    register_enum_models,
+    register_response_schema_models,
+    register_schema_models,
+)
 from controllers.console import console_ns
 from controllers.console.workspace import plugin_permission_required
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
+from core.helper.position_helper import is_filtered
+from core.plugin.entities.plugin import PluginCategory, PluginDeclaration, PluginInstallationSource
 from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.plugin.plugin_service import PluginService
+from core.tools.builtin_tool.providers._positions import BuiltinToolProviderSort
+from core.tools.entities.common_entities import I18nObject
+from core.tools.entities.tool_entities import ToolProviderType
+from core.tools.tool_manager import ToolManager
 from fields.base import ResponseModel
 from graphon.model_runtime.utils.encoders import jsonable_encoder
+from libs.helper import dump_response
 from libs.login import current_account_with_tenant, login_required
 from models.account import TenantPluginAutoUpgradeStrategy, TenantPluginPermission
+from models.provider_ids import ToolProviderID
 from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService
 from services.plugin.plugin_parameter_service import PluginParameterService
 from services.plugin.plugin_permission_service import PluginPermissionService
+from services.tools.tools_transform_service import ToolTransformService
+
+
+class AutoUpgradeSettingsResponse(TypedDict):
+    strategy_setting: TenantPluginAutoUpgradeStrategy.StrategySetting
+    upgrade_time_of_day: int
+    upgrade_mode: TenantPluginAutoUpgradeStrategy.UpgradeMode
+    exclude_plugins: list[str]
+    include_plugins: list[str]


 class ParserList(BaseModel):
@@ -30,6 +53,11 @@ class ParserList(BaseModel):
    page_size: int = Field(default=256, ge=1, le=256, description="Page size (1-256)")


+class PluginCategoryListQuery(BaseModel):
+    page: int = Field(default=1, ge=1, description="Page number")
+    page_size: int = Field(default=256, ge=1, le=256, description="Page size (1-256)")
+
+
 class ParserLatest(BaseModel):
    plugin_ids: list[str]

@@ -88,8 +116,8 @@ class ParserUninstall(BaseModel):


 class ParserPermissionChange(BaseModel):
-    install_permission: TenantPluginPermission.InstallPermission
-    debug_permission: TenantPluginPermission.DebugPermission
+    install_permission: TenantPluginPermission.InstallPermission = TenantPluginPermission.InstallPermission.EVERYONE
+    debug_permission: TenantPluginPermission.DebugPermission = TenantPluginPermission.DebugPermission.EVERYONE


 class ParserDynamicOptions(BaseModel):
@@ -125,13 +153,40 @@ class PluginAutoUpgradeSettingsPayload(BaseModel):
    include_plugins: list[str] = Field(default_factory=list)


-class ParserPreferencesChange(BaseModel):
-    permission: PluginPermissionSettingsPayload
+class PluginAutoUpgradeChangeResponse(ResponseModel):
+    success: bool
+    message: str | None = None
+
+
+class PluginAutoUpgradeSettingsResponseModel(ResponseModel):
+    strategy_setting: TenantPluginAutoUpgradeStrategy.StrategySetting
+    upgrade_time_of_day: int
+    upgrade_mode: TenantPluginAutoUpgradeStrategy.UpgradeMode
+    exclude_plugins: list[str]
+    include_plugins: list[str]
+
+
+class PluginAutoUpgradeFetchResponse(ResponseModel):
+    category: TenantPluginAutoUpgradeStrategy.PluginCategory
+    auto_upgrade: PluginAutoUpgradeSettingsResponseModel
+
+
+class ParserAutoUpgradeChange(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    category: TenantPluginAutoUpgradeStrategy.PluginCategory
    auto_upgrade: PluginAutoUpgradeSettingsPayload


+class ParserAutoUpgradeFetch(BaseModel):
+    category: TenantPluginAutoUpgradeStrategy.PluginCategory
+
+
 class ParserExcludePlugin(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
    plugin_id: str
+    category: TenantPluginAutoUpgradeStrategy.PluginCategory


 class ParserReadme(BaseModel):
@@ -145,9 +200,67 @@ class PluginDebuggingKeyResponse(ResponseModel):
    port: int


+class PluginCategoryInstalledPluginResponse(ResponseModel):
+    id: str
+    name: str
+    tenant_id: str
+    plugin_id: str
+    plugin_unique_identifier: str
+    endpoints_active: int
+    endpoints_setups: int
+    installation_id: str
+    declaration: PluginDeclaration
+    runtime_type: str
+    version: str
+    created_at: datetime
+    updated_at: datetime
+    source: PluginInstallationSource
+    checksum: str
+    meta: Mapping[str, Any]
+
+
+class PluginCategoryBuiltinToolResponse(ResponseModel):
+    model_config = ConfigDict(extra="allow")
+
+    author: str
+    name: str
+    label: I18nObject
+    description: I18nObject
+    parameters: list[Mapping[str, Any]] | None = None
+    labels: list[str]
+    output_schema: Mapping[str, object]
+
+
+class PluginCategoryBuiltinToolProviderResponse(ResponseModel):
+    model_config = ConfigDict(extra="allow")
+
+    id: str
+    author: str
+    name: str
+    plugin_id: str | None
+    plugin_unique_identifier: str | None
+    description: I18nObject
+    icon: str | Mapping[str, str]
+    icon_dark: str | Mapping[str, str] | None
+    label: I18nObject
+    type: ToolProviderType
+    team_credentials: Mapping[str, object]
+    is_team_authorization: bool
+    allow_delete: bool
+    tools: list[PluginCategoryBuiltinToolResponse]
+    labels: list[str]
+
+
+class PluginCategoryListResponse(ResponseModel):
+    plugins: list[PluginCategoryInstalledPluginResponse]
+    builtin_tools: list[PluginCategoryBuiltinToolProviderResponse]
+    has_more: bool
+
+
 register_schema_models(
    console_ns,
    ParserList,
+    PluginCategoryListQuery,
    PluginAutoUpgradeSettingsPayload,
    PluginPermissionSettingsPayload,
    ParserLatest,
@@ -164,21 +277,57 @@ register_schema_models(
    ParserPermissionChange,
    ParserDynamicOptions,
    ParserDynamicOptionsWithCredentials,
-    ParserPreferencesChange,
+    ParserAutoUpgradeChange,
+    ParserAutoUpgradeFetch,
    ParserExcludePlugin,
    ParserReadme,
 )
-register_response_schema_models(console_ns, PluginDebuggingKeyResponse, SuccessResponse)
+register_response_schema_models(
+    console_ns,
+    PluginAutoUpgradeChangeResponse,
+    PluginAutoUpgradeFetchResponse,
+    PluginAutoUpgradeSettingsResponseModel,
+    PluginDebuggingKeyResponse,
+    PluginCategoryInstalledPluginResponse,
+    PluginCategoryBuiltinToolResponse,
+    PluginCategoryBuiltinToolProviderResponse,
+    PluginCategoryListResponse,
+    SuccessResponse,
+)

 register_enum_models(
    console_ns,
    TenantPluginPermission.DebugPermission,
+    TenantPluginAutoUpgradeStrategy.PluginCategory,
    TenantPluginAutoUpgradeStrategy.UpgradeMode,
    TenantPluginAutoUpgradeStrategy.StrategySetting,
    TenantPluginPermission.InstallPermission,
 )


+def _default_auto_upgrade_settings(
+    tenant_id: str,
+    category: TenantPluginAutoUpgradeStrategy.PluginCategory,
+) -> AutoUpgradeSettingsResponse:
+    return {
+        "strategy_setting": PluginAutoUpgradeService.default_strategy_setting_for_category(category),
+        "upgrade_time_of_day": PluginAutoUpgradeService.default_upgrade_time_of_day(tenant_id),
+        "upgrade_mode": TenantPluginAutoUpgradeStrategy.UpgradeMode.EXCLUDE,
+        "exclude_plugins": [],
+        "include_plugins": [],
+    }
+
+
+def _auto_upgrade_settings_to_dict(strategy: TenantPluginAutoUpgradeStrategy) -> AutoUpgradeSettingsResponse:
+    return {
+        "strategy_setting": strategy.strategy_setting,
+        "upgrade_time_of_day": strategy.upgrade_time_of_day,
+        "upgrade_mode": strategy.upgrade_mode,
+        "exclude_plugins": strategy.exclude_plugins,
+        "include_plugins": strategy.include_plugins,
+    }
+
+
 def _read_upload_content(file: FileStorage, max_size: int) -> bytes:
    """
    Read the uploaded file and validate its actual size before delegating to the plugin service.
@@ -193,6 +342,33 @@ def _read_upload_content(file: FileStorage, max_size: int) -> bytes:
    return content


+def _list_hardcoded_builtin_tool_providers(tenant_id: str) -> list[dict[str, Any]]:
+    db_builtin_providers = {
+        str(ToolProviderID(provider.provider)): provider
+        for provider in ToolManager.list_default_builtin_providers(tenant_id)
+    }
+    builtin_providers = []
+
+    for provider in ToolManager.list_hardcoded_providers():
+        if is_filtered(
+            include_set=dify_config.POSITION_TOOL_INCLUDES_SET,
+            exclude_set=dify_config.POSITION_TOOL_EXCLUDES_SET,
+            data=provider,
+            name_func=lambda provider_controller: provider_controller.entity.identity.name,
+        ):
+            continue
+
+        user_provider = ToolTransformService.builtin_provider_to_user_provider(
+            provider_controller=provider,
+            db_provider=db_builtin_providers.get(provider.entity.identity.name),
+            decrypt_credentials=False,
+        )
+        ToolTransformService.repack_provider(tenant_id=tenant_id, provider=user_provider)
+        builtin_providers.append(user_provider)
+
+    return [provider.to_dict() for provider in BuiltinToolProviderSort.sort(builtin_providers)]
+
+
@console_ns.route("/workspaces/current/plugin/debugging-key")
 class PluginDebuggingKeyApi(Resource):
    @console_ns.response(200, "Success", console_ns.models[PluginDebuggingKeyResponse.__name__])
@@ -230,6 +406,41 @@ class PluginListApi(Resource):
        return jsonable_encoder({"plugins": plugins_with_total.list, "total": plugins_with_total.total})


+@console_ns.route("/workspaces/current/plugin/<string:category>/list")
+class PluginCategoryListApi(Resource):
+    @console_ns.doc(params=query_params_from_model(PluginCategoryListQuery))
+    @console_ns.response(200, "Success", console_ns.models[PluginCategoryListResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, category: str):
+        _, tenant_id = current_account_with_tenant()
+        args = PluginCategoryListQuery.model_validate(request.args.to_dict(flat=True))
+
+        try:
+            plugin_category = PluginCategory(category)
+        except ValueError:
+            return {"code": "invalid_param", "message": "invalid plugin category"}, 400
+
+        try:
+            plugins = PluginService.list_by_category(tenant_id, plugin_category, args.page, args.page_size)
+        except PluginDaemonClientSideError as e:
+            return {"code": "plugin_error", "message": e.description}, 400
+
+        builtin_tools = []
+        if plugin_category == PluginCategory.Tool:
+            builtin_tools = _list_hardcoded_builtin_tool_providers(tenant_id)
+
+        return dump_response(
+            PluginCategoryListResponse,
+            {
+                "plugins": jsonable_encoder(plugins.list),
+                "builtin_tools": builtin_tools,
+                "has_more": plugins.has_more,
+            },
+        )
+
+
@console_ns.route("/workspaces/current/plugin/list/latest-versions")
 class PluginListLatestVersionsApi(Resource):
    @console_ns.expect(console_ns.models[ParserLatest.__name__])
@@ -632,11 +843,13 @@ class PluginChangePermissionApi(Resource):

        tenant_id = current_tenant_id

-        return {
-            "success": PluginPermissionService.change_permission(
-                tenant_id, args.install_permission, args.debug_permission
-            )
-        }
+        set_permission_result = PluginPermissionService.change_permission(
+            tenant_id, args.install_permission, args.debug_permission
+        )
+        if not set_permission_result:
+            return jsonable_encoder({"success": False, "message": "Failed to set permission"})
+
+        return jsonable_encoder({"success": True})


@console_ns.route("/workspaces/current/plugin/permission/fetch")
@@ -725,9 +938,10 @@ class PluginFetchDynamicSelectOptionsWithCredentialsApi(Resource):
        return jsonable_encoder({"options": options})


-@console_ns.route("/workspaces/current/plugin/preferences/change")
-class PluginChangePreferencesApi(Resource):
-    @console_ns.expect(console_ns.models[ParserPreferencesChange.__name__])
+@console_ns.route("/workspaces/current/plugin/auto-upgrade/change")
+class PluginChangeAutoUpgradeApi(Resource):
+    @console_ns.expect(console_ns.models[ParserAutoUpgradeChange.__name__])
+    @console_ns.response(200, "Success", console_ns.models[PluginAutoUpgradeChangeResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
@@ -736,38 +950,17 @@ class PluginChangePreferencesApi(Resource):
        if not user.is_admin_or_owner:
            raise Forbidden()

-        args = ParserPreferencesChange.model_validate(console_ns.payload)
-
-        permission = args.permission
-
-        install_permission = permission.install_permission
-        debug_permission = permission.debug_permission
+        args = ParserAutoUpgradeChange.model_validate(console_ns.payload)

        auto_upgrade = args.auto_upgrade
-
-        strategy_setting = auto_upgrade.strategy_setting
-        upgrade_time_of_day = auto_upgrade.upgrade_time_of_day
-        upgrade_mode = auto_upgrade.upgrade_mode
-        exclude_plugins = auto_upgrade.exclude_plugins
-        include_plugins = auto_upgrade.include_plugins
-
-        # set permission
-        set_permission_result = PluginPermissionService.change_permission(
-            tenant_id,
-            install_permission,
-            debug_permission,
-        )
-        if not set_permission_result:
-            return jsonable_encoder({"success": False, "message": "Failed to set permission"})
-
-        # set auto upgrade strategy
        set_auto_upgrade_strategy_result = PluginAutoUpgradeService.change_strategy(
            tenant_id,
-            strategy_setting,
-            upgrade_time_of_day,
-            upgrade_mode,
-            exclude_plugins,
-            include_plugins,
+            auto_upgrade.strategy_setting,
+            auto_upgrade.upgrade_time_of_day,
+            auto_upgrade.upgrade_mode,
+            auto_upgrade.exclude_plugins,
+            auto_upgrade.include_plugins,
+            category=args.category,
        )
        if not set_auto_upgrade_strategy_result:
            return jsonable_encoder({"success": False, "message": "Failed to set auto upgrade strategy"})
@@ -775,48 +968,36 @@ class PluginChangePreferencesApi(Resource):
        return jsonable_encoder({"success": True})


-@console_ns.route("/workspaces/current/plugin/preferences/fetch")
-class PluginFetchPreferencesApi(Resource):
+@console_ns.route("/workspaces/current/plugin/auto-upgrade/fetch")
+class PluginFetchAutoUpgradeApi(Resource):
+    @console_ns.doc(params=query_params_from_model(ParserAutoUpgradeFetch))
+    @console_ns.response(200, "Success", console_ns.models[PluginAutoUpgradeFetchResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
    def get(self):
        _, tenant_id = current_account_with_tenant()

-        permission = PluginPermissionService.get_permission(tenant_id)
-        permission_dict = {
-            "install_permission": TenantPluginPermission.InstallPermission.EVERYONE,
-            "debug_permission": TenantPluginPermission.DebugPermission.EVERYONE,
-        }
+        args = ParserAutoUpgradeFetch.model_validate(request.args.to_dict(flat=True))
+        auto_upgrade = PluginAutoUpgradeService.get_strategy(tenant_id, args.category)
+        auto_upgrade_dict = (
+            _auto_upgrade_settings_to_dict(auto_upgrade)
+            if auto_upgrade
+            else _default_auto_upgrade_settings(tenant_id, args.category)
+        )

-        if permission:
-            permission_dict["install_permission"] = permission.install_permission
-            permission_dict["debug_permission"] = permission.debug_permission
-
-        auto_upgrade = PluginAutoUpgradeService.get_strategy(tenant_id)
-        auto_upgrade_dict = {
-            "strategy_setting": TenantPluginAutoUpgradeStrategy.StrategySetting.DISABLED,
-            "upgrade_time_of_day": 0,
-            "upgrade_mode": TenantPluginAutoUpgradeStrategy.UpgradeMode.EXCLUDE,
-            "exclude_plugins": [],
-            "include_plugins": [],
-        }
-
-        if auto_upgrade:
-            auto_upgrade_dict = {
-                "strategy_setting": auto_upgrade.strategy_setting,
-                "upgrade_time_of_day": auto_upgrade.upgrade_time_of_day,
-                "upgrade_mode": auto_upgrade.upgrade_mode,
-                "exclude_plugins": auto_upgrade.exclude_plugins,
-                "include_plugins": auto_upgrade.include_plugins,
+        return jsonable_encoder(
+            {
+                "category": args.category,
+                "auto_upgrade": auto_upgrade_dict,
            }
-
-        return jsonable_encoder({"permission": permission_dict, "auto_upgrade": auto_upgrade_dict})
+        )


-@console_ns.route("/workspaces/current/plugin/preferences/autoupgrade/exclude")
+@console_ns.route("/workspaces/current/plugin/auto-upgrade/exclude")
 class PluginAutoUpgradeExcludePluginApi(Resource):
    @console_ns.expect(console_ns.models[ParserExcludePlugin.__name__])
+    @console_ns.response(200, "Success", console_ns.models[SuccessResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
@@ -826,7 +1007,9 @@ class PluginAutoUpgradeExcludePluginApi(Resource):

        args = ParserExcludePlugin.model_validate(console_ns.payload)

-        return jsonable_encoder({"success": PluginAutoUpgradeService.exclude_plugin(tenant_id, args.plugin_id)})
+        return jsonable_encoder(
+            {"success": PluginAutoUpgradeService.exclude_plugin(tenant_id, args.plugin_id, args.category)}
+        )


@console_ns.route("/workspaces/current/plugin/readme")
@@ -0,0 +1,428 @@
+import logging
+import re
+from urllib.parse import quote
+
+from flask import Response, request
+from flask_restx import Resource, marshal
+from sqlalchemy.orm import Session, sessionmaker
+from werkzeug.datastructures import MultiDict
+from werkzeug.exceptions import NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.snippets.payloads import (
+    CreateSnippetPayload,
+    IncludeSecretQuery,
+    SnippetImportPayload,
+    SnippetListQuery,
+    UpdateSnippetPayload,
+)
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from extensions.ext_database import db
+from fields.snippet_fields import snippet_fields, snippet_list_fields, snippet_pagination_fields
+from libs.login import current_account_with_tenant, login_required
+from models.snippet import SnippetType
+from services.app_dsl_service import ImportStatus
+from services.snippet_dsl_service import SnippetDslService
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+_TAG_IDS_BRACKET_PATTERN = re.compile(r"^tag_ids\[(\d+)\]$")
+_CREATOR_IDS_BRACKET_PATTERN = re.compile(r"^creator_ids\[(\d+)\]$")
+
+
+def _snippet_service() -> SnippetService:
+    return SnippetService(sessionmaker(bind=db.engine, expire_on_commit=False))
+
+
+def _normalize_snippet_list_query_args(query_args: MultiDict[str, str]) -> dict[str, str | list[str]]:
+    normalized: dict[str, str | list[str]] = {}
+    indexed_tag_ids: list[tuple[int, str]] = []
+    indexed_creator_ids: list[tuple[int, str]] = []
+
+    for key in query_args:
+        match = _TAG_IDS_BRACKET_PATTERN.fullmatch(key)
+        if match:
+            indexed_tag_ids.extend((int(match.group(1)), value) for value in query_args.getlist(key))
+            continue
+
+        match = _CREATOR_IDS_BRACKET_PATTERN.fullmatch(key)
+        if match:
+            indexed_creator_ids.extend((int(match.group(1)), value) for value in query_args.getlist(key))
+            continue
+
+        value = query_args.get(key)
+        if value is not None:
+            normalized[key] = value
+
+    if indexed_tag_ids:
+        normalized["tag_ids"] = [value for _, value in sorted(indexed_tag_ids)]
+    if indexed_creator_ids:
+        normalized["creators"] = [value for _, value in sorted(indexed_creator_ids)]
+
+    return normalized
+
+
+# Register Pydantic models with Swagger
+register_schema_models(
+    console_ns,
+    SnippetListQuery,
+    CreateSnippetPayload,
+    UpdateSnippetPayload,
+    SnippetImportPayload,
+    IncludeSecretQuery,
+)
+
+# Create namespace models for marshaling
+snippet_model = console_ns.model("Snippet", snippet_fields)
+snippet_list_model = console_ns.model("SnippetList", snippet_list_fields)
+snippet_pagination_model = console_ns.model("SnippetPagination", snippet_pagination_fields)
+
+
+@console_ns.route("/workspaces/current/customized-snippets")
+class CustomizedSnippetsApi(Resource):
+    @console_ns.doc("list_customized_snippets")
+    @console_ns.expect(console_ns.models.get(SnippetListQuery.__name__))
+    @console_ns.response(200, "Snippets retrieved successfully", snippet_pagination_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """List customized snippets with pagination and search."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        query = SnippetListQuery.model_validate(_normalize_snippet_list_query_args(request.args))
+
+        snippet_service = _snippet_service()
+        snippets, total, has_more = snippet_service.get_snippets(
+            tenant_id=current_tenant_id,
+            page=query.page,
+            limit=query.limit,
+            keyword=query.keyword,
+            is_published=query.is_published,
+            creators=query.creators,
+            tag_ids=query.tag_ids,
+        )
+
+        return {
+            "data": marshal(snippets, snippet_list_fields),
+            "page": query.page,
+            "limit": query.limit,
+            "total": total,
+            "has_more": has_more,
+        }, 200
+
+    @console_ns.doc("create_customized_snippet")
+    @console_ns.expect(console_ns.models.get(CreateSnippetPayload.__name__))
+    @console_ns.response(201, "Snippet created successfully", snippet_model)
+    @console_ns.response(400, "Invalid request")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        """Create a new customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        payload = CreateSnippetPayload.model_validate(console_ns.payload or {})
+
+        try:
+            snippet_type = SnippetType(payload.type)
+        except ValueError:
+            snippet_type = SnippetType.NODE
+
+        try:
+            if payload.graph is not None:
+                SnippetService.validate_snippet_graph_forbidden_nodes(payload.graph)
+
+            snippet_service = _snippet_service()
+            snippet = snippet_service.create_snippet(
+                tenant_id=current_tenant_id,
+                name=payload.name,
+                description=payload.description,
+                snippet_type=snippet_type,
+                icon_info=payload.icon_info.model_dump() if payload.icon_info else None,
+                input_fields=[f.model_dump() for f in payload.input_fields] if payload.input_fields else None,
+                account=current_user,
+            )
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 201
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>")
+class CustomizedSnippetDetailApi(Resource):
+    @console_ns.doc("get_customized_snippet")
+    @console_ns.response(200, "Snippet retrieved successfully", snippet_model)
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, snippet_id: str):
+        """Get customized snippet details."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet_service = _snippet_service()
+        snippet = snippet_service.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("update_customized_snippet")
+    @console_ns.expect(console_ns.models.get(UpdateSnippetPayload.__name__))
+    @console_ns.response(200, "Snippet updated successfully", snippet_model)
+    @console_ns.response(400, "Invalid request")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def patch(self, snippet_id: str):
+        """Update customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        snippet_service = _snippet_service()
+        snippet = snippet_service.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        payload = UpdateSnippetPayload.model_validate(console_ns.payload or {})
+        update_data = payload.model_dump(exclude_unset=True)
+
+        if "icon_info" in update_data and update_data["icon_info"] is not None:
+            update_data["icon_info"] = payload.icon_info.model_dump() if payload.icon_info else None
+
+        if not update_data:
+            return {"message": "No valid fields to update"}, 400
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                snippet = session.merge(snippet)
+                snippet = SnippetService.update_snippet(
+                    session=session,
+                    snippet=snippet,
+                    account_id=current_user.id,
+                    data=update_data,
+                )
+                session.commit()
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("delete_customized_snippet")
+    @console_ns.response(204, "Snippet deleted successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def delete(self, snippet_id: str):
+        """Delete customized snippet."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet_service = _snippet_service()
+        snippet = snippet_service.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            SnippetService.delete_snippet(
+                session=session,
+                snippet=snippet,
+            )
+            session.commit()
+
+        return "", 204
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/export")
+class CustomizedSnippetExportApi(Resource):
+    @console_ns.doc("export_customized_snippet")
+    @console_ns.doc(description="Export snippet configuration as DSL")
+    @console_ns.doc(params={"snippet_id": "Snippet ID to export"})
+    @console_ns.response(200, "Snippet exported successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, snippet_id: str):
+        """Export snippet as DSL."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet_service = _snippet_service()
+        snippet = snippet_service.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        # Get include_secret parameter
+        query = IncludeSecretQuery.model_validate(request.args.to_dict())
+
+        with Session(db.engine) as session:
+            export_service = SnippetDslService(session)
+            result = export_service.export_snippet_dsl(snippet=snippet, include_secret=query.include_secret == "true")
+
+        # Set filename with .snippet extension
+        filename = f"{snippet.name}.snippet"
+        encoded_filename = quote(filename)
+
+        response = Response(
+            result,
+            mimetype="application/x-yaml",
+        )
+        response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+        response.headers["Content-Type"] = "application/x-yaml"
+
+        return response
+
+
+@console_ns.route("/workspaces/current/customized-snippets/imports")
+class CustomizedSnippetImportApi(Resource):
+    @console_ns.doc("import_customized_snippet")
+    @console_ns.doc(description="Import snippet from DSL")
+    @console_ns.expect(console_ns.models.get(SnippetImportPayload.__name__))
+    @console_ns.response(200, "Snippet imported successfully")
+    @console_ns.response(202, "Import pending confirmation")
+    @console_ns.response(400, "Import failed")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        """Import snippet from DSL."""
+        current_user, _ = current_account_with_tenant()
+        payload = SnippetImportPayload.model_validate(console_ns.payload or {})
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.import_snippet(
+                account=current_user,
+                import_mode=payload.mode,
+                yaml_content=payload.yaml_content,
+                yaml_url=payload.yaml_url,
+                snippet_id=payload.snippet_id,
+                name=payload.name,
+                description=payload.description,
+            )
+            session.commit()
+
+        # Return appropriate status code based on result
+        status = result.status
+        if status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        elif status == ImportStatus.PENDING:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/imports/<string:import_id>/confirm")
+class CustomizedSnippetImportConfirmApi(Resource):
+    @console_ns.doc("confirm_snippet_import")
+    @console_ns.doc(description="Confirm a pending snippet import")
+    @console_ns.doc(params={"import_id": "Import ID to confirm"})
+    @console_ns.response(200, "Import confirmed successfully")
+    @console_ns.response(400, "Import failed")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self, import_id: str):
+        """Confirm a pending snippet import."""
+        current_user, _ = current_account_with_tenant()
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.confirm_import(import_id=import_id, account=current_user)
+            session.commit()
+
+        if result.status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/check-dependencies")
+class CustomizedSnippetCheckDependenciesApi(Resource):
+    @console_ns.doc("check_snippet_dependencies")
+    @console_ns.doc(description="Check dependencies for a snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID"})
+    @console_ns.response(200, "Dependencies checked successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, snippet_id: str):
+        """Check dependencies for a snippet."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet_service = _snippet_service()
+        snippet = snippet_service.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.check_dependencies(snippet=snippet)
+
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/use-count/increment")
+class CustomizedSnippetUseCountIncrementApi(Resource):
+    @console_ns.doc("increment_snippet_use_count")
+    @console_ns.doc(description="Increment snippet use count by 1")
+    @console_ns.doc(params={"snippet_id": "Snippet ID"})
+    @console_ns.response(200, "Use count incremented successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self, snippet_id: str):
+        """Increment snippet use count when it is inserted into a workflow."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet_service = _snippet_service()
+        snippet = snippet_service.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            SnippetService.increment_use_count(session=session, snippet=snippet)
+            session.commit()
+            session.refresh(snippet)
+
+        return {"result": "success", "use_count": snippet.use_count}, 200
@@ -20,7 +20,7 @@ from controllers.console.wraps import (
    setup_required,
 )
 from core.db.session_factory import session_factory
-from core.entities.mcp_provider import MCPAuthentication, MCPConfiguration
+from core.entities.mcp_provider import IdentityMode, MCPAuthentication, MCPConfiguration
 from core.mcp.auth.auth_flow import auth, handle_callback
 from core.mcp.error import MCPAuthError, MCPError, MCPRefreshTokenError
 from core.mcp.mcp_client import MCPClient
@@ -210,6 +210,30 @@ class MCPProviderBasePayload(BaseModel):
    configuration: dict[str, Any] | None = Field(default_factory=dict)
    headers: dict[str, Any] | None = Field(default_factory=dict)
    authentication: dict[str, Any] | None = Field(default_factory=dict)
+    # None means "leave unchanged" on update; the controller resolves it to a
+    # concrete IdentityMode before calling the service (see _resolve_identity_mode).
+    identity_mode: IdentityMode | None = None
+
+
+def _resolve_identity_mode(requested: IdentityMode | None, *, current: IdentityMode) -> IdentityMode:
+    """Resolve the effective MCP identity_mode for a create/update request.
+
+    Keeps two API-layer concerns out of the service so the service always
+    receives a concrete value:
+
+    * ``None`` means "leave unchanged" (update semantics) — fall back to
+      ``current`` (``IdentityMode.OFF`` for a brand-new provider).
+    * Identity forwarding is an enterprise-only capability. On non-enterprise
+      deployments any non-OFF value is coerced back to OFF so a persisted row
+      can never imply forwarding that the runtime won't perform. This gates the
+      API surface to match the backend gate in
+      ``MCPTool._forwarding_requested`` — both the API and the backend
+      invocation must be gated on ``dify_config.ENTERPRISE_ENABLED``.
+    """
+    mode = current if requested is None else requested
+    if mode != IdentityMode.OFF and not dify_config.ENTERPRISE_ENABLED:
+        return IdentityMode.OFF
+    return mode


 class MCPProviderCreatePayload(MCPProviderBasePayload):
@@ -1000,6 +1024,7 @@ class ToolProviderMCPApi(Resource):
                headers=payload.headers or {},
                configuration=configuration,
                authentication=authentication,
+                identity_mode=_resolve_identity_mode(payload.identity_mode, current=IdentityMode.OFF),
            )

        # 2) Try to fetch tools immediately after creation so they appear without a second save.
@@ -1054,6 +1079,11 @@ class ToolProviderMCPApi(Resource):
        # Step 3: Perform database update in a transaction
        with sessionmaker(db.engine).begin() as session:
            service = MCPToolManageService(session=session)
+            # Resolve "leave unchanged" (None) against the stored value, and gate
+            # the result on ENTERPRISE_ENABLED — both are API-layer concerns, so
+            # the service receives a concrete IdentityMode.
+            existing = service.get_provider(provider_id=payload.provider_id, tenant_id=current_tenant_id)
+            identity_mode = _resolve_identity_mode(payload.identity_mode, current=IdentityMode(existing.identity_mode))
            service.update_provider(
                tenant_id=current_tenant_id,
                provider_id=payload.provider_id,
@@ -1067,6 +1097,7 @@ class ToolProviderMCPApi(Resource):
                configuration=configuration,
                authentication=authentication,
                validation_result=validation_result,
+                identity_mode=identity_mode,
            )

        return {"result": "success"}
@@ -31,9 +31,9 @@ from controllers.console.wraps import (
 from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from fields.base import ResponseModel
-from libs.helper import TimestampField, dump_response, to_timestamp
+from libs.helper import OptionalTimestampField, TimestampField, dump_response, to_timestamp
 from libs.login import login_required
-from models.account import Account, Tenant, TenantCustomConfigDict, TenantStatus
+from models.account import Account, Tenant, TenantAccountJoin, TenantCustomConfigDict, TenantStatus
 from services.account_service import TenantService
 from services.billing_service import BillingService, SubscriptionPlan
 from services.enterprise.enterprise_service import EnterpriseService
@@ -144,6 +144,7 @@ tenants_fields = {
    "plan": fields.String,
    "status": fields.String,
    "created_at": TimestampField,
+    "last_opened_at": OptionalTimestampField,
    "current": fields.Boolean,
 }

@@ -158,7 +159,12 @@ class TenantListApi(Resource):
    @with_current_user
    @with_current_tenant_id
    def get(self, current_tenant_id: str, current_user: Account):
-        tenants = TenantService.get_join_tenants(current_user)
+        tenant_rows: list[tuple[Tenant, TenantAccountJoin]] = [
+            (tenant, membership)
+            for tenant, membership in TenantService.get_workspaces_for_account(db.session, current_user.id)
+            if tenant.status == TenantStatus.NORMAL
+        ]
+        tenants = [tenant for tenant, _ in tenant_rows]
        tenant_dicts = []
        is_enterprise_only = dify_config.ENTERPRISE_ENABLED and not dify_config.BILLING_ENABLED
        is_saas = dify_config.EDITION == "CLOUD" and dify_config.BILLING_ENABLED
@@ -171,7 +177,7 @@ class TenantListApi(Resource):
                if not tenant_plans:
                    logger.warning("get_plan_bulk returned empty result, falling back to legacy feature path")

-        for tenant in tenants:
+        for tenant, membership in tenant_rows:
            plan: str = CloudPlan.SANDBOX
            if is_saas:
                tenant_plan = tenant_plans.get(tenant.id)
@@ -190,6 +196,7 @@ class TenantListApi(Resource):
                "name": tenant.name,
                "status": tenant.status,
                "created_at": tenant.created_at,
+                "last_opened_at": membership.last_opened_at,
                "plan": plan,
                "current": tenant.id == current_tenant_id if current_tenant_id else False,
            }
@@ -1,3 +1,3 @@
-from fastopenapi.routers import FlaskRouter
+from fastopenapi.routers.flask import FlaskRouter

 console_router = FlaskRouter()
@@ -10,7 +10,7 @@ from typing import TYPE_CHECKING, Any, Literal, overload
 from flask import Flask, current_app
 from pydantic import ValidationError
 from sqlalchemy import select
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session, sessionmaker

 import contexts
 from configs import dify_config
@@ -68,6 +68,25 @@ def _extract_trace_session_id_from_debug_args(args: Mapping[str, Any] | Any) ->


 class WorkflowAppGenerator(BaseAppGenerator):
+    @staticmethod
+    def _ensure_snippet_start_node_in_worker(*, session: Session, workflow: Workflow) -> Workflow:
+        """Re-apply snippet virtual Start injection after worker reloads workflow from DB."""
+        if workflow.kind_or_standard != "snippet":
+            return workflow
+
+        from models.snippet import CustomizedSnippet
+        from services.snippet_generate_service import SnippetGenerateService
+
+        snippet = session.scalar(
+            select(CustomizedSnippet).where(
+                CustomizedSnippet.id == workflow.app_id,
+                CustomizedSnippet.tenant_id == workflow.tenant_id,
+            )
+        )
+        if snippet is None:
+            return workflow
+        return SnippetGenerateService.ensure_start_node_for_worker(workflow, snippet)
+
    @staticmethod
    def _should_prepare_user_inputs(args: Mapping[str, Any]) -> bool:
        return not bool(args.get(SKIP_PREPARE_USER_INPUTS_KEY))
@@ -592,6 +611,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
                if workflow is None:
                    raise ValueError("Workflow not found")

+                workflow = self._ensure_snippet_start_node_in_worker(session=session, workflow=workflow)
+
                # Determine system_user_id based on invocation source
                is_external_api_call = application_generate_entity.invoke_from in {
                    InvokeFrom.WEB_APP,
@@ -11,6 +11,7 @@ from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, Workfl
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from core.workflow.node_factory import get_default_root_node_id
 from core.workflow.nodes.agent_v2.session_cleanup_layer import build_workflow_agent_session_cleanup_layer
+from core.workflow.snippet_start import get_compatible_start_aliases
 from core.workflow.system_variables import build_bootstrap_variables, build_system_variables
 from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
@@ -118,7 +119,15 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                ),
            )
            root_node_id = self._root_node_id or get_default_root_node_id(self._workflow.graph_dict)
-            add_node_inputs_to_pool(variable_pool, node_id=root_node_id, inputs=inputs)
+            add_node_inputs_to_pool(
+                variable_pool,
+                node_id=root_node_id,
+                inputs=inputs,
+                aliases=get_compatible_start_aliases(
+                    workflow_kind=getattr(self._workflow, "kind_or_standard", None),
+                    root_node_id=root_node_id,
+                ),
+            )

            graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
            graph = self._init_graph(
@@ -37,6 +37,13 @@ class MCPSupportGrantType(StrEnum):
    REFRESH_TOKEN = "refresh_token"


+class IdentityMode(StrEnum):
+    """How Dify forwards the end-user's identity to an MCP server."""
+
+    OFF = "off"
+    IDP_TOKEN = "idp_token"
+
+
 class MCPAuthentication(BaseModel):
    client_id: str
    client_secret: str | None = None
@@ -76,6 +83,8 @@ class MCPProviderEntity(BaseModel):
    created_at: datetime
    updated_at: datetime

+    identity_mode: IdentityMode = IdentityMode.OFF
+
    @classmethod
    def from_db_model(cls, db_provider: MCPToolProvider) -> MCPProviderEntity:
        """Create entity from database model with decryption"""
@@ -96,6 +105,7 @@ class MCPProviderEntity(BaseModel):
            icon=db_provider.icon or "",
            created_at=db_provider.created_at,
            updated_at=db_provider.updated_at,
+            identity_mode=IdentityMode(db_provider.identity_mode),
        )

    @property
@@ -170,6 +180,7 @@ class MCPProviderEntity(BaseModel):
            "updated_at": int(self.updated_at.timestamp()),
            "label": I18nObject(en_US=self.name, zh_Hans=self.name).to_dict(),
            "description": I18nObject(en_US="", zh_Hans="").to_dict(),
+            "identity_mode": self.identity_mode,
        }

        # Add configuration
@@ -316,6 +316,7 @@ class IndexingRunner:
        qa_preview_texts: list[QAPreviewDetail] = []

        total_segments = 0
+        deleted_preview_images = False
        # doc_form represents the segmentation method (general, parent-child, QA)
        index_type = doc_form
        index_processor = IndexProcessorFactory(index_type).init_index_processor()
@@ -368,6 +369,10 @@ class IndexingRunner:
                            upload_file_id,
                        )
                    db.session.delete(image_file)
+                    deleted_preview_images = True
+
+        if deleted_preview_images:
+            db.session.commit()

        if doc_form and doc_form == "qa_model":
            return IndexingEstimate(total_segments=total_segments * 20, qa_preview=qa_preview_texts, preview=[])
@@ -40,6 +40,7 @@ class MCPClientWithAuthRetry(MCPClient):
        provider_entity: MCPProviderEntity | None = None,
        authorization_code: str | None = None,
        by_server_id: bool = False,
+        forward_identity_active: bool = False,
    ):
        """
        Initialize the MCP client with auth retry capability.
@@ -52,12 +53,15 @@ class MCPClientWithAuthRetry(MCPClient):
            provider_entity: Provider entity for authentication
            authorization_code: Optional authorization code for initial auth
            by_server_id: Whether to look up provider by server ID
+            forward_identity_active: If True, suppress the static-OAuth retry
+                on 401 — the forwarded identity must propagate as-is.
        """
        super().__init__(server_url, headers, timeout, sse_read_timeout)

        self.provider_entity = provider_entity
        self.authorization_code = authorization_code
        self.by_server_id = by_server_id
+        self.forward_identity_active = forward_identity_active
        self._has_retried = False

    def _handle_auth_error(self, error: MCPAuthError) -> None:
@@ -73,6 +77,8 @@ class MCPClientWithAuthRetry(MCPClient):
        Raises:
            MCPAuthError: If authentication fails or max retries reached
        """
+        if self.forward_identity_active:
+            raise error
        if not self.provider_entity:
            raise error
        if self._has_retried:
@@ -7,7 +7,7 @@ import threading
 import time
 from collections.abc import Mapping
 from datetime import timedelta
-from typing import TYPE_CHECKING, Any, TypedDict
+from typing import TYPE_CHECKING, Any, TypedDict, override
 from uuid import UUID, uuid4

 from cachetools import LRUCache
@@ -221,9 +221,10 @@ class TracingProviderConfigEntry(TypedDict):


 class OpsTraceProviderConfigMap(collections.UserDict[str, TracingProviderConfigEntry]):
-    def __getitem__(self, provider: str) -> TracingProviderConfigEntry:
+    @override
+    def __getitem__(self, key: str) -> TracingProviderConfigEntry:
        try:
-            match provider:
+            match key:
                case TracingProviderEnum.LANGFUSE:
                    from dify_trace_langfuse.config import LangfuseConfig
                    from dify_trace_langfuse.langfuse_trace import LangFuseDataTrace
@@ -330,9 +331,9 @@ class OpsTraceProviderConfigMap(collections.UserDict[str, TracingProviderConfigE
                    }

                case _:
-                    raise KeyError(f"Unsupported tracing provider: {provider}")
+                    raise KeyError(f"Unsupported tracing provider: {key}")
        except ImportError:
-            raise ImportError(f"Provider {provider} is not installed.")
+            raise ImportError(f"Provider {key} is not installed.")


 provider_config_map = OpsTraceProviderConfigMap()
@@ -168,6 +168,7 @@ class PluginInstallTask(BasePluginEntity):
 class PluginInstallTaskStartResponse(BaseModel):
    all_installed: bool = Field(description="Whether all plugins are installed.")
    task_id: str = Field(description="The ID of the install task.")
+    task: PluginInstallTask | None = Field(default=None, description="The install task.")


 class PluginVerification(BaseModel):
@@ -206,6 +207,11 @@ class PluginListResponse(BaseModel):
    total: int


+class PluginListWithoutTotalResponse(BaseModel):
+    list: list[PluginEntity]
+    has_more: bool
+
+
 class PluginDynamicSelectOptionsResponse(BaseModel):
    options: Sequence[PluginParameterOption] = Field(description="The options of the dynamic select.")

@@ -6,6 +6,7 @@ from requests import HTTPError
 from core.plugin.entities.bundle import PluginBundleDependency
 from core.plugin.entities.plugin import (
    MissingPluginDependency,
+    PluginCategory,
    PluginDeclaration,
    PluginEntity,
    PluginInstallation,
@@ -16,6 +17,7 @@ from core.plugin.entities.plugin_daemon import (
    PluginInstallTask,
    PluginInstallTaskStartResponse,
    PluginListResponse,
+    PluginListWithoutTotalResponse,
    PluginReadmeResponse,
 )
 from core.plugin.impl.base import BasePluginClient
@@ -74,6 +76,16 @@ class PluginInstaller(BasePluginClient):
            params={"page": page, "page_size": page_size, "response_type": "paged"},
        )

+    def list_plugins_by_category(
+        self, tenant_id: str, category: PluginCategory, page: int, page_size: int
+    ) -> PluginListWithoutTotalResponse:
+        return self._request_with_plugin_daemon_response(
+            "GET",
+            f"plugin/{tenant_id}/management/{category.value}/list",
+            PluginListWithoutTotalResponse,
+            params={"page": page, "page_size": page_size, "response_type": "paged"},
+        )
+
    def upload_pkg(
        self,
        tenant_id: str,
@@ -23,6 +23,7 @@ from core.helper.marketplace import download_plugin_pkg
 from core.helper.model_provider_cache import ProviderCredentialsCache, ProviderCredentialsCacheType
 from core.plugin.entities.bundle import PluginBundleDependency
 from core.plugin.entities.plugin import (
+    PluginCategory,
    PluginDeclaration,
    PluginEntity,
    PluginInstallation,
@@ -33,6 +34,7 @@ from core.plugin.entities.plugin_daemon import (
    PluginInstallTask,
    PluginInstallTaskStatus,
    PluginListResponse,
+    PluginListWithoutTotalResponse,
    PluginModelProviderEntity,
    PluginVerification,
 )
@@ -295,6 +297,19 @@ class PluginService:
        plugins = manager.list_plugins_with_total(tenant_id, page, page_size)
        return plugins

+    @staticmethod
+    def list_by_category(
+        tenant_id: str, category: PluginCategory, page: int, page_size: int
+    ) -> PluginListWithoutTotalResponse:
+        """
+        List plugins in one category with a has-more cursor signal and without calculating total.
+
+        The daemon scans tenant installations in the existing list order and stops once it finds one extra match.
+        This keeps pagination usable before category is persisted on installation rows.
+        """
+        manager = PluginInstaller()
+        return manager.list_plugins_by_category(tenant_id, category, page, page_size)
+
    @staticmethod
    def list_installations_from_ids(tenant_id: str, ids: Sequence[str]) -> Sequence[PluginInstallation]:
        """
@@ -1,7 +1,7 @@
 import base64
 import logging
 import pickle
-from typing import Any, cast
+from typing import Any, cast, override

 import numpy as np
 from sqlalchemy import select
@@ -25,6 +25,7 @@ class CacheEmbedding(Embeddings):
    def __init__(self, model_instance: ModelInstance):
        self._model_instance = model_instance

+    @override
    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        """Embed search docs in batches of 10."""
        # use doc embedding cache or store if not exists
@@ -106,6 +107,7 @@ class CacheEmbedding(Embeddings):

        return text_embeddings

+    @override
    def embed_multimodal_documents(self, multimodel_documents: list[dict[str, Any]]) -> list[list[float]]:
        """Embed file documents."""
        # use doc embedding cache or store if not exists
@@ -189,6 +191,7 @@ class CacheEmbedding(Embeddings):

        return multimodel_embeddings

+    @override
    def embed_query(self, text: str) -> list[float]:
        """Embed query text."""
        # use doc embedding cache or store if not exists
@@ -232,6 +235,7 @@ class CacheEmbedding(Embeddings):

        return embedding_results  # type: ignore

+    @override
    def embed_multimodal_query(self, multimodel_document: dict[str, Any]) -> list[float]:
        """Embed multimodal documents."""
        # use doc embedding cache or store if not exists
@@ -1,13 +1,32 @@
-"""Abstract interface for document loader implementations."""
+"""Excel document extractor used for RAG ingestion.

+Supports cell hyperlinks for both `.xls` and `.xlsx`, and embedded worksheet images
+for `.xlsx` files by converting them into markdown image links. Embedded images are
+stored with deterministic keys derived from the source upload file and anchor cell so
+retries can safely reuse the same assets.
+"""
+
+import hashlib
+import logging
+import mimetypes
 import os
 from typing import TypedDict, override

 import pandas as pd
 from openpyxl import load_workbook
+from sqlalchemy import select

+from configs import dify_config
+from core.db.session_factory import session_factory
 from core.rag.extractor.extractor_base import BaseExtractor
 from core.rag.models.document import Document
+from extensions.ext_storage import storage
+from extensions.storage.storage_type import StorageType
+from libs.datetime_utils import naive_utc_now
+from models.enums import CreatorUserRole
+from models.model import UploadFile
+
+logger = logging.getLogger(__name__)


 class Candidate(TypedDict):
@@ -16,17 +35,42 @@ class Candidate(TypedDict):
    map: dict[int, str]


+class SheetImageCandidate(TypedDict):
+    anchor: tuple[int, int]
+    content_hash: str
+    file_key: str
+    image_bytes: bytes
+    image_ext: str
+
+
 class ExcelExtractor(BaseExtractor):
    """Load Excel files.

-
    Args:
        file_path: Path to the file to load.
    """

-    def __init__(self, file_path: str, encoding: str | None = None, autodetect_encoding: bool = False):
+    _file_path: str
+    _encoding: str | None
+    _autodetect_encoding: bool
+    _tenant_id: str | None
+    _user_id: str | None
+    _source_file_id: str | None
+
+    def __init__(
+        self,
+        file_path: str,
+        tenant_id: str | None = None,
+        user_id: str | None = None,
+        source_file_id: str | None = None,
+        encoding: str | None = None,
+        autodetect_encoding: bool = False,
+    ):
        """Initialize with file path."""
        self._file_path = file_path
+        self._tenant_id = tenant_id
+        self._user_id = user_id
+        self._source_file_id = source_file_id
        self._encoding = encoding
        self._autodetect_encoding = autodetect_encoding

@@ -37,7 +81,8 @@ class ExcelExtractor(BaseExtractor):
        file_extension = os.path.splitext(self._file_path)[-1].lower()

        if file_extension == ".xlsx":
-            wb = load_workbook(self._file_path, read_only=True, data_only=True)
+            # Worksheet drawing objects, including embedded images, are not available in read-only mode.
+            wb = load_workbook(self._file_path, data_only=True)
            try:
                for sheet_name in wb.sheetnames:
                    sheet = wb[sheet_name]
@@ -45,10 +90,15 @@ class ExcelExtractor(BaseExtractor):
                    if not column_map:
                        continue
                    start_row = header_row_idx + 1
+                    sheet_image_map = self._extract_images_from_sheet(
+                        sheet_name=sheet_name,
+                        sheet=sheet,
+                        valid_columns={column_idx + 1 for column_idx in column_map},
+                        min_row=start_row,
+                    )
                    for row in sheet.iter_rows(min_row=start_row, max_col=max_col_idx, values_only=False):
-                        if all(cell.value is None for cell in row):
-                            continue
                        page_content = []
+                        row_has_content = False
                        for col_idx, cell in enumerate(row):
                            value = cell.value
                            if col_idx in column_map:
@@ -56,14 +106,27 @@ class ExcelExtractor(BaseExtractor):
                                if hasattr(cell, "hyperlink") and cell.hyperlink:
                                    target = getattr(cell.hyperlink, "target", None)
                                    if target:
-                                        value = f"[{value}]({target})"
+                                        display_value = value if value is not None and str(value).strip() else target
+                                        value = f"[{display_value}]({target})"
+                                cell_row = getattr(cell, "row", None)
+                                cell_column = getattr(cell, "column", None)
+                                image_links = (
+                                    sheet_image_map.get((cell_row, cell_column), [])
+                                    if isinstance(cell_row, int) and isinstance(cell_column, int)
+                                    else []
+                                )
                                if value is None:
                                    value = ""
                                elif not isinstance(value, str):
                                    value = str(value)
-                                value = value.strip().replace('"', '\\"')
+                                if image_links:
+                                    value = " ".join(filter(None, [value, " ".join(image_links)]))
+                                value = value.strip()
+                                if value:
+                                    row_has_content = True
+                                value = value.replace('"', '\\"')
                                page_content.append(f'"{col_name}":"{value}"')
-                        if page_content:
+                        if row_has_content and page_content:
                            documents.append(
                                Document(page_content=";".join(page_content), metadata={"source": self._file_path})
                            )
@@ -89,6 +152,166 @@ class ExcelExtractor(BaseExtractor):

        return documents

+    def _extract_images_from_sheet(
+        self, sheet_name: str, sheet, valid_columns: set[int], min_row: int
+    ) -> dict[tuple[int, int], list[str]]:
+        """
+        Extract embedded worksheet images and map them to their anchor cell.
+
+        Images are stored with deterministic keys derived from the source upload file,
+        sheet, anchor cell, and content hash so retried tasks can reuse the same
+        UploadFile rows and storage objects.
+        """
+        if not self._tenant_id or not self._user_id or not self._source_file_id:
+            return {}
+
+        images = getattr(sheet, "_images", None) or []
+        image_candidates: list[SheetImageCandidate] = []
+
+        for image in images:
+            marker = getattr(getattr(image, "anchor", None), "_from", None)
+            row_idx = getattr(marker, "row", None)
+            col_idx = getattr(marker, "col", None)
+            if row_idx is None or col_idx is None:
+                continue
+            if row_idx + 1 < min_row or col_idx + 1 not in valid_columns:
+                continue
+
+            image_bytes = self._get_image_bytes(image)
+            if not image_bytes:
+                continue
+
+            image_ext = self._get_image_extension(image)
+            if not image_ext:
+                continue
+
+            anchor_row = row_idx + 1
+            anchor_column = col_idx + 1
+            content_hash = self._hash_image_bytes(image_bytes)
+            image_candidates.append(
+                {
+                    "anchor": (anchor_row, anchor_column),
+                    "content_hash": content_hash,
+                    "file_key": self._build_image_file_key(
+                        sheet_name=sheet_name,
+                        anchor_row=anchor_row,
+                        anchor_column=anchor_column,
+                        content_hash=content_hash,
+                        image_ext=image_ext,
+                    ),
+                    "image_bytes": image_bytes,
+                    "image_ext": image_ext,
+                }
+            )
+
+        if not image_candidates:
+            return {}
+
+        image_map: dict[tuple[int, int], list[str]] = {}
+        base_url = dify_config.FILES_URL
+        candidate_keys = sorted({candidate["file_key"] for candidate in image_candidates})
+
+        with session_factory.create_session() as session:
+            existing_upload_files = session.scalars(
+                select(UploadFile).where(
+                    UploadFile.tenant_id == self._tenant_id,
+                    UploadFile.key.in_(candidate_keys),
+                )
+            ).all()
+            upload_files_by_key = {upload_file.key: upload_file for upload_file in existing_upload_files}
+            new_upload_files: list[UploadFile] = []
+
+            for candidate in image_candidates:
+                upload_file = upload_files_by_key.get(candidate["file_key"])
+                if upload_file is None:
+                    storage.save(candidate["file_key"], candidate["image_bytes"])
+                    mime_type, _ = mimetypes.guess_type(candidate["file_key"])
+                    upload_file = UploadFile(
+                        tenant_id=self._tenant_id,
+                        storage_type=StorageType(dify_config.STORAGE_TYPE),
+                        key=candidate["file_key"],
+                        name=candidate["file_key"],
+                        size=len(candidate["image_bytes"]),
+                        extension=candidate["image_ext"],
+                        mime_type=mime_type or "",
+                        created_by=self._user_id,
+                        created_by_role=CreatorUserRole.ACCOUNT,
+                        created_at=naive_utc_now(),
+                        used=True,
+                        used_by=self._user_id,
+                        used_at=naive_utc_now(),
+                        hash=candidate["content_hash"],
+                    )
+                    upload_files_by_key[candidate["file_key"]] = upload_file
+                    new_upload_files.append(upload_file)
+
+                image_map.setdefault(candidate["anchor"], []).append(
+                    f"![image]({base_url}/files/{upload_file.id}/file-preview)"
+                )
+
+            if new_upload_files:
+                session.add_all(new_upload_files)
+                session.commit()
+
+        return image_map
+
+    @staticmethod
+    def _hash_image_bytes(image_bytes: bytes) -> str:
+        """Return a stable content hash for extracted image bytes."""
+        return hashlib.sha256(image_bytes).hexdigest()
+
+    def _build_image_file_key(
+        self,
+        *,
+        sheet_name: str,
+        anchor_row: int,
+        anchor_column: int,
+        content_hash: str,
+        image_ext: str,
+    ) -> str:
+        """Build a deterministic storage key for an embedded worksheet image."""
+        assert self._tenant_id is not None, "tenant_id is required for image extraction"
+        assert self._source_file_id is not None, "source_file_id is required for image extraction"
+
+        normalized_ext = image_ext.strip().lower()
+        sheet_hash = hashlib.sha256(sheet_name.encode("utf-8")).hexdigest()[:16]
+        return (
+            f"image_files/{self._tenant_id}/{self._source_file_id}/"
+            f"{sheet_hash}_r{anchor_row}_c{anchor_column}_{content_hash}.{normalized_ext}"
+        )
+
+    def _get_image_bytes(self, image) -> bytes | None:
+        """Return embedded image bytes from an openpyxl image object."""
+        data_loader = getattr(image, "_data", None)
+        if not callable(data_loader):
+            return None
+
+        try:
+            data = data_loader()
+            if isinstance(data, bytes):
+                return data
+            if isinstance(data, bytearray):
+                return bytes(data)
+            logger.warning("Unexpected embedded image payload type: %s", type(data).__name__)
+            return None
+        except Exception:
+            logger.warning("Failed to read embedded image bytes from Excel sheet", exc_info=True)
+            return None
+
+    def _get_image_extension(self, image) -> str | None:
+        """Resolve an image extension from openpyxl metadata."""
+        image_format = getattr(image, "format", None)
+        if isinstance(image_format, str) and image_format.strip():
+            return image_format.strip().lower()
+
+        image_path = getattr(image, "path", None)
+        if isinstance(image_path, str):
+            _, extension = os.path.splitext(image_path)
+            if extension:
+                return extension.lstrip(".").lower()
+
+        return None
+
    def _find_header_and_columns(self, sheet, scan_rows=10) -> tuple[int, dict[int, str], int]:
        """
        Scan first N rows to find the most likely header row.
@@ -113,7 +113,12 @@ class ExtractProcessor:
                    unstructured_api_key = dify_config.UNSTRUCTURED_API_KEY or ""

                    if file_extension in {".xlsx", ".xls"}:
-                        extractor = ExcelExtractor(file_path)
+                        extractor = ExcelExtractor(
+                            file_path,
+                            upload_file.tenant_id,
+                            upload_file.created_by,
+                            upload_file.id,
+                        )
                    elif file_extension == ".pdf":
                        assert upload_file is not None
                        extractor = PdfExtractor(file_path, upload_file.tenant_id, upload_file.created_by)
@@ -151,7 +156,12 @@ class ExtractProcessor:
                        extractor = TextExtractor(file_path, autodetect_encoding=True)
                else:
                    if file_extension in {".xlsx", ".xls"}:
-                        extractor = ExcelExtractor(file_path)
+                        extractor = ExcelExtractor(
+                            file_path,
+                            upload_file.tenant_id,
+                            upload_file.created_by,
+                            upload_file.id,
+                        )
                    elif file_extension == ".pdf":
                        assert upload_file is not None
                        extractor = PdfExtractor(file_path, upload_file.tenant_id, upload_file.created_by)
@@ -3,7 +3,7 @@
 import logging
 import re
 import uuid
-from typing import Any, TypedDict, cast
+from typing import Any, TypedDict, cast, override

 logger = logging.getLogger(__name__)

@@ -61,6 +61,7 @@ class ParagraphFormatPreviewDict(TypedDict):


 class ParagraphIndexProcessor(BaseIndexProcessor):
+    @override
    def extract(self, extract_setting: ExtractSetting, **kwargs) -> list[Document]:
        text_docs = ExtractProcessor.extract(
            extract_setting=extract_setting,
@@ -71,6 +72,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):

        return text_docs

+    @override
    def transform(self, documents: list[Document], current_user: Account | None = None, **kwargs) -> list[Document]:
        process_rule = kwargs.get("process_rule")
        if not process_rule:
@@ -120,6 +122,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
            all_documents.extend(split_documents)
        return all_documents

+    @override
    def load(
        self,
        dataset: Dataset,
@@ -142,6 +145,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
            else:
                keyword.add_texts(documents)

+    @override
    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
        # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
        # This method is called for actual deletion scenarios (e.g., when segment is deleted).
@@ -178,6 +182,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
            else:
                keyword.delete()

+    @override
    def retrieve(
        self,
        retrieval_method: RetrievalMethod,
@@ -206,6 +211,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
                docs.append(doc)
        return docs

+    @override
    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
        documents: list[Any] = []
        all_multimodal_documents: list[Any] = []
@@ -271,6 +277,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
                keyword = Keyword(dataset)
                keyword.add_texts(documents)

+    @override
    def format_preview(self, chunks: Any) -> ParagraphFormatPreviewDict:
        if isinstance(chunks, list):
            preview = []
@@ -285,6 +292,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
        else:
            raise ValueError("Chunks is not a list")

+    @override
    def generate_summary_preview(
        self,
        tenant_id: str,
@@ -3,7 +3,7 @@
 import json
 import logging
 import uuid
-from typing import Any, TypedDict
+from typing import Any, TypedDict, override

 from sqlalchemy import delete, select

@@ -44,6 +44,7 @@ class ParentChildFormatPreviewDict(TypedDict):


 class ParentChildIndexProcessor(BaseIndexProcessor):
+    @override
    def extract(self, extract_setting: ExtractSetting, **kwargs) -> list[Document]:
        text_docs = ExtractProcessor.extract(
            extract_setting=extract_setting,
@@ -54,6 +55,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):

        return text_docs

+    @override
    def transform(self, documents: list[Document], current_user: Account | None = None, **kwargs) -> list[Document]:
        process_rule = kwargs.get("process_rule")
        if not process_rule:
@@ -129,6 +131,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):

        return all_documents

+    @override
    def load(
        self,
        dataset: Dataset,
@@ -149,6 +152,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
            if multimodal_documents and dataset.is_multimodal:
                vector.create_multimodal(multimodal_documents)

+    @override
    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
        # node_ids is segment's node_ids
        # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
@@ -219,6 +223,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                    )
                    db.session.commit()

+    @override
    def retrieve(
        self,
        retrieval_method: RetrievalMethod,
@@ -283,6 +288,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                    child_nodes.append(child_document)
        return child_nodes

+    @override
    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
        parent_childs = ParentChildStructureChunk.model_validate(chunks)
        documents = []
@@ -356,6 +362,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                if all_multimodal_documents and dataset.is_multimodal:
                    vector.create_multimodal(all_multimodal_documents)

+    @override
    def format_preview(self, chunks: Any) -> ParentChildFormatPreviewDict:
        parent_childs = ParentChildStructureChunk.model_validate(chunks)
        preview = []
@@ -369,6 +376,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
        }
        return result

+    @override
    def generate_summary_preview(
        self,
        tenant_id: str,
@@ -4,7 +4,7 @@ import logging
 import re
 import threading
 import uuid
-from typing import Any, TypedDict
+from typing import Any, TypedDict, override

 import pandas as pd
 from flask import Flask, current_app
@@ -43,6 +43,7 @@ class QAFormatPreviewDict(TypedDict):


 class QAIndexProcessor(BaseIndexProcessor):
+    @override
    def extract(self, extract_setting: ExtractSetting, **kwargs) -> list[Document]:
        text_docs = ExtractProcessor.extract(
            extract_setting=extract_setting,
@@ -52,6 +53,7 @@ class QAIndexProcessor(BaseIndexProcessor):
        )
        return text_docs

+    @override
    def transform(self, documents: list[Document], current_user: Account | None = None, **kwargs) -> list[Document]:
        preview = kwargs.get("preview")
        process_rule = kwargs.get("process_rule")
@@ -139,6 +141,7 @@ class QAIndexProcessor(BaseIndexProcessor):
            raise ValueError(str(e))
        return text_docs

+    @override
    def load(
        self,
        dataset: Dataset,
@@ -153,6 +156,7 @@ class QAIndexProcessor(BaseIndexProcessor):
            if multimodal_documents and dataset.is_multimodal:
                vector.create_multimodal(multimodal_documents)

+    @override
    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
        # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
        # This method is called for actual deletion scenarios (e.g., when segment is deleted).
@@ -183,6 +187,7 @@ class QAIndexProcessor(BaseIndexProcessor):
        else:
            vector.delete()

+    @override
    def retrieve(
        self,
        retrieval_method: RetrievalMethod,
@@ -211,6 +216,7 @@ class QAIndexProcessor(BaseIndexProcessor):
                docs.append(doc)
        return docs

+    @override
    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
        qa_chunks = QAStructureChunk.model_validate(chunks)
        documents = []
@@ -234,6 +240,7 @@ class QAIndexProcessor(BaseIndexProcessor):
            else:
                raise ValueError("Indexing technique must be high quality.")

+    @override
    def format_preview(self, chunks: Any) -> QAFormatPreviewDict:
        qa_chunks = QAStructureChunk.model_validate(chunks)
        preview = []
@@ -246,6 +253,7 @@ class QAIndexProcessor(BaseIndexProcessor):
        }
        return result

+    @override
    def generate_summary_preview(
        self,
        tenant_id: str,
@@ -1,4 +1,5 @@
 import base64
+from typing import override

 from core.model_manager import ModelInstance, ModelManager
 from core.rag.index_processor.constant.doc_type import DocType
@@ -16,6 +17,7 @@ class RerankModelRunner(BaseRerankRunner):
    def __init__(self, rerank_model_instance: ModelInstance):
        self.rerank_model_instance = rerank_model_instance

+    @override
    def run(
        self,
        query: str,
@@ -1,5 +1,6 @@
 import math
 from collections import Counter
+from typing import override

 import numpy as np

@@ -19,6 +20,7 @@ class WeightRerankRunner(BaseRerankRunner):
        self.tenant_id = tenant_id
        self.weights = weights

+    @override
    def run(
        self,
        query: str,
@@ -5,7 +5,7 @@ from __future__ import annotations
 import codecs
 import re
 from collections.abc import Set as AbstractSet
-from typing import Any, Literal
+from typing import Any, Literal, override

 from core.model_manager import ModelInstance
 from core.rag.splitter.text_splitter import RecursiveCharacterTextSplitter
@@ -51,6 +51,7 @@ class FixedRecursiveCharacterTextSplitter(EnhanceRecursiveCharacterTextSplitter)
        self._fixed_separator = codecs.decode(fixed_separator, "unicode_escape")
        self._separators = separators or ["\n\n", "\n", "。", ". ", " ", ""]

+    @override
    def split_text(self, text: str) -> list[str]:
        """Split incoming text and return chunks."""
        if self._fixed_separator:
@@ -7,7 +7,7 @@ from abc import ABC, abstractmethod
 from collections.abc import Callable, Iterable, Sequence
 from collections.abc import Set as AbstractSet
 from dataclasses import dataclass
-from typing import Any, Literal
+from typing import Any, Literal, override

 from core.rag.models.document import BaseDocumentTransformer, Document

@@ -148,10 +148,12 @@ class TextSplitter(BaseDocumentTransformer, ABC):
            )
        return cls(length_function=lambda x: [_huggingface_tokenizer_length(text) for text in x], **kwargs)

+    @override
    def transform_documents(self, documents: Sequence[Document], **kwargs: Any) -> Sequence[Document]:
        """Transform sequence of documents by splitting them."""
        return self.split_documents(list(documents))

+    @override
    async def atransform_documents(self, documents: Sequence[Document], **kwargs: Any) -> Sequence[Document]:
        """Asynchronously transform a sequence of documents by splitting them."""
        raise NotImplementedError
@@ -211,6 +213,7 @@ class TokenTextSplitter(TextSplitter):
        self._allowed_special: Literal["all"] | AbstractSet[str] = allowed_special
        self._disallowed_special: Literal["all"] | AbstractSet[str] = disallowed_special

+    @override
    def split_text(self, text: str) -> list[str]:
        def _encode(_text: str) -> list[int]:
            return self._tokenizer.encode(
@@ -287,5 +290,6 @@ class RecursiveCharacterTextSplitter(TextSplitter):

        return final_chunks

+    @override
    def split_text(self, text: str) -> list[str]:
        return self._split_text(text, self._separators)
@@ -1,6 +1,6 @@
 from abc import abstractmethod
 from os import listdir, path
-from typing import Any
+from typing import Any, override

 from core.entities.provider_entities import ProviderConfig
 from core.helper.module_import_helper import load_single_subclass_from_source
@@ -105,6 +105,7 @@ class BuiltinToolProviderController(ToolProviderController):
        """
        return self.tools

+    @override
    def get_credentials_schema(self) -> list[ProviderConfig]:
        """
        returns the credentials schema of the provider
@@ -182,6 +183,7 @@ class BuiltinToolProviderController(ToolProviderController):
        )

    @property
+    @override
    def provider_type(self) -> ToolProviderType:
        """
        returns the type of the provider
@@ -1,8 +1,9 @@
-from typing import Any
+from typing import Any, override

 from core.tools.builtin_tool.provider import BuiltinToolProviderController


 class AudioToolProvider(BuiltinToolProviderController):
+    @override
    def _validate_credentials(self, user_id: str, credentials: dict[str, Any]):
        pass
@@ -1,6 +1,6 @@
 import io
 from collections.abc import Generator
-from typing import Any
+from typing import Any, override

 from core.model_manager import ModelManager
 from core.plugin.entities.parameters import PluginParameterOption
@@ -14,6 +14,7 @@ from services.model_provider_service import ModelProviderService


 class ASRTool(BuiltinTool):
+    @override
    def _invoke(
        self,
        user_id: str,
@@ -56,6 +57,7 @@ class ASRTool(BuiltinTool):
                items.append((provider, model.model))
        return items

+    @override
    def get_runtime_parameters(
        self,
        conversation_id: str | None = None,
@@ -1,6 +1,6 @@
 import io
 from collections.abc import Generator
-from typing import Any
+from typing import Any, override

 from core.model_manager import ModelManager
 from core.plugin.entities.parameters import PluginParameterOption
@@ -12,6 +12,7 @@ from services.model_provider_service import ModelProviderService


 class TTSTool(BuiltinTool):
+    @override
    def _invoke(
        self,
        user_id: str,
@@ -66,6 +67,7 @@ class TTSTool(BuiltinTool):
                items.append((provider, model.model, voices))
        return items

+    @override
    def get_runtime_parameters(
        self,
        conversation_id: str | None = None,
@@ -1,8 +1,9 @@
-from typing import Any
+from typing import Any, override

 from core.tools.builtin_tool.provider import BuiltinToolProviderController


 class CodeToolProvider(BuiltinToolProviderController):
+    @override
    def _validate_credentials(self, user_id: str, credentials: dict[str, Any]):
        pass
@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any
+from typing import Any, override

 from core.helper.code_executor.code_executor import CodeExecutor, CodeLanguage
 from core.tools.builtin_tool.tool import BuiltinTool
@@ -8,6 +8,7 @@ from core.tools.errors import ToolInvokeError


 class SimpleCode(BuiltinTool):
+    @override
    def _invoke(
        self,
        user_id: str,
@@ -1,8 +1,9 @@
-from typing import Any
+from typing import Any, override

 from core.tools.builtin_tool.provider import BuiltinToolProviderController


 class WikiPediaProvider(BuiltinToolProviderController):
+    @override
    def _validate_credentials(self, user_id: str, credentials: dict[str, Any]):
        pass
@@ -1,6 +1,6 @@
 from collections.abc import Generator
 from datetime import UTC, datetime
-from typing import Any
+from typing import Any, override

 from pytz import timezone as pytz_timezone  # type: ignore[import-untyped]

@@ -9,6 +9,7 @@ from core.tools.entities.tool_entities import ToolInvokeMessage


 class CurrentTimeTool(BuiltinTool):
+    @override
    def _invoke(
        self,
        user_id: str,
@@ -1,6 +1,6 @@
 from collections.abc import Generator
 from datetime import datetime
-from typing import Any
+from typing import Any, override

 import pytz  # type: ignore[import-untyped]

@@ -10,6 +10,7 @@ from core.tools.errors import ToolInvokeError


 class LocaltimeToTimestampTool(BuiltinTool):
+    @override
    def _invoke(
        self,
        user_id: str,
@@ -1,6 +1,6 @@
 from collections.abc import Generator
 from datetime import datetime
-from typing import Any
+from typing import Any, override

 import pytz  # type: ignore[import-untyped]

@@ -10,6 +10,7 @@ from core.tools.errors import ToolInvokeError


 class TimestampToLocaltimeTool(BuiltinTool):
+    @override
    def _invoke(
        self,
        user_id: str,
@@ -1,6 +1,6 @@
 from collections.abc import Generator
 from datetime import datetime
-from typing import Any
+from typing import Any, override

 import pytz  # type: ignore[import-untyped]

@@ -10,6 +10,7 @@ from core.tools.errors import ToolInvokeError


 class TimezoneConversionTool(BuiltinTool):
+    @override
    def _invoke(
        self,
        user_id: str,
@@ -1,13 +1,14 @@
 import calendar
 from collections.abc import Generator
 from datetime import datetime
-from typing import Any
+from typing import Any, override

 from core.tools.builtin_tool.tool import BuiltinTool
 from core.tools.entities.tool_entities import ToolInvokeMessage


 class WeekdayTool(BuiltinTool):
+    @override
    def _invoke(
        self,
        user_id: str,
@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any
+from typing import Any, override

 from core.tools.builtin_tool.tool import BuiltinTool
 from core.tools.entities.tool_entities import ToolInvokeMessage
@@ -8,6 +8,7 @@ from core.tools.utils.web_reader_tool import get_url


 class WebscraperTool(BuiltinTool):
+    @override
    def _invoke(
        self,
        user_id: str,
@@ -1,9 +1,10 @@
-from typing import Any
+from typing import Any, override

 from core.tools.builtin_tool.provider import BuiltinToolProviderController


 class WebscraperProvider(BuiltinToolProviderController):
+    @override
    def _validate_credentials(self, user_id: str, credentials: dict[str, Any]):
        """
        Validate credentials
@@ -1,5 +1,7 @@
 from __future__ import annotations

+from typing import override
+
 from core.tools.__base.tool import Tool
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.entities.tool_entities import ToolProviderType
@@ -26,6 +28,7 @@ class BuiltinTool(Tool):
        super().__init__(**kwargs)
        self.provider = provider

+    @override
    def fork_tool_runtime(self, runtime: ToolRuntime) -> BuiltinTool:
        """
        fork a new tool with metadata
@@ -56,6 +59,7 @@ class BuiltinTool(Tool):
            caller_user_id=self.runtime.user_id,
        )

+    @override
    def tool_provider_type(self) -> ToolProviderType:
        return ToolProviderType.BUILT_IN

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

+from typing import override
+
 from pydantic import Field
 from sqlalchemy import select

@@ -122,6 +124,7 @@ class ApiToolProviderController(ToolProviderController):
        )

    @property
+    @override
    def provider_type(self) -> ToolProviderType:
        return ToolProviderType.API

@@ -194,6 +197,7 @@ class ApiToolProviderController(ToolProviderController):
        self.tools = tools
        return tools

+    @override
    def get_tool(self, tool_name: str) -> ApiTool:
        """
        get tool by name
@@ -2,7 +2,7 @@ import json
 from collections.abc import Generator
 from dataclasses import dataclass
 from os import getenv
-from typing import Any, Union
+from typing import Any, Union, override
 from urllib.parse import urlencode

 import httpx
@@ -45,6 +45,7 @@ class ApiTool(Tool):
        self.api_bundle = api_bundle
        self.provider_id = provider_id

+    @override
    def fork_tool_runtime(self, runtime: ToolRuntime):
        """
        fork a new tool with metadata
@@ -77,6 +78,7 @@ class ApiTool(Tool):
        # For credential validation, always return as string
        return parsed_response.to_string()

+    @override
    def tool_provider_type(self) -> ToolProviderType:
        return ToolProviderType.API

@@ -373,6 +375,7 @@ class ApiTool(Tool):
        except ValueError:
            return value

+    @override
    def _invoke(
        self,
        user_id: str,
@@ -54,6 +54,9 @@ class ToolProviderApiEntity(BaseModel):
    configuration: MCPConfiguration | None = Field(
        default=None, description="The timeout and sse_read_timeout of the MCP tool"
    )
+    # M3 — user-identity forwarding selector. Round-tripped through the
+    # console API so the create/edit modal can hydrate the toggle state.
+    identity_mode: str = Field(default="off", description="Identity-forwarding mechanism: 'off' or 'idp_token'")
    # Workflow
    workflow_app_id: str | None = Field(default=None, description="The app id of the workflow tool")

@@ -92,6 +95,9 @@ class ToolProviderApiEntity(BaseModel):
                optional_fields.update(self.optional_field("is_dynamic_registration", self.is_dynamic_registration))
                optional_fields.update(self.optional_field("masked_headers", self.masked_headers))
                optional_fields.update(self.optional_field("original_headers", self.original_headers))
+                # M3 — forwarding selector. Always emit ("off" is a valid
+                # value that the UI must hydrate, not skip).
+                optional_fields["identity_mode"] = self.identity_mode
            case ToolProviderType.WORKFLOW:
                optional_fields.update(self.optional_field("workflow_app_id", self.workflow_app_id))
            case _:
@@ -1,6 +1,6 @@
-from typing import Any, Self
+from typing import Any, Self, override

-from core.entities.mcp_provider import MCPProviderEntity
+from core.entities.mcp_provider import IdentityMode, MCPProviderEntity
 from core.mcp.types import Tool as RemoteMCPTool
 from core.tools.__base.tool_provider import ToolProviderController
 from core.tools.__base.tool_runtime import ToolRuntime
@@ -28,6 +28,7 @@ class MCPToolProviderController(ToolProviderController):
        headers: dict[str, str] | None = None,
        timeout: float | None = None,
        sse_read_timeout: float | None = None,
+        identity_mode: IdentityMode = IdentityMode.OFF,
    ):
        super().__init__(entity)
        self.entity: ToolProviderEntityWithPlugin = entity
@@ -37,8 +38,10 @@ class MCPToolProviderController(ToolProviderController):
        self.headers = headers or {}
        self.timeout = timeout
        self.sse_read_timeout = sse_read_timeout
+        self.identity_mode: IdentityMode = identity_mode

    @property
+    @override
    def provider_type(self) -> ToolProviderType:
        """
        returns the type of the provider
@@ -105,6 +108,7 @@ class MCPToolProviderController(ToolProviderController):
            headers=entity.headers,
            timeout=entity.timeout,
            sse_read_timeout=entity.sse_read_timeout,
+            identity_mode=entity.identity_mode,
        )

    def _validate_credentials(self, user_id: str, credentials: dict[str, Any]):
@@ -113,6 +117,7 @@ class MCPToolProviderController(ToolProviderController):
        """
        pass

+    @override
    def get_tool(self, tool_name: str) -> MCPTool:
        """
        return tool with given name
@@ -134,6 +139,7 @@ class MCPToolProviderController(ToolProviderController):
            headers=self.headers,
            timeout=self.timeout,
            sse_read_timeout=self.sse_read_timeout,
+            identity_mode=self.identity_mode,
        )

    def get_tools(self) -> list[MCPTool]:
@@ -151,6 +157,7 @@ class MCPToolProviderController(ToolProviderController):
                headers=self.headers,
                timeout=self.timeout,
                sse_read_timeout=self.sse_read_timeout,
+                identity_mode=self.identity_mode,
            )
            for tool_entity in self.entity.tools
        ]
@@ -4,8 +4,10 @@ import base64
 import json
 import logging
 from collections.abc import Generator, Mapping
-from typing import Any, cast
+from typing import Any, cast, override

+from configs import dify_config
+from core.entities.mcp_provider import IdentityMode
 from core.mcp.auth_client import MCPClientWithAuthRetry
 from core.mcp.error import MCPConnectionError
 from core.mcp.types import (
@@ -25,6 +27,11 @@ from graphon.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetada

 logger = logging.getLogger(__name__)

+# Custom header used to carry the forwarded SSO access token. Picked to avoid
+# stomping on the workspace-scoped Authorization header (provider OAuth /
+# user-supplied custom credentials), which would silently break those flows.
+FORWARDED_IDENTITY_HEADER = "X-Dify-SSO-Access-Token"
+

 class MCPTool(Tool):
    def __init__(
@@ -38,6 +45,7 @@ class MCPTool(Tool):
        headers: dict[str, str] | None = None,
        timeout: float | None = None,
        sse_read_timeout: float | None = None,
+        identity_mode: IdentityMode = IdentityMode.OFF,
    ):
        super().__init__(entity, runtime)
        self.tenant_id = tenant_id
@@ -47,11 +55,14 @@ class MCPTool(Tool):
        self.headers = headers or {}
        self.timeout = timeout
        self.sse_read_timeout = sse_read_timeout
+        self.identity_mode: IdentityMode = identity_mode
        self._latest_usage = LLMUsage.empty_usage()

+    @override
    def tool_provider_type(self) -> ToolProviderType:
        return ToolProviderType.MCP

+    @override
    def _invoke(
        self,
        user_id: str,
@@ -60,7 +71,7 @@ class MCPTool(Tool):
        app_id: str | None = None,
        message_id: str | None = None,
    ) -> Generator[ToolInvokeMessage, None, None]:
-        result = self.invoke_remote_mcp_tool(tool_parameters)
+        result = self.invoke_remote_mcp_tool(tool_parameters, user_id=user_id, app_id=app_id)

        # Extract usage metadata from MCP protocol's _meta field
        self._latest_usage = self._derive_usage_from_result(result)
@@ -223,6 +234,7 @@ class MCPTool(Tool):
                            return found
        return None

+    @override
    def fork_tool_runtime(self, runtime: ToolRuntime) -> MCPTool:
        return MCPTool(
            entity=self.entity,
@@ -234,6 +246,7 @@ class MCPTool(Tool):
            headers=self.headers,
            timeout=self.timeout,
            sse_read_timeout=self.sse_read_timeout,
+            identity_mode=self.identity_mode,
        )

    def _handle_none_parameter(self, parameter: dict[str, Any]) -> dict[str, Any]:
@@ -246,7 +259,26 @@ class MCPTool(Tool):
            if value is not None and not (isinstance(value, str) and value.strip() == "")
        }

-    def invoke_remote_mcp_tool(self, tool_parameters: dict[str, Any]) -> CallToolResult:
+    @property
+    def _forwarding_requested(self) -> bool:
+        """True only when the configured identity_mode wants forwarding AND
+        the deployment actually has the enterprise side that can mint tokens.
+        Non-enterprise installs treat the DB value as a no-op — a stale row
+        won't trigger a 5xx against a missing inner-API endpoint."""
+        return self.identity_mode != IdentityMode.OFF and dify_config.ENTERPRISE_ENABLED
+
+    def invoke_remote_mcp_tool(
+        self,
+        tool_parameters: dict[str, Any],
+        user_id: str | None = None,
+        app_id: str | None = None,
+    ) -> CallToolResult:
+        # Fail closed: forwarding requires user_id (refuse before any DB I/O).
+        if self._forwarding_requested and not user_id:
+            raise ToolInvokeError(
+                "Forward-user-identity is enabled for this MCP provider but no end-user context was supplied."
+            )
+
        headers = self.headers.copy() if self.headers else {}
        tool_parameters = self._handle_none_parameter(tool_parameters)

@@ -271,6 +303,15 @@ class MCPTool(Tool):
                if tokens and tokens.access_token:
                    headers["Authorization"] = f"{tokens.token_type.capitalize()} {tokens.access_token}"

+        # Forwarded identity rides in a custom header so workspace-scoped
+        # provider credentials (Authorization / custom Headers) keep working
+        # untouched. The MCP server is expected to read X-Dify-SSO-Access-Token
+        # when identity forwarding is configured.
+        forward_identity_active = False
+        if self._forwarding_requested and user_id:
+            self._inject_forwarded_identity(headers, user_id=user_id, app_id=app_id, audience=server_url)
+            forward_identity_active = True
+
        # Step 2: Session is now closed, perform network operations without holding database connection
        # MCPClientWithAuthRetry will create a new session lazily only if auth retry is needed
        try:
@@ -280,9 +321,44 @@ class MCPTool(Tool):
                timeout=self.timeout,
                sse_read_timeout=self.sse_read_timeout,
                provider_entity=provider_entity,
+                forward_identity_active=forward_identity_active,
            ) as mcp_client:
                return mcp_client.invoke_tool(tool_name=self.entity.identity.name, tool_args=tool_parameters)
        except MCPConnectionError as e:
            raise ToolInvokeError(f"Failed to connect to MCP server: {e}") from e
        except Exception as e:
            raise ToolInvokeError(f"Failed to invoke tool: {e}") from e
+
+    def _inject_forwarded_identity(
+        self,
+        headers: dict[str, str],
+        *,
+        user_id: str,
+        app_id: str | None,
+        audience: str,
+    ) -> None:
+        """Call the enterprise IssueMCPToken endpoint and stamp the issued
+        token into X-Dify-SSO-Access-Token.
+
+        A custom header is used (rather than Authorization) so it composes
+        with workspace-scoped provider credentials — the user may have OAuth
+        tokens or a custom Authorization header configured on the MCP
+        provider, and forwarding must not silently overwrite them.
+
+        Errors are surfaced as ToolInvokeError so the workflow halts with a
+        clear message instead of silently dropping identity and hitting the
+        MCP server unauthenticated.
+        """
+        from services.enterprise.base import MCPTokenError
+        from services.enterprise.enterprise_service import EnterpriseService
+
+        try:
+            token, _expires_at = EnterpriseService.issue_mcp_token(
+                user_id=user_id,
+                tenant_id=self.tenant_id,
+                app_id=app_id,
+                audience=audience,
+            )
+        except MCPTokenError as e:
+            raise ToolInvokeError(f"Failed to obtain forwarded identity token: {e}") from e
+        headers[FORWARDED_IDENTITY_HEADER] = token
@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, override

 from core.plugin.impl.tool import PluginToolManager
 from core.tools.__base.tool_runtime import ToolRuntime
@@ -23,6 +23,7 @@ class PluginToolProviderController(BuiltinToolProviderController):
        self.plugin_unique_identifier = plugin_unique_identifier

    @property
+    @override
    def provider_type(self) -> ToolProviderType:
        """
        returns the type of the provider
@@ -31,6 +32,7 @@ class PluginToolProviderController(BuiltinToolProviderController):
        """
        return ToolProviderType.PLUGIN

+    @override
    def _validate_credentials(self, user_id: str, credentials: dict[str, Any]):
        """
        validate the credentials of the provider
@@ -1,7 +1,7 @@
 from __future__ import annotations

 from collections.abc import Generator
-from typing import Any
+from typing import Any, override

 from core.plugin.impl.tool import PluginToolManager
 from core.plugin.utils.converter import convert_parameters_to_plugin_format
@@ -20,9 +20,11 @@ class PluginTool(Tool):
        self.plugin_unique_identifier = plugin_unique_identifier
        self.runtime_parameters: list[ToolParameter] | None = None

+    @override
    def tool_provider_type(self) -> ToolProviderType:
        return ToolProviderType.PLUGIN

+    @override
    def _invoke(
        self,
        user_id: str,
@@ -48,6 +50,7 @@ class PluginTool(Tool):
            message_id=message_id,
        )

+    @override
    def fork_tool_runtime(self, runtime: ToolRuntime) -> PluginTool:
        return PluginTool(
            entity=self.entity,
@@ -57,6 +60,7 @@ class PluginTool(Tool):
            plugin_unique_identifier=self.plugin_unique_identifier,
        )

+    @override
    def get_runtime_parameters(
        self,
        conversation_id: str | None = None,
@@ -412,7 +412,7 @@ class ToolManager:
            tool_invoke_from=ToolInvokeFrom.AGENT,
            credential_id=agent_tool.credential_id,
        )
-        runtime_parameters = {}
+        runtime_parameters: dict[str, Any] = {}
        parameters = tool_entity.get_merged_runtime_parameters()
        runtime_parameters = cls._convert_tool_parameters_type(
            parameters, variable_pool, agent_tool.tool_parameters, typ="agent"
@@ -501,7 +501,7 @@ class ToolManager:
            tool_invoke_from=ToolInvokeFrom.PLUGIN,
            credential_id=credential_id,
        )
-        runtime_parameters = {}
+        runtime_parameters: dict[str, Any] = {}
        parameters = tool_entity.get_merged_runtime_parameters()
        for parameter in parameters:
            if parameter.form == ToolParameter.ToolParameterForm.FORM:
@@ -1070,7 +1070,7 @@ class ToolManager:
        from graphon.nodes.tool.entities import ToolNodeData
        from graphon.nodes.tool.exc import ToolParameterError

-        runtime_parameters = {}
+        runtime_parameters: dict[str, Any] = {}
        for parameter in parameters:
            if (
                parameter.type
@@ -1,4 +1,5 @@
 import threading
+from typing import override

 from flask import Flask, current_app
 from pydantic import BaseModel, Field
@@ -46,6 +47,7 @@ class DatasetMultiRetrieverTool(DatasetRetrieverBaseTool):
            name=f"dataset_{tenant_id.replace('-', '_')}", tenant_id=tenant_id, dataset_ids=dataset_ids, **kwargs
        )

+    @override
    def _run(self, query: str) -> str:
        threads = []
        all_documents: list[RagDocument] = []
@@ -1,4 +1,4 @@
-from typing import Any, cast
+from typing import Any, cast, override

 from pydantic import BaseModel, Field
 from sqlalchemy import select
@@ -56,6 +56,7 @@ class DatasetRetrieverTool(DatasetRetrieverBaseTool):
            **kwargs,
        )

+    @override
    def _run(self, query: str) -> str:
        dataset_stmt = select(Dataset).where(Dataset.tenant_id == self.tenant_id, Dataset.id == self.dataset_id)
        dataset = db.session.scalar(dataset_stmt)
@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any
+from typing import Any, override

 from core.app.app_config.entities import DatasetRetrieveConfigEntity
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -85,6 +85,7 @@ class DatasetRetrieverTool(Tool):

        return tools

+    @override
    def get_runtime_parameters(
        self,
        conversation_id: str | None = None,
@@ -105,9 +106,11 @@ class DatasetRetrieverTool(Tool):
            ),
        ]

+    @override
    def tool_provider_type(self) -> ToolProviderType:
        return ToolProviderType.DATASET_RETRIEVAL

+    @override
    def _invoke(
        self,
        user_id: str,
@@ -63,7 +63,7 @@ def get_url(url: str, user_agent: str | None = None) -> str:
        response = remote_fetcher.make_request("GET", url, headers=headers, follow_redirects=True, timeout=(120, 300))
    elif response.status_code == 403:
        scraper = cloudscraper.create_scraper()
-        scraper.perform_request = remote_fetcher.make_request
+        object.__setattr__(scraper, "perform_request", remote_fetcher.make_request)
        response = scraper.get(url, headers=headers, timeout=(120, 300))

    if response.status_code != 200:
@@ -1,6 +1,7 @@
 from __future__ import annotations

 from collections.abc import Mapping
+from typing import override

 from pydantic import Field
 from sqlalchemy import select
@@ -80,6 +81,7 @@ class WorkflowToolProviderController(ToolProviderController):
        return controller

    @property
+    @override
    def provider_type(self) -> ToolProviderType:
        return ToolProviderType.WORKFLOW

@@ -3,7 +3,7 @@ from __future__ import annotations
 import json
 import logging
 from collections.abc import Generator, Mapping, Sequence
-from typing import Any, cast
+from typing import Any, cast, override

 from sqlalchemy import select

@@ -67,6 +67,7 @@ class WorkflowTool(Tool):

        super().__init__(entity=entity, runtime=runtime)

+    @override
    def tool_provider_type(self) -> ToolProviderType:
        """
        get the tool provider type
@@ -75,6 +76,7 @@ class WorkflowTool(Tool):
        """
        return ToolProviderType.WORKFLOW

+    @override
    def _invoke(
        self,
        user_id: str,
@@ -206,6 +208,7 @@ class WorkflowTool(Tool):
                            return found
        return None

+    @override
    def fork_tool_runtime(self, runtime: ToolRuntime) -> WorkflowTool:
        """
        fork a new tool with metadata
@@ -363,11 +366,24 @@ class WorkflowTool(Tool):
                            files.append(file_dict)
                    except Exception:
                        logger.exception("Failed to transform file %s", file)
+            elif parameter.type == ToolParameter.ToolParameterType.FILES:
+                value = tool_parameters.get(parameter.name)
+                if not parameter.required and self._is_empty_files_parameter_value(value):
+                    value = []
+                parameters_result[parameter.name] = value
            else:
                parameters_result[parameter.name] = tool_parameters.get(parameter.name)

        return parameters_result, files

+    @staticmethod
+    def _is_empty_files_parameter_value(value: Any) -> bool:
+        """Identify empty optional file-list placeholders before workflow input validation."""
+
+        if value is None or value == "":
+            return True
+        return isinstance(value, list) and all(item is None or item == "" for item in value)
+
    def _extract_files(self, outputs: dict[str, Any]) -> tuple[dict[str, Any], list[File]]:
        """
        extract files from the result
@@ -6,7 +6,7 @@ import time
 from abc import ABC, abstractmethod
 from collections.abc import Mapping
 from datetime import datetime
-from typing import Any
+from typing import Any, override

 from pydantic import BaseModel

@@ -62,6 +62,7 @@ class TriggerDebugEventPoller(ABC):


 class PluginTriggerDebugEventPoller(TriggerDebugEventPoller):
+    @override
    def poll(self) -> TriggerDebugEvent | None:
        from services.trigger.trigger_service import TriggerService

@@ -103,6 +104,7 @@ class PluginTriggerDebugEventPoller(TriggerDebugEventPoller):


 class WebhookTriggerDebugEventPoller(TriggerDebugEventPoller):
+    @override
    def poll(self) -> TriggerDebugEvent | None:
        pool_key = build_webhook_pool_key(
            tenant_id=self.tenant_id,
@@ -190,6 +192,7 @@ class ScheduleTriggerDebugEventPoller(TriggerDebugEventPoller):
            inputs={},
        )

+    @override
    def poll(self) -> TriggerDebugEvent | None:
        schedule_debug_runtime = self.get_or_create_schedule_debug_runtime()
        if schedule_debug_runtime.next_run_at > naive_utc_now():
@@ -1,5 +1,5 @@
 from collections.abc import Mapping
-from typing import Union
+from typing import Union, override

 from core.entities.provider_entities import BasicProviderConfig, ProviderConfig
 from core.helper.provider_cache import ProviderCredentialsCache
@@ -16,6 +16,7 @@ class TriggerProviderCredentialsCache(ProviderCredentialsCache):
    def __init__(self, tenant_id: str, provider_id: str, credential_id: str):
        super().__init__(tenant_id=tenant_id, provider_id=provider_id, credential_id=credential_id)

+    @override
    def _generate_cache_key(self, **kwargs) -> str:
        tenant_id = kwargs["tenant_id"]
        provider_id = kwargs["provider_id"]
@@ -29,6 +30,7 @@ class TriggerProviderOAuthClientParamsCache(ProviderCredentialsCache):
    def __init__(self, tenant_id: str, provider_id: str):
        super().__init__(tenant_id=tenant_id, provider_id=provider_id)

+    @override
    def _generate_cache_key(self, **kwargs) -> str:
        tenant_id = kwargs["tenant_id"]
        provider_id = kwargs["provider_id"]
@@ -41,6 +43,7 @@ class TriggerProviderPropertiesCache(ProviderCredentialsCache):
    def __init__(self, tenant_id: str, provider_id: str, subscription_id: str):
        super().__init__(tenant_id=tenant_id, provider_id=provider_id, subscription_id=subscription_id)

+    @override
    def _generate_cache_key(self, **kwargs) -> str:
        tenant_id = kwargs["tenant_id"]
        provider_id = kwargs["provider_id"]
@@ -12,6 +12,7 @@ examples accurate or the LLM will invent fields.
 """

 import json
+from collections.abc import Iterable
 from typing import Any

 # Per-node-type configuration cheatsheet.
@@ -22,11 +23,24 @@ from typing import Any
 # both ``WorkflowService.sync_draft_workflow``'s structural checks and the
 # runtime entity validation each node performs when the workflow runs.
 #
+# The cheatsheet is assembled DYNAMICALLY per request: the planner decides
+# which node types the workflow needs, and ``build_node_config_cheatsheet``
+# stitches together only the snippets for those types (plus the always-needed
+# wrapper / shared-field / edge-handle preamble, and the containers section
+# when an iteration / loop is planned). This keeps the builder prompt tight —
+# a 3-node summariser no longer carries the schema for 12 unrelated node
+# types — and lets each snippet document its FULL schema (e.g. a "file" start
+# variable's required ``allowed_file_types``) without bloating every prompt.
+#
 # The postprocessor in ``runner.py`` fills missing wrapper fields (``type``,
 # ``positionAbsolute``, ``width``, ``height``, ``sourcePosition`` /
 # ``targetPosition``, edge ``data.sourceType`` / ``data.targetType``), so the
 # LLM only needs to emit semantically meaningful fields.
-NODE_CONFIG_CHEATSHEET = """\
+
+# Always-included preamble: the node/edge wrapper shape and the shared
+# ``data`` fields that apply to every node type, plus the "## Per type" header
+# the per-type snippets slot under.
+_CHEATSHEET_PREAMBLE = """\
 ## Node wrapper (every node, top-level)

    {"id": "node1" (digits + letters only — see "Node IDs" below),
@@ -46,14 +60,26 @@ Children of iteration / loop containers additionally need
     "desc": "<one-liner>",
     "selected": false}

-## Per type — additional "data" fields
+## Per type — additional "data" fields (only the node types in your plan are shown)"""

+
+# node_type → its per-type schema snippet. Keyed by the exact ``node_type``
+# string the planner emits so ``build_node_config_cheatsheet`` can look each
+# one up directly. Iteration / loop are documented in the Containers section
+# (they are subgraphs, not leaf nodes) rather than here.
+_NODE_SNIPPETS: dict[str, str] = {
+    "start": """\
 - start:
    {"variables": [
       {"variable": "url",   "label": "URL",   "type": "text-input",
        "required": true,  "max_length": 256,  "options": []},
       {"variable": "topic", "label": "Topic", "type": "paragraph",
-        "required": false, "max_length": 4096, "options": []}
+        "required": false, "max_length": 4096, "options": []},
+       {"variable": "doc",   "label": "Document", "type": "file",
+        "required": true,
+        "allowed_file_types": ["document"],
+        "allowed_file_upload_methods": ["local_file", "remote_url"],
+        "allowed_file_extensions": []}
    ]}
    EVERY user-supplied value referenced by a downstream node
    (``{{#node-id.var#}}`` in a prompt / answer / template, or
@@ -62,19 +88,29 @@ Children of iteration / loop containers additionally need
    If the planner's ``start_inputs`` list is non-empty, use it verbatim
    (the user prompt section "Start inputs" surfaces it). Types:
    text-input | paragraph | select | number | file | file-list.
+    For a "file" or "file-list" variable you MUST also set
+    ``allowed_file_types`` to a NON-EMPTY subset of
+    ["document", "image", "audio", "video", "custom"] — it is a REQUIRED
+    field and the draft fails to load (showing "supported file types is
+    required") without it. Choose by purpose: ["document"] for text
+    extraction (PDF / Word / PPT / Markdown / …), ["image"] for vision,
+    etc. Always set ``allowed_file_upload_methods`` to
+    ["local_file", "remote_url"]. Only when you include "custom" must you
+    also set ``allowed_file_extensions`` to a non-empty list like
+    [".epub", ".rtf"]; otherwise leave it [].
    In Advanced-Chat mode ``sys.query`` and ``sys.files`` are automatic
    system variables — downstream nodes may reference them; do NOT add
-    them to ``variables``.
-
+    them to ``variables``.""",
+    "end": """\
 - end (Workflow mode only):
    {"outputs": [
        {"variable": "result", "value_selector": ["<src-node-id>", "<out-var>"]}
-    ]}
-
+    ]}""",
+    "answer": """\
 - answer (Advanced Chat mode only):
    {"variables": [],
-     "answer": "<text with {{#<src>.<var>#}} placeholders>"}
-
+     "answer": "<text with {{#<src>.<var>#}} placeholders>"}""",
+    "llm": """\
 - llm:
    {"model": {"provider": "<provider>", "name": "<model>", "mode": "chat",
               "completion_params": {"temperature": 0.7}},
@@ -100,26 +136,26 @@ Children of iteration / loop containers additionally need
            values are the translations.
            Input: {{#node1.text#}}
      * Each placeholder only resolves the variable from its source node —
-        it cannot be a Jinja template or call a function.
-
+        it cannot be a Jinja template or call a function.""",
+    "knowledge-retrieval": """\
 - knowledge-retrieval:
    {"query_variable_selector": ["<src>", "<var>"],
     "query_attachment_selector": [],
     "dataset_ids": [],
     "retrieval_mode": "multiple",
     "multiple_retrieval_config": {"top_k": 4, "score_threshold": null,
-                                   "reranking_enable": false}}
-
+                                   "reranking_enable": false}}""",
+    "code": """\
 - code  (escape hatch — only if no installed tool fits):
    {"code_language": "python3",
     "code": "def main(arg1: str) -> dict:\\n    return {'result': arg1}",
     "variables": [{"variable": "arg1", "value_selector": ["<src>", "<var>"]}],
-     "outputs": {"result": {"type": "string", "children": null}}}
-
+     "outputs": {"result": {"type": "string", "children": null}}}""",
+    "template-transform": """\
 - template-transform:
    {"template": "Hello {{ name }}",
-     "variables": [{"variable": "name", "value_selector": ["<src>", "<var>"]}]}
-
+     "variables": [{"variable": "name", "value_selector": ["<src>", "<var>"]}]}""",
+    "http-request": """\
 - http-request  (escape hatch — only if no installed tool fits):
    {"variables": [], "method": "get", "url": "https://example.com",
     "authorization": {"type": "no-auth", "config": null},
@@ -129,8 +165,8 @@ Children of iteration / loop containers additionally need
     "timeout": {"max_connect_timeout": 0, "max_read_timeout": 0,
                 "max_write_timeout": 0},
     "retry_config": {"retry_enabled": true, "max_retries": 3,
-                      "retry_interval": 100}}
-
+                      "retry_interval": 100}}""",
+    "tool": """\
 - tool  (PREFERRED for external actions when listed in Available tools):
    {"provider_id": "<provider>",            # provider portion of provider/tool
     "provider_type": "builtin",             # exact value from catalogue
@@ -144,8 +180,8 @@ Children of iteration / loop containers additionally need
    Parameter ``type`` is one of:
      "mixed"    — string template referencing variables ({{#...#}})
      "variable" — direct reference, value is ["<src>", "<var>"]
-      "constant" — literal value
-
+      "constant" — literal value""",
+    "if-else": """\
 - if-else:
    {"_targetBranches": [{"id": "true", "name": "IF"},
                         {"id": "false", "name": "ELSE"}],
@@ -158,8 +194,8 @@ Children of iteration / loop containers additionally need
                        "comparison_operator": "is",
                        "value": "<value>"}]}
     ]}
-    Source handle for downstream edges = the case_id ("true" / "false").
-
+    Source handle for downstream edges = the case_id ("true" / "false").""",
+    "question-classifier": """\
 - question-classifier:
    {"query_variable_selector": ["<src>", "<var>"],
     "model": {"provider": "<p>", "name": "<m>", "mode": "chat",
@@ -169,8 +205,8 @@ Children of iteration / loop containers additionally need
     "_targetBranches": [{"id": "1", "name": ""}, {"id": "2", "name": ""}],
     "vision": {"enabled": false},
     "instruction": ""}
-    Source handle for downstream edges = the class_id ("1" / "2" / ...).
-
+    Source handle for downstream edges = the class_id ("1" / "2" / ...).""",
+    "parameter-extractor": """\
 - parameter-extractor:
    {"query": [["<src>", "<var>"]],          # array of value_selector arrays
     "model": {"provider": "<p>", "name": "<m>", "mode": "chat",
@@ -179,8 +215,8 @@ Children of iteration / loop containers additionally need
                     "description": "<purpose>", "required": true}],
     "reasoning_mode": "prompt",
     "vision": {"enabled": false},
-     "instruction": ""}
-
+     "instruction": ""}""",
+    "document-extractor": """\
 - document-extractor:
    {"variable_selector": ["<src>", "<file-var>"],   # a file / file-list input
     "is_array_file": false}                          # true when the input is a
@@ -188,8 +224,9 @@ Children of iteration / loop containers additionally need
    Single output variable ``text``: a string when ``is_array_file`` is false,
    an array of strings (one per file) when it is true. ``variable_selector``
    MUST point at a ``start`` variable declared with type "file" / "file-list"
-    (or ``sys.files`` in Advanced-Chat mode).
-
+    (or ``sys.files`` in Advanced-Chat mode). That start variable MUST set a
+    non-empty ``allowed_file_types`` (use ["document"] for document text).""",
+    "variable-aggregator": """\
 - variable-aggregator  (merge mutually-exclusive branches into one output):
    {"output_type": "string",        # VarType of the merged value — one of
                                     # string | number | object | array[string] |
@@ -200,8 +237,8 @@ Children of iteration / loop containers additionally need
    Output variable: ``output`` (the first branch that actually ran). Place it
    after an ``if-else`` / ``question-classifier`` to rejoin paths before the
    ``end`` / ``answer`` node. Each entry of ``variables`` is a value_selector
-    array, NOT a placeholder string.
-
+    array, NOT a placeholder string.""",
+    "list-operator": """\
 - list-operator  (filter / sort / slice an array variable):
    {"variable": ["<src>", "<array-var>"],
     "filter_by": {"enabled": false, "conditions": []},
@@ -210,8 +247,12 @@ Children of iteration / loop containers additionally need
     "limit": {"enabled": false, "size": 10}}
    Enable only the sub-features you need; ``conditions`` reuse the if-else
    condition shape (key / comparison_operator / value). Outputs: ``result``
-    (the processed array), ``first_record``, ``last_record``.
+    (the processed array), ``first_record``, ``last_record``.""",
+}

+
+# Pulled into the cheatsheet only when an iteration / loop appears in the plan.
+_CONTAINERS_SECTION = """\
 ## Containers — iteration / loop

 These are SUBGRAPH nodes. To use one you MUST emit, in order:
@@ -270,16 +311,59 @@ These are SUBGRAPH nodes. To use one you MUST emit, in order:

 5. The container's incoming/outgoing edges connect to the container's id
   (``nodeK``), NOT to inner nodes. The first inner edge connects from
-   ``nodeKstart``.
+   ``nodeKstart``."""

+
+# Always-included trailer: edge handle conventions for every graph.
+_EDGE_HANDLES_SECTION = """\
 ## Edge handles

 - Most nodes:           sourceHandle "source", targetHandle "target".
 - if-else cases:        sourceHandle is the case_id ("true" / "false" / ...).
 - question-classifier:  sourceHandle is the class_id ("1" / "2" / ...).
 - iteration-start /     sourceHandle "source"; the edge from the *start node
-  loop-start:           is what kicks off the first inner step.
-"""
+  loop-start:           is what kicks off the first inner step."""
+
+
+# Container node types are described in ``_CONTAINERS_SECTION`` rather than as
+# leaf snippets; their presence in a plan pulls that section in.
+_CONTAINER_NODE_TYPES = frozenset({"iteration", "loop"})
+
+
+def build_node_config_cheatsheet(node_types: Iterable[str] | None = None) -> str:
+    """
+    Assemble the builder cheatsheet for exactly the node types in the plan.
+
+    ``node_types`` is the set of ``node_type`` strings the planner chose. We
+    emit the always-on preamble (wrapper / shared fields), then only the
+    per-type snippets for the requested types (``start`` is always included —
+    every graph has one), the Containers section when an iteration / loop is
+    planned, and the edge-handles trailer. Unknown / unrecognised type strings
+    are ignored (the runtime / structural validator catches genuinely bogus
+    types).
+
+    ``None`` returns the FULL cheatsheet (every snippet + containers) — used to
+    build the static back-compat constants below and as a safe fallback.
+    """
+    if node_types is None:
+        requested: set[str] = set(_NODE_SNIPPETS) | set(_CONTAINER_NODE_TYPES)
+    else:
+        requested = {str(t).strip() for t in node_types if str(t).strip()}
+        requested.add("start")  # every workflow has exactly one start node
+
+    parts: list[str] = [_CHEATSHEET_PREAMBLE]
+    # Iterate _NODE_SNIPPETS (not ``requested``) to keep a stable, readable order.
+    parts.extend(snippet for node_type, snippet in _NODE_SNIPPETS.items() if node_type in requested)
+    if requested & _CONTAINER_NODE_TYPES:
+        parts.append(_CONTAINERS_SECTION)
+    parts.append(_EDGE_HANDLES_SECTION)
+    return "\n\n".join(parts) + "\n"
+
+
+# Full cheatsheet (all node types) — retained as a module constant so callers
+# and tests that want the complete reference can import it directly. The
+# dynamic per-request prompt is built by ``get_builder_system_prompt``.
+NODE_CONFIG_CHEATSHEET = build_node_config_cheatsheet()


 _BASE_SYSTEM_PROMPT_HEAD = """You are a Dify workflow builder.
@@ -402,21 +486,24 @@ _ADVANCED_CHAT_MODE_RULES = """# Mode-specific rules — Advanced Chat (Chatflow
 """


-BUILDER_SYSTEM_PROMPT_WORKFLOW = (
-    _BASE_SYSTEM_PROMPT_HEAD
-    + _WORKFLOW_MODE_RULES
-    + _BASE_SYSTEM_PROMPT_TAIL
-    + NODE_CONFIG_CHEATSHEET
-    + _BASE_SYSTEM_PROMPT_FOOTER
-)
+def _assemble_builder_system_prompt(mode: str, node_types: Iterable[str] | None) -> str:
+    """Stitch the builder system prompt for ``mode`` around a cheatsheet built
+    for ``node_types`` (``None`` → full cheatsheet)."""
+    mode_rules = _ADVANCED_CHAT_MODE_RULES if mode == "advanced-chat" else _WORKFLOW_MODE_RULES
+    return (
+        _BASE_SYSTEM_PROMPT_HEAD
+        + mode_rules
+        + _BASE_SYSTEM_PROMPT_TAIL
+        + build_node_config_cheatsheet(node_types)
+        + _BASE_SYSTEM_PROMPT_FOOTER
+    )

-BUILDER_SYSTEM_PROMPT_ADVANCED_CHAT = (
-    _BASE_SYSTEM_PROMPT_HEAD
-    + _ADVANCED_CHAT_MODE_RULES
-    + _BASE_SYSTEM_PROMPT_TAIL
-    + NODE_CONFIG_CHEATSHEET
-    + _BASE_SYSTEM_PROMPT_FOOTER
-)
+
+# Static full-cheatsheet prompts — the back-compat default returned by
+# ``get_builder_system_prompt`` when the caller doesn't pin a node-type set.
+BUILDER_SYSTEM_PROMPT_WORKFLOW = _assemble_builder_system_prompt("workflow", None)
+
+BUILDER_SYSTEM_PROMPT_ADVANCED_CHAT = _assemble_builder_system_prompt("advanced-chat", None)


 BUILDER_USER_PROMPT = """# User instruction
@@ -546,8 +633,16 @@ def format_plan_block(plan_nodes: list[dict[str, Any]]) -> str:
    return "\n".join(lines)


-def get_builder_system_prompt(mode: str) -> str:
-    """Pick the system prompt branch for Workflow vs Advanced Chat."""
-    if mode == "advanced-chat":
-        return BUILDER_SYSTEM_PROMPT_ADVANCED_CHAT
-    return BUILDER_SYSTEM_PROMPT_WORKFLOW
+def get_builder_system_prompt(mode: str, node_types: Iterable[str] | None = None) -> str:
+    """
+    Build the builder system prompt for ``mode``, with a cheatsheet scoped to
+    ``node_types`` (the planner's chosen node types).
+
+    When ``node_types`` is ``None`` we return the cached full-cheatsheet
+    constant (back-compat default). When the runner passes the plan's node-type
+    set we assemble a fresh prompt carrying only the relevant per-type schemas,
+    so the builder isn't handed config for node types the workflow never uses.
+    """
+    if node_types is None:
+        return BUILDER_SYSTEM_PROMPT_ADVANCED_CHAT if mode == "advanced-chat" else BUILDER_SYSTEM_PROMPT_WORKFLOW
+    return _assemble_builder_system_prompt(mode, node_types)
@@ -74,6 +74,21 @@ _DEFAULT_VIEWPORT: GraphViewportDict = {"x": 0.0, "y": 0.0, "zoom": 0.7}
 _DEFAULT_NODE_WIDTH = 244
 _DEFAULT_NODE_HEIGHT = 100

+# Start-node input variable types that carry file uploads. Mirrors
+# ``graphon.variables.input_entities.VariableEntityType.FILE / FILE_LIST``.
+_FILE_VARIABLE_TYPES = frozenset({"file", "file-list"})
+
+# Backstop defaults for a file / file-list start variable when the builder
+# omits the required upload config. ``allowed_file_types`` is a REQUIRED field
+# (Studio rejects the draft with "supported file types is required" when it's
+# empty — see ``config-var/config-modal/utils.ts``); we default to every
+# standard type so no valid upload is rejected. ``custom`` is intentionally
+# excluded because it would in turn require a non-empty
+# ``allowed_file_extensions``. The real fix is the builder now documenting and
+# emitting these fields; this is the safety net that guarantees a loadable draft.
+_DEFAULT_ALLOWED_FILE_TYPES = ("document", "image", "audio", "video")
+_DEFAULT_FILE_UPLOAD_METHODS = ("local_file", "remote_url")
+
 # Token ceiling for the planner call when the caller didn't pin one. The plan
 # is a short JSON node list (a handful of nodes with labels/purposes), so this
 # is generous headroom while still bounding a runaway response. The builder is
@@ -434,7 +449,7 @@ class WorkflowGenerator:
            if isinstance(parsed, dict):
                if attempt > 0:
                    logger.info("Workflow generator: %s JSON parse recovered on retry", stage)
-                return cast(dict[str, Any], parsed)
+                return parsed
            last_detail = f"Non-object JSON: {type(parsed).__name__}"
            logger.info("Workflow generator: %s non-object JSON on attempt %s", stage, attempt + 1)
        raise _StageJSONError(stage, last_detail or "JSON parse failed")
@@ -512,8 +527,15 @@ class WorkflowGenerator:
            tool_catalogue_section=format_builder_tool_catalogue_section(tool_catalogue_text),
            start_inputs_section=format_start_inputs_section(start_inputs or []),
        )
+        # Scope the builder cheatsheet to exactly the node types the planner
+        # chose, so the prompt carries each type's FULL schema (e.g. a file
+        # start variable's required ``allowed_file_types``) without dragging in
+        # config for unrelated node types.
+        plan_node_types = {
+            str(node.get("node_type") or "").strip() for node in plan_nodes if str(node.get("node_type") or "").strip()
+        }
        messages = [
-            SystemPromptMessage(content=get_builder_system_prompt(mode)),
+            SystemPromptMessage(content=get_builder_system_prompt(mode, plan_node_types)),
            UserPromptMessage(content=user_prompt),
        ]
        parsed = cls._invoke_and_parse_json(
@@ -658,6 +680,13 @@ class WorkflowGenerator:
        # variables before we surface them as errors.
        cls._reconcile_variable_references(nodes=nodes, mode=mode)

+        # Schema backstop: a "file" / "file-list" start variable MUST carry a
+        # non-empty ``allowed_file_types`` or Studio refuses to load the draft
+        # ("supported file types is required"). The builder is now told to set
+        # it, but we fill safe defaults for any variable that still lacks it so
+        # the generated workflow always loads and runs.
+        cls._normalize_start_file_variables(nodes=nodes)
+
        return cast(GraphDict, {"nodes": nodes, "edges": deduped_edges, "viewport": viewport})

    # ------------------------------------------------------------------
@@ -693,6 +722,21 @@ class WorkflowGenerator:
    # remapping when we defensively strip hyphens out of LLM-emitted ids.
    _ID_FIELDS: ClassVar = frozenset({"start_node_id", "iteration_id", "loop_id", "parentId"})

+    # ``data`` keys whose value is a plain string list, never a
+    # ``[node_id, var]`` value-selector — so the reference walker must not read
+    # a 2-element one as a selector. ``default`` holds an input's default value;
+    # ``options`` holds select choices; the ``allowed_file_*`` keys hold a file
+    # variable's upload config (types / extensions / methods).
+    _NON_SELECTOR_LIST_KEYS: ClassVar = frozenset(
+        {
+            "default",
+            "options",
+            "allowed_file_types",
+            "allowed_file_extensions",
+            "allowed_file_upload_methods",
+        }
+    )
+
    @classmethod
    def _reconcile_variable_references(cls, *, nodes: list[dict[str, Any]], mode: WorkflowGenerationMode) -> None:
        """
@@ -747,12 +791,16 @@ class WorkflowGenerator:
            # Known selector shapes: 2-element [node_id, var] lists.
            for k, v in value.items():
                # ``value_selector`` / ``query_variable_selector`` / etc.: a
-                # flat 2-element list of strings.
+                # flat 2-element list of strings. Skip keys whose value is a
+                # plain string list that merely HAPPENS to have two entries —
+                # a 2-option ``select`` or a file variable's two allowed upload
+                # methods are NOT ``[node_id, var]`` selectors and must not be
+                # mistaken for references.
                if (
                    isinstance(v, list)
                    and len(v) == 2
                    and all(isinstance(x, str) for x in v)
-                    and k != "default"  # default values for input variables are not selectors
+                    and k not in cls._NON_SELECTOR_LIST_KEYS
                ):
                    node_id, var = v[0].strip(), v[1].strip()
                    if node_id and var:
@@ -923,6 +971,96 @@ class WorkflowGenerator:
            }
        )

+    @classmethod
+    def _normalize_start_file_variables(cls, *, nodes: list[dict[str, Any]]) -> None:
+        """
+        Fill the required upload config on every file / file-list start variable.
+
+        A start variable of type ``file`` / ``file-list`` is invalid without a
+        non-empty ``allowed_file_types`` — Studio rejects the draft with
+        "supported file types is required" (see the front-end validator in
+        ``config-var/config-modal/utils.ts``) and the workflow never runs. The
+        builder prompt now documents these fields, but LLMs still drop them, so
+        we backfill safe defaults here:
+
+          * a start variable a ``document-extractor`` consumes but that wasn't
+            declared as a file type → promoted to ``file`` (or ``file-list``
+            when the extractor's ``is_array_file`` is set), defaulting its
+            allowed types to ``["document"]`` (what extraction needs);
+          * empty / missing ``allowed_file_types`` → every standard file type;
+          * ``custom`` present without ``allowed_file_extensions`` → drop
+            ``custom`` (it would otherwise require a non-empty extension list);
+          * empty / missing ``allowed_file_upload_methods`` → local + remote;
+          * ensure ``allowed_file_extensions`` is at least an empty list.
+
+        Idempotent: a variable that already declares valid file config is left
+        untouched.
+        """
+        start_node = next(
+            (n for n in nodes if (n.get("data") or {}).get("type") == BuiltinNodeTypes.START),
+            None,
+        )
+        if start_node is None:
+            return
+        variables = (start_node.get("data") or {}).get("variables")
+        if not isinstance(variables, list):
+            return
+
+        # Start variables a document-extractor reads → whether it wants an
+        # array (file-list). These MUST be file inputs even if the builder
+        # mistyped them (e.g. declared "paragraph"), or the extractor fails at
+        # run time. ``["document"]`` is the right default for text extraction.
+        extractor_file_vars = cls._document_extractor_start_vars(nodes=nodes, start_id=start_node.get("id", ""))
+
+        for var in variables:
+            if not isinstance(var, dict):
+                continue
+            name = var.get("variable")
+            if name in extractor_file_vars and var.get("type") not in _FILE_VARIABLE_TYPES:
+                var["type"] = "file-list" if extractor_file_vars[name] else "file"
+                var.setdefault("allowed_file_types", ["document"])
+            if var.get("type") not in _FILE_VARIABLE_TYPES:
+                continue
+            allowed_types = var.get("allowed_file_types")
+            if not isinstance(allowed_types, list) or not allowed_types:
+                allowed_types = list(_DEFAULT_ALLOWED_FILE_TYPES)
+                var["allowed_file_types"] = allowed_types
+            # ``custom`` demands a non-empty extension list; without one, drop it
+            # so the variable doesn't trip the "file extensions required" check.
+            extensions = var.get("allowed_file_extensions")
+            has_extensions = isinstance(extensions, list) and bool(extensions)
+            if "custom" in allowed_types and not has_extensions:
+                pruned = [t for t in allowed_types if t != "custom"]
+                var["allowed_file_types"] = pruned or list(_DEFAULT_ALLOWED_FILE_TYPES)
+            methods = var.get("allowed_file_upload_methods")
+            if not isinstance(methods, list) or not methods:
+                var["allowed_file_upload_methods"] = list(_DEFAULT_FILE_UPLOAD_METHODS)
+            if not isinstance(var.get("allowed_file_extensions"), list):
+                var["allowed_file_extensions"] = []
+
+    @classmethod
+    def _document_extractor_start_vars(cls, *, nodes: list[dict[str, Any]], start_id: str) -> dict[str, bool]:
+        """
+        Map start-variable name → ``is_array_file`` for every start variable a
+        ``document-extractor`` node reads via its ``variable_selector``.
+
+        When two extractors read the same variable we keep ``True`` (file-list)
+        if any of them wants an array, since a file-list also satisfies a
+        single-file read.
+        """
+        out: dict[str, bool] = {}
+        if not start_id:
+            return out
+        for node in nodes:
+            data = node.get("data") or {}
+            if data.get("type") != BuiltinNodeTypes.DOCUMENT_EXTRACTOR:
+                continue
+            selector = data.get("variable_selector")
+            if isinstance(selector, list) and len(selector) == 2 and selector[0] == start_id:
+                var_name = selector[1]
+                out[var_name] = out.get(var_name, False) or bool(data.get("is_array_file"))
+        return out
+
    @classmethod
    def _fill_node_defaults(cls, node: dict[str, Any]) -> None:
        """Ensure every node has the wrapper-level fields the Studio canvas needs."""
@@ -10,7 +10,7 @@ from __future__ import annotations
 import enum
 import uuid
 from collections.abc import Mapping, Sequence
-from typing import Annotated, Any, ClassVar, Literal
+from typing import Annotated, Any, ClassVar, Literal, override

 import bleach
 import markdown
@@ -158,6 +158,7 @@ class EmailDeliveryMethod(_DeliveryMethodBase):
    type: Literal[DeliveryMethodType.EMAIL] = DeliveryMethodType.EMAIL
    config: EmailDeliveryConfig

+    @override
    def extract_variable_selectors(self) -> Sequence[Sequence[str]]:
        variable_template_parser = VariableTemplateParser(template=self.config.body)
        selectors: list[Sequence[str]] = []
@@ -195,13 +195,16 @@ class _LazyNodeTypeClassesMapping(MutableMapping[NodeType, Mapping[str, type[Nod
        snapshot.update(self._overrides)
        return snapshot

+    @override
    def __getitem__(self, key: NodeType) -> Mapping[str, type[Node]]:
        return self._snapshot()[key]

+    @override
    def __setitem__(self, key: NodeType, value: Mapping[str, type[Node]]) -> None:
        self._deleted.discard(key)
        self._overrides[key] = value

+    @override
    def __delitem__(self, key: NodeType) -> None:
        if key in self._overrides:
            del self._overrides[key]
@@ -211,9 +214,11 @@ class _LazyNodeTypeClassesMapping(MutableMapping[NodeType, Mapping[str, type[Nod
            return
        raise KeyError(key)

+    @override
    def __iter__(self) -> Iterator[NodeType]:
        return iter(self._snapshot())

+    @override
    def __len__(self) -> int:
        return len(self._snapshot())

@@ -2,7 +2,7 @@ from __future__ import annotations

 from collections.abc import Callable, Generator, Mapping, Sequence
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Literal, cast, overload
+from typing import TYPE_CHECKING, Any, Literal, cast, overload, override

 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -136,6 +136,7 @@ class DifyFileReferenceFactory(FileReferenceFactoryProtocol):
    def __init__(self, run_context: Mapping[str, Any] | DifyRunContext) -> None:
        self._run_context = resolve_dify_run_context(run_context)

+    @override
    def build_from_mapping(self, *, mapping: Mapping[str, Any]):
        return file_factory.build_from_mapping(
            mapping=mapping,
@@ -151,25 +152,31 @@ class DifyPreparedLLM(LLMProtocol):
        self._model_instance = model_instance

    @property
+    @override
    def provider(self) -> str:
        return self._model_instance.provider

    @property
+    @override
    def model_name(self) -> str:
        return self._model_instance.model_name

    @property
+    @override
    def parameters(self) -> Mapping[str, Any]:
        return self._model_instance.parameters

    @parameters.setter
+    @override
    def parameters(self, value: Mapping[str, Any]) -> None:
        self._model_instance.parameters = value

    @property
+    @override
    def stop(self) -> Sequence[str] | None:
        return self._model_instance.stop

+    @override
    def get_model_schema(self) -> AIModelEntity:
        model_schema = cast(LargeLanguageModel, self._model_instance.model_type_instance).get_model_schema(
            self._model_instance.model_name,
@@ -179,6 +186,7 @@ class DifyPreparedLLM(LLMProtocol):
            raise ValueError(f"Model schema not found for {self._model_instance.model_name}")
        return model_schema

+    @override
    def get_llm_num_tokens(self, prompt_messages: Sequence[PromptMessage]) -> int:
        return self._model_instance.get_llm_num_tokens(prompt_messages)

@@ -204,6 +212,7 @@ class DifyPreparedLLM(LLMProtocol):
        stream: Literal[True],
    ) -> Generator[LLMResultChunk, None, None]: ...

+    @override
    def invoke_llm(
        self,
        *,
@@ -243,6 +252,7 @@ class DifyPreparedLLM(LLMProtocol):
        stream: Literal[True],
    ) -> Generator[LLMResultChunkWithStructuredOutput, None, None]: ...

+    @override
    def invoke_llm_with_structured_output(
        self,
        *,
@@ -263,11 +273,13 @@ class DifyPreparedLLM(LLMProtocol):
            stream=stream,
        )

+    @override
    def is_structured_output_parse_error(self, error: Exception) -> bool:
        return isinstance(error, OutputParserError)


 class DifyPromptMessageSerializer(PromptMessageSerializerProtocol):
+    @override
    def serialize(
        self,
        *,
@@ -294,6 +306,7 @@ class DifyRetrieverAttachmentLoader(RetrieverAttachmentLoaderProtocol):
        self._file_reference_factory = file_reference_factory
        self._segment_access_checker = segment_access_checker

+    @override
    def load(self, *, segment_id: str) -> Sequence[File]:
        if not is_retriever_segment_access_granted(segment_id):
            return []
@@ -341,6 +354,7 @@ class DifyToolFileManager(ToolFileManagerProtocol):
        self._manager = ToolFileManager()
        self._conversation_id_getter = conversation_id_getter

+    @override
    def create_file_by_raw(
        self,
        *,
@@ -358,6 +372,7 @@ class DifyToolFileManager(ToolFileManagerProtocol):
            filename=filename,
        )

+    @override
    def get_file_generator_by_tool_file_id(self, tool_file_id: str):
        return self._manager.get_file_generator_by_tool_file_id(tool_file_id)

@@ -394,9 +409,11 @@ class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):
    def file_reference_factory(self) -> FileReferenceFactoryProtocol:
        return self._file_reference_factory

+    @override
    def build_file_reference(self, *, mapping: Mapping[str, Any]):
        return self._file_reference_factory.build_from_mapping(mapping=mapping)

+    @override
    def get_runtime(
        self,
        *,
@@ -447,6 +464,7 @@ class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):
            )
        )

+    @override
    def get_runtime_parameters(
        self,
        *,
@@ -458,6 +476,7 @@ class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):
            for parameter in (tool.get_merged_runtime_parameters() or [])
        ]

+    @override
    def invoke(
        self,
        *,
@@ -503,6 +522,7 @@ class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):

        return self._adapt_messages(transformed_messages, provider_name=provider_name)

+    @override
    def get_usage(
        self,
        *,
@@ -745,6 +765,7 @@ class DifyHumanInputNodeRuntime(HumanInputNodeRuntimeProtocol):
            form_repository=form_repository,
        )

+    @override
    def get_form(self, *, node_id: str) -> HumanInputFormStateProtocol | None:
        repo = self.build_form_repository()
        return repo.get_form(node_id)
@@ -766,6 +787,7 @@ class DifyHumanInputNodeRuntime(HumanInputNodeRuntimeProtocol):
            )
        return restored_data

+    @override
    def create_form(
        self,
        *,
@@ -1,7 +1,7 @@
 from __future__ import annotations

 from collections.abc import Generator, Mapping, Sequence
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, override

 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
 from core.workflow.system_variables import SystemVariableKey, get_system_text
@@ -56,9 +56,11 @@ class AgentNode(Node[AgentNodeData]):
        self._message_transformer = message_transformer

    @classmethod
+    @override
    def version(cls) -> str:
        return "1"

+    @override
    def populate_start_event(self, event) -> None:
        dify_ctx = DifyRunContext.model_validate(self.require_run_context_value(DIFY_RUN_CONTEXT_KEY))
        event.extras["agent_strategy"] = {
@@ -69,6 +71,7 @@ class AgentNode(Node[AgentNodeData]):
            ),
        }

+    @override
    def _run(self) -> Generator[NodeEventBase, None, None]:
        from core.plugin.impl.exc import PluginDaemonClientSideError

@@ -167,6 +170,7 @@ class AgentNode(Node[AgentNodeData]):
            )

    @classmethod
+    @override
    def _extract_variable_selector_to_variable_mapping(
        cls,
        *,
@@ -1,11 +1,14 @@
 from __future__ import annotations

+from typing import override
+
 from factories.agent_factory import get_plugin_agent_strategy

 from .strategy_protocols import AgentStrategyPresentationProvider, AgentStrategyResolver, ResolvedAgentStrategy


 class PluginAgentStrategyResolver(AgentStrategyResolver):
+    @override
    def resolve(
        self,
        *,
@@ -21,6 +24,7 @@ class PluginAgentStrategyResolver(AgentStrategyResolver):


 class PluginAgentStrategyPresentationProvider(AgentStrategyPresentationProvider):
+    @override
    def get_icon(self, *, tenant_id: str, agent_strategy_provider_name: str) -> str | None:
        from core.plugin.impl.plugin import PluginInstaller

@@ -2,7 +2,7 @@ from __future__ import annotations

 import logging
 from collections.abc import Generator, Mapping, Sequence
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, override

 from agenton.compositor import CompositorSessionSnapshot

@@ -101,12 +101,15 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
        self._session_store = session_store

    @classmethod
+    @override
    def version(cls) -> str:
        return "2"

+    @override
    def populate_start_event(self, event) -> None:
        event.extras["agent_node"] = {"version": "2", "agent_node_kind": self.node_data.agent_node_kind}

+    @override
    def _run(self) -> Generator[NodeEventBase, None, None]:
        dify_ctx = DifyRunContext.model_validate(self.require_run_context_value(DIFY_RUN_CONTEXT_KEY))
        workflow_id = self.graph_init_params.workflow_id
@@ -577,6 +580,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
        metadata["agent_backend"] = agent_backend

    @classmethod
+    @override
    def _extract_variable_selector_to_variable_mapping(
        cls,
        *,
@@ -1,7 +1,7 @@
 from __future__ import annotations

 from collections.abc import Mapping
-from typing import Any, Protocol, cast
+from typing import Any, Protocol

 from dify_agent.layers.dify_plugin import (
    DifyPluginCredentialValue,
@@ -197,7 +197,7 @@ class WorkflowAgentPluginToolsBuilder:
            credentials=self._normalize_credentials(runtime.credentials, tool_name=tool_config.tool_name),
            runtime_parameters=runtime_parameters,
            parameters=parameters,
-            parameters_json_schema=cast(dict[str, Any], tool_runtime.get_llm_parameters_json_schema()),
+            parameters_json_schema=tool_runtime.get_llm_parameters_json_schema(),
        )

    @staticmethod
@@ -1,5 +1,5 @@
 from collections.abc import Generator, Mapping, Sequence
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, override

 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
 from core.datasource.datasource_manager import DatasourceManager
@@ -49,10 +49,12 @@ class DatasourceNode(Node[DatasourceNodeData]):
        )
        self.datasource_manager = DatasourceManager

+    @override
    def populate_start_event(self, event) -> None:
        event.provider_id = f"{self.node_data.plugin_id}/{self.node_data.provider_name}"
        event.provider_type = self.node_data.provider_type

+    @override
    def _run(self) -> Generator:
        """
        Run the datasource node
@@ -183,6 +185,7 @@ class DatasourceNode(Node[DatasourceNodeData]):
            )

    @classmethod
+    @override
    def _extract_variable_selector_to_variable_mapping(
        cls,
        *,
@@ -219,5 +222,6 @@ class DatasourceNode(Node[DatasourceNodeData]):
        return result

    @classmethod
+    @override
    def version(cls) -> str:
        return "1"
@@ -1,6 +1,6 @@
 import logging
 from collections.abc import Mapping
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, override

 from core.rag.index_processor.index_processor import IndexProcessor
 from core.rag.index_processor.index_processor_base import SummaryIndexSettingDict
@@ -46,7 +46,8 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
        self.index_processor = IndexProcessor()
        self.summary_index_service = SummaryIndex()

-    def _run(self) -> NodeRunResult:  # type: ignore
+    @override
+    def _run(self) -> NodeRunResult:
        node_data = self.node_data
        variable_pool = self.graph_runtime_state.variable_pool

@@ -145,6 +146,7 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
        return rst

    @classmethod
+    @override
    def version(cls) -> str:
        return "1"

@@ -6,7 +6,7 @@ the workflow node registry.

 import logging
 from collections.abc import Mapping, Sequence
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, override

 from core.app.app_config.entities import DatasetRetrieveConfigEntity
 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
@@ -87,9 +87,11 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
        self._rag_retrieval = DatasetRetrieval()

    @classmethod
+    @override
    def version(cls):
        return "1"

+    @override
    def _run(self) -> NodeRunResult:
        usage = LLMUsage.empty_usage()
        if not self._node_data.query_variable_selector and not self._node_data.query_attachment_selector:
@@ -134,7 +136,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
                inputs=variables,
                process_data={"usage": jsonable_encoder(usage)},
-                outputs=outputs,  # type: ignore
+                outputs=outputs,
                metadata={
                    WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: usage.total_tokens,
                    WorkflowNodeExecutionMetadataKey.TOTAL_PRICE: usage.total_price,
@@ -327,6 +329,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
        )

    @classmethod
+    @override
    def _extract_variable_selector_to_variable_mapping(
        cls,
        *,
@@ -1,5 +1,5 @@
 from collections.abc import Mapping
-from typing import Any
+from typing import Any, override

 from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE
 from core.workflow.variable_prefixes import SYSTEM_VARIABLE_NODE_ID
@@ -15,6 +15,7 @@ class TriggerEventNode(Node[TriggerEventNodeData]):
    execution_type = NodeExecutionType.ROOT

    @classmethod
+    @override
    def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
        return {
            "type": "plugin",
@@ -30,12 +31,15 @@ class TriggerEventNode(Node[TriggerEventNodeData]):
        }

    @classmethod
+    @override
    def version(cls) -> str:
        return "1"

+    @override
    def populate_start_event(self, event) -> None:
        event.provider_id = self.node_data.provider_id

+    @override
    def _run(self) -> NodeRunResult:
        """
        Run the plugin trigger node.
--- a/Show More
+++ b/Show More