fix: can not create empty knowledge (#36336)

Co-authored-by: Joel <iamjoel007@gmail.com>

.agents/skills/component-refactoring/SKILL.md

@@ -63,7 +63,7 @@ pnpm analyze-component <path> --json
 
 ```typescript
 // ❌ Before: Complex state logic in component
-const Configuration: FC = () => {
+function Configuration() {
   const [modelConfig, setModelConfig] = useState<ModelConfig>(...)
   const [datasetConfigs, setDatasetConfigs] = useState<DatasetConfigs>(...)
   const [completionParams, setCompletionParams] = useState<FormValue>({})
@@ -85,7 +85,7 @@ export const useModelConfig = (appId: string) => {
 }
 
 // Component becomes cleaner
-const Configuration: FC = () => {
+function Configuration() {
   const { modelConfig, setModelConfig } = useModelConfig(appId)
   return <div>...</div>
 }
@@ -189,8 +189,6 @@ const Template = useMemo(() => {
 
 **Dify Convention**:
 - This skill is for component decomposition, not query/mutation design.
-- When refactoring data fetching, follow `web/AGENTS.md`.
-- Use `frontend-query-mutation` for contracts, query shape, data-fetching wrappers, query/mutation call-site patterns, conditional queries, invalidation, and mutation error handling.
 - Do not introduce deprecated `useInvalid` / `useReset`.
 - Do not add thin passthrough `useQuery` wrappers during refactoring; only extract a custom hook when it truly orchestrates multiple queries/mutations or shared derived state.
 

.agents/skills/component-refactoring/references/complexity-patterns.md

@@ -60,8 +60,10 @@ const Template = useMemo(() => {
 **After** (complexity: ~3):
 
 ```typescript
+import type { ComponentType } from 'react'
+
 // Define lookup table outside component
-const TEMPLATE_MAP: Record<AppModeEnum, Record<string, FC<TemplateProps>>> = {
+const TEMPLATE_MAP: Record<AppModeEnum, Record<string, ComponentType<TemplateProps>>> = {
   [AppModeEnum.CHAT]: {
     [LanguagesSupported[1]]: TemplateChatZh,
     [LanguagesSupported[7]]: TemplateChatJa,

.agents/skills/component-refactoring/references/component-splitting.md

@@ -65,10 +65,10 @@ interface ConfigurationHeaderProps {
   onPublish: () => void
 }
 
-const ConfigurationHeader: FC<ConfigurationHeaderProps> = ({
+function ConfigurationHeader({
   isAdvancedMode,
   onPublish,
-}) => {
+}: ConfigurationHeaderProps) {
   const { t } = useTranslation()
   
   return (
@@ -136,7 +136,7 @@ const AppInfo = () => {
 }
 
 // ✅ After: Separate view components
-const AppInfoExpanded: FC<AppInfoViewProps> = ({ appDetail, onAction }) => {
+function AppInfoExpanded({ appDetail, onAction }: AppInfoViewProps) {
   return (
     <div className="expanded">
       {/* Clean, focused expanded view */}
@@ -144,7 +144,7 @@ const AppInfoExpanded: FC<AppInfoViewProps> = ({ appDetail, onAction }) => {
   )
 }
 
-const AppInfoCollapsed: FC<AppInfoViewProps> = ({ appDetail, onAction }) => {
+function AppInfoCollapsed({ appDetail, onAction }: AppInfoViewProps) {
   return (
     <div className="collapsed">
       {/* Clean, focused collapsed view */}
@@ -203,12 +203,12 @@ interface AppInfoModalsProps {
   onSuccess: () => void
 }
 
-const AppInfoModals: FC<AppInfoModalsProps> = ({
+function AppInfoModals({
   appDetail,
   activeModal,
   onClose,
   onSuccess,
-}) => {
+}: AppInfoModalsProps) {
   const handleEdit = async (data) => { /* logic */ }
   const handleDuplicate = async (data) => { /* logic */ }
   const handleDelete = async () => { /* logic */ }
@@ -296,7 +296,7 @@ interface OperationItemProps {
   onAction: (id: string) => void
 }
 
-const OperationItem: FC<OperationItemProps> = ({ operation, onAction }) => {
+function OperationItem({ operation, onAction }: OperationItemProps) {
   return (
     <div className="operation-item">
       <span className="icon">{operation.icon}</span>
@@ -435,7 +435,7 @@ interface ChildProps {
   onSubmit: () => void
 }
 
-const Child: FC<ChildProps> = ({ value, onChange, onSubmit }) => {
+function Child({ value, onChange, onSubmit }: ChildProps) {
   return (
     <div>
       <input value={value} onChange={e => onChange(e.target.value)} />

.agents/skills/component-refactoring/references/hook-extraction.md

@@ -112,13 +112,13 @@ export const useModelConfig = ({
 
 ```typescript
 // Before: 50+ lines of state management
-const Configuration: FC = () => {
+function Configuration() {
   const [modelConfig, setModelConfig] = useState<ModelConfig>(...)
   // ... lots of related state and effects
 }
 
 // After: Clean component
-const Configuration: FC = () => {
+function Configuration() {
   const {
     modelConfig,
     setModelConfig,
@@ -159,8 +159,6 @@ const Configuration: FC = () => {
 
 When hook extraction touches query or mutation code, do not use this reference as the source of truth for data-layer patterns.
 
-- Follow `web/AGENTS.md` first.
-- Use `frontend-query-mutation` for contracts, query shape, data-fetching wrappers, query/mutation call-site patterns, conditional queries, invalidation, and mutation error handling.
 - Do not introduce deprecated `useInvalid` / `useReset`.
 - Do not extract thin passthrough `useQuery` hooks; only extract orchestration hooks.
 

.agents/skills/e2e-cucumber-playwright/SKILL.md

@@ -23,7 +23,7 @@ Use this skill for Dify's repository-level E2E suite in `e2e/`. Use [`e2e/AGENTS
    - `e2e/scripts/run-cucumber.ts` and `e2e/cucumber.config.ts` when tags or execution flow matter
 3. Read [`references/playwright-best-practices.md`](references/playwright-best-practices.md) only when locator, assertion, isolation, or waiting choices are involved.
 4. Read [`references/cucumber-best-practices.md`](references/cucumber-best-practices.md) only when scenario wording, step granularity, tags, or expression design are involved.
-5. Re-check official docs with Context7 before introducing a new Playwright or Cucumber pattern.
+5. Re-check official Playwright or Cucumber docs with the available documentation tools before introducing a new framework pattern.
 
 ## Local Rules
 

.agents/skills/frontend-code-review/references/performance.md

@@ -9,18 +9,18 @@ Category: Performance
 
 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.
 
-## Complex prop memoization
+## Complex prop stability
 
-IsUrgent: True
+IsUrgent: False
 Category: Performance
 
 ### Description
 
-Wrap complex prop values (objects, arrays, maps) in `useMemo` prior to passing them into child components to guarantee stable references and prevent unnecessary renders.
+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.
 
 Update this file when adding, editing, or removing Performance rules so the catalog remains accurate.
 
-Wrong:
+Risky:
 
 ```tsx
 <HeavyComp
@@ -31,7 +31,7 @@ Wrong:
 />
 ```
 
-Right:
+Better when stable identity matters:
 
 ```tsx
 const config = useMemo(() => ({

.agents/skills/frontend-query-mutation/SKILL.md

@@ -1,46 +0,0 @@
----
-name: frontend-query-mutation
-description: Guide for implementing Dify frontend query and mutation patterns with TanStack Query and oRPC. Trigger when creating or updating contracts in web/contract, wiring router composition, consuming consoleQuery or marketplaceQuery in components or services, deciding whether to call queryOptions()/mutationOptions() directly or extract a helper or use-* hook, configuring oRPC experimental_defaults/default options, handling conditional queries, cache updates/invalidation, mutation error handling, or migrating legacy service calls to contract-first query and mutation helpers.
----
-
-# Frontend Query & Mutation
-
-## Intent
-
-- Keep contract as the single source of truth in `web/contract/*`.
-- Prefer contract-shaped `queryOptions()` and `mutationOptions()`.
-- Keep default cache behavior with `consoleQuery`/`marketplaceQuery` setup, and keep business orchestration in feature vertical hooks when direct contract calls are not enough.
-- Treat `web/service/use-*` query or mutation wrappers as legacy migration targets, not the preferred destination.
-- Keep abstractions minimal to preserve TypeScript inference.
-
-## Workflow
-
-1. Identify the change surface.
-   - Read `references/contract-patterns.md` for contract files, router composition, client helpers, and query or mutation call-site shape.
-   - Read `references/runtime-rules.md` for conditional queries, default options, cache updates/invalidation, error handling, and legacy migrations.
-   - Read both references when a task spans contract shape and runtime behavior.
-2. Implement the smallest abstraction that fits the task.
-   - Default to direct `useQuery(...)` or `useMutation(...)` calls with oRPC helpers at the call site.
-   - Extract a small shared query helper only when multiple call sites share the same extra options.
-   - Create or keep feature hooks only for real orchestration or shared domain behavior.
-   - When touching thin `web/service/use-*` wrappers, migrate them away when feasible.
-3. Preserve Dify conventions.
-   - Keep contract inputs in `{ params, query?, body? }` shape.
-   - Bind default cache updates/invalidation in `createTanstackQueryUtils(...experimental_defaults...)`; use feature hooks only for workflows that cannot be expressed as default operation behavior.
-   - Prefer `mutate(...)`; use `mutateAsync(...)` only when Promise semantics are required.
-
-## Files Commonly Touched
-
-- `web/contract/console/*.ts`
-- `web/contract/marketplace.ts`
-- `web/contract/router.ts`
-- `web/service/client.ts`
-- legacy `web/service/use-*.ts` files when migrating wrappers away
-- component and hook call sites using `consoleQuery` or `marketplaceQuery`
-
-## References
-
-- Use `references/contract-patterns.md` for contract shape, router registration, query and mutation helpers, and anti-patterns that degrade inference.
-- Use `references/runtime-rules.md` for conditional queries, invalidation, `mutate` versus `mutateAsync`, and legacy migration rules.
-
-Treat this skill as the single query and mutation entry point for Dify frontend work. Keep detailed rules in the reference files instead of duplicating them in project docs.

.agents/skills/frontend-query-mutation/agents/openai.yaml

@@ -1,4 +0,0 @@
-interface:
-  display_name: "Frontend Query & Mutation"
-  short_description: "Dify TanStack Query, oRPC, and default option patterns"
-  default_prompt: "Use this skill when implementing or reviewing Dify frontend contracts, query and mutation call sites, oRPC default options, conditional queries, cache updates/invalidation, or legacy query/mutation migrations."

.agents/skills/frontend-query-mutation/references/contract-patterns.md

@@ -1,129 +0,0 @@
-# Contract Patterns
-
-## Table of Contents
-
-- Intent
-- Minimal structure
-- Core workflow
-- Query usage decision rule
-- Mutation usage decision rule
-- Thin hook decision rule
-- Anti-patterns
-- Contract rules
-- Type export
-
-## Intent
-
-- Keep contract as the single source of truth in `web/contract/*`.
-- Default query usage to call-site `useQuery(consoleQuery|marketplaceQuery.xxx.queryOptions(...))` when endpoint behavior maps 1:1 to the contract.
-- Keep abstractions minimal and preserve TypeScript inference.
-
-## Minimal Structure
-
-```text
-web/contract/
-├── base.ts
-├── router.ts
-├── marketplace.ts
-└── console/
-    ├── billing.ts
-    └── ...other domains
-web/service/client.ts
-```
-
-## Core Workflow
-
-1. Define contract in `web/contract/console/{domain}.ts` or `web/contract/marketplace.ts`.
-   - Use `base.route({...}).output(type<...>())` as the baseline.
-   - Add `.input(type<...>())` only when the request has `params`, `query`, or `body`.
-   - For `GET` without input, omit `.input(...)`; do not use `.input(type<unknown>())`.
-2. Register contract in `web/contract/router.ts`.
-   - Import directly from domain files and nest by API prefix.
-3. Consume from UI call sites via oRPC query utilities.
-
-```typescript
-import { useQuery } from '@tanstack/react-query'
-import { consoleQuery } from '@/service/client'
-
-const invoiceQuery = useQuery(consoleQuery.billing.invoices.queryOptions({
-  staleTime: 5 * 60 * 1000,
-  throwOnError: true,
-  select: invoice => invoice.url,
-}))
-```
-
-## Query Usage Decision Rule
-
-1. Default to direct `*.queryOptions(...)` usage at the call site.
-2. If 3 or more call sites share the same extra options, extract a small query helper, not a `use-*` passthrough hook.
-3. Create or keep feature hooks only for orchestration.
-   - Combine multiple queries or mutations.
-   - Share domain-level derived state or invalidation helpers.
-   - Prefer `web/features/{domain}/hooks/*` for feature-owned workflows.
-4. Treat `web/service/use-{domain}.ts` as legacy.
-   - Do not create new thin service wrappers for oRPC contracts.
-   - When touching existing wrappers, inline direct `consoleQuery` or `marketplaceQuery` consumption when the wrapper is only a passthrough.
-
-```typescript
-const invoicesBaseQueryOptions = () =>
-  consoleQuery.billing.invoices.queryOptions({ retry: false })
-
-const invoiceQuery = useQuery({
-  ...invoicesBaseQueryOptions(),
-  throwOnError: true,
-})
-```
-
-## Mutation Usage Decision Rule
-
-1. Default to mutation helpers from `consoleQuery` or `marketplaceQuery`, for example `useMutation(consoleQuery.billing.bindPartnerStack.mutationOptions(...))`.
-2. If the mutation flow is heavily custom, use oRPC clients as `mutationFn`, for example `consoleClient.xxx` or `marketplaceClient.xxx`, instead of handwritten non-oRPC mutation logic.
-
-```typescript
-const createTagMutation = useMutation(consoleQuery.tags.create.mutationOptions())
-```
-
-## Thin Hook Decision Rule
-
-Remove thin hooks when they only rename a single oRPC query or mutation helper.
-Keep hooks when they orchestrate business behavior across multiple operations, own local workflow state, or normalize a feature-specific API.
-Prefer feature vertical hooks for kept orchestration. Do not move new contract-first wrappers into `web/service/use-*`.
-
-Use:
-
-```typescript
-const deleteTagMutation = useMutation(consoleQuery.tags.delete.mutationOptions())
-```
-
-Keep:
-
-```typescript
-const applyTagBindingsMutation = useApplyTagBindingsMutation()
-```
-
-`useApplyTagBindingsMutation` is acceptable because it coordinates bind and unbind requests, computes deltas, and exposes a feature-level workflow rather than a single endpoint passthrough.
-
-## Anti-Patterns
-
-- Do not wrap `useQuery` with `options?: Partial<UseQueryOptions>`.
-- Do not split local `queryKey` and `queryFn` when oRPC `queryOptions` already exists and fits the use case.
-- Do not create thin `use-*` passthrough hooks for a single endpoint.
-- Do not create business-layer helpers whose only purpose is to call `consoleQuery.xxx.mutationOptions()` or `queryOptions()`.
-- Do not introduce new `web/service/use-*` files for oRPC contract passthroughs.
-- These patterns can degrade inference, especially around `throwOnError` and `select`, and add unnecessary indirection.
-
-## Contract Rules
-
-- Input structure: always use `{ params, query?, body? }`.
-- No-input `GET`: omit `.input(...)`; do not use `.input(type<unknown>())`.
-- Path params: use `{paramName}` in the path and match it in the `params` object.
-- Router nesting: group by API prefix, for example `/billing/*` becomes `billing: {}`.
-- No barrel files: import directly from specific files.
-- Types: import from `@/types/` and use the `type<T>()` helper.
-- Mutations: prefer `mutationOptions`; use explicit `mutationKey` mainly for defaults, filtering, and devtools.
-
-## Type Export
-
-```typescript
-export type ConsoleInputs = InferContractRouterInputs<typeof consoleRouterContract>
-```

.agents/skills/frontend-query-mutation/references/runtime-rules.md

@@ -1,172 +0,0 @@
-# Runtime Rules
-
-## Table of Contents
-
-- Conditional queries
-- oRPC default options
-- Cache invalidation
-- Key API guide
-- `mutate` vs `mutateAsync`
-- Legacy migration
-
-## Conditional Queries
-
-Prefer contract-shaped `queryOptions(...)`.
-When required input is missing, prefer `input: skipToken` instead of placeholder params or non-null assertions.
-Use `enabled` only for extra business gating after the input itself is already valid.
-
-```typescript
-import { skipToken, useQuery } from '@tanstack/react-query'
-
-// Disable the query by skipping input construction.
-function useAccessMode(appId: string | undefined) {
-  return useQuery(consoleQuery.accessControl.appAccessMode.queryOptions({
-    input: appId
-      ? { params: { appId } }
-      : skipToken,
-  }))
-}
-
-// Avoid runtime-only guards that bypass type checking.
-function useBadAccessMode(appId: string | undefined) {
-  return useQuery(consoleQuery.accessControl.appAccessMode.queryOptions({
-    input: { params: { appId: appId! } },
-    enabled: !!appId,
-  }))
-}
-```
-
-## oRPC Default Options
-
-Use `experimental_defaults` in `createTanstackQueryUtils` when a contract operation should always carry shared TanStack Query behavior, such as default stale time, mutation cache writes, or invalidation.
-
-Place defaults at the query utility creation point in `web/service/client.ts`:
-
-```typescript
-export const consoleQuery = createTanstackQueryUtils(consoleClient, {
-  path: ['console'],
-  experimental_defaults: {
-    tags: {
-      create: {
-        mutationOptions: {
-          onSuccess: (tag, _variables, _result, context) => {
-            context.client.setQueryData(
-              consoleQuery.tags.list.queryKey({
-                input: {
-                  query: {
-                    type: tag.type,
-                  },
-                },
-              }),
-              (oldTags: Tag[] | undefined) => oldTags ? [tag, ...oldTags] : oldTags,
-            )
-          },
-        },
-      },
-    },
-  },
-})
-```
-
-Rules:
-
-- Keep defaults inline in the `consoleQuery` or `marketplaceQuery` initialization when they need sibling oRPC key builders.
-- Do not create a wrapper function solely to host `createTanstackQueryUtils`.
-- Do not split defaults into a vertical feature file if that forces handwritten operation paths such as `generateOperationKey(['console', ...])`.
-- Keep feature-level orchestration in the feature vertical; keep query utility lifecycle defaults with the query utility.
-- Prefer call-site callbacks for UI feedback only; shared cache behavior belongs in oRPC defaults when it is tied to a contract operation.
-
-## Cache Invalidation
-
-Bind shared invalidation in oRPC defaults when it is tied to a contract operation.
-Use feature vertical hooks only for multi-operation workflows or domain orchestration that cannot live in a single operation default.
-Components may add UI feedback in call-site callbacks, but they should not decide which queries to invalidate.
-
-Use:
-
-- `.key()` for namespace or prefix invalidation
-- `.queryKey(...)` only for exact cache reads or writes such as `getQueryData` and `setQueryData`
-- `queryClient.invalidateQueries(...)` in mutation `onSuccess`
-
-Do not use deprecated `useInvalid` from `use-base.ts`.
-
-```typescript
-// Feature orchestration owns cache invalidation only when defaults are not enough.
-export const useUpdateAccessMode = () => {
-  const queryClient = useQueryClient()
-
-  return useMutation(consoleQuery.accessControl.updateAccessMode.mutationOptions({
-    onSuccess: () => {
-      queryClient.invalidateQueries({
-        queryKey: consoleQuery.accessControl.appWhitelistSubjects.key(),
-      })
-    },
-  }))
-}
-
-// Component only adds UI behavior.
-updateAccessMode({ appId, mode }, {
-  onSuccess: () => toast.success('...'),
-})
-
-// Avoid putting invalidation knowledge in the component.
-mutate({ appId, mode }, {
-  onSuccess: () => {
-    queryClient.invalidateQueries({
-      queryKey: consoleQuery.accessControl.appWhitelistSubjects.key(),
-    })
-  },
-})
-```
-
-## Key API Guide
-
-- `.key(...)`
-  - Use for partial matching operations.
-  - Prefer it for invalidation, refetch, and cancel patterns.
-  - Example: `queryClient.invalidateQueries({ queryKey: consoleQuery.billing.key() })`
-- `.queryKey(...)`
-  - Use for a specific query's full key.
-  - Prefer it for exact cache addressing and direct reads or writes.
-- `.mutationKey(...)`
-  - Use for a specific mutation's full key.
-  - Prefer it for mutation defaults registration, mutation-status filtering, and devtools grouping.
-
-## `mutate` vs `mutateAsync`
-
-Prefer `mutate` by default.
-Use `mutateAsync` only when Promise semantics are truly required, such as parallel mutations or sequential steps with result dependencies.
-
-Rules:
-
-- Event handlers should usually call `mutate(...)` with `onSuccess` or `onError`.
-- Every `await mutateAsync(...)` must be wrapped in `try/catch`.
-- Do not use `mutateAsync` when callbacks already express the flow clearly.
-
-```typescript
-// Default case.
-mutation.mutate(data, {
-  onSuccess: result => router.push(result.url),
-})
-
-// Promise semantics are required.
-try {
-  const order = await createOrder.mutateAsync(orderData)
-  await confirmPayment.mutateAsync({ orderId: order.id, token })
-  router.push(`/orders/${order.id}`)
-}
-catch (error) {
-  toast.error(error instanceof Error ? error.message : 'Unknown error')
-}
-```
-
-## Legacy Migration
-
-When touching old code, migrate it toward these rules:
-
-| Old pattern | New pattern |
-|---|---|
-| `useInvalid(key)` in service wrappers | oRPC defaults, or a feature vertical hook for real orchestration |
-| component-triggered invalidation after mutation | move invalidation into oRPC defaults or a feature vertical hook |
-| imperative fetch plus manual invalidation | wrap it in `useMutation(...mutationOptions(...))` |
-| `await mutateAsync()` without `try/catch` | switch to `mutate(...)` or add `try/catch` |

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

@@ -5,7 +5,7 @@ description: Generate Vitest + React Testing Library tests for Dify frontend com
 
 # Dify Frontend Testing Skill
 
-This skill enables Claude to generate high-quality, comprehensive frontend tests for the Dify project following established conventions and best practices.
+This skill enables Codex to generate high-quality, comprehensive frontend tests for the Dify project following established conventions and best practices.
 
 > **⚠️ Authoritative Source**: This skill is derived from `web/docs/test.md`. Use Vitest mock/timer APIs (`vi.*`).
 
@@ -24,35 +24,27 @@ Apply this skill when the user:
 **Do NOT apply** when:
 
 - User is asking about backend/API tests (Python/pytest)
-- User is asking about E2E tests (Playwright/Cypress)
+- User is asking about E2E tests (Cucumber + Playwright under `e2e/`)
 - User is only asking conceptual questions without code context
 
 ## Quick Reference
 
-### Tech Stack
-
-| Tool | Version | Purpose |
-|------|---------|---------|
-| Vitest | 4.0.16 | Test runner |
-| React Testing Library | 16.0 | Component testing |
-| jsdom | - | Test environment |
-| nock | 14.0 | HTTP mocking |
-| TypeScript | 5.x | Type safety |
-
 ### Key Commands
 
+Run these commands from `web/`. From the repository root, prefix them with `pnpm -C web`.
+
 ```bash
 # Run all tests
 pnpm test
 
 # Watch mode
-pnpm test:watch
+pnpm test --watch
 
 # Run specific file
 pnpm test path/to/file.spec.tsx
 
 # Generate coverage report
-pnpm test:coverage
+pnpm test --coverage
 
 # Analyze component complexity
 pnpm analyze-component <path>
@@ -228,7 +220,10 @@ Every test should clearly separate:
 ### 2. Black-Box Testing
 
 - Test observable behavior, not implementation details
-- Use semantic queries (getByRole, getByLabelText)
+- Use semantic queries (`getByRole` with accessible `name`, `getByLabelText`, `getByPlaceholderText`, `getByText`, and scoped `within(...)`)
+- Treat `getByTestId` as a last resort. If a control cannot be found by role/name, label, landmark, or dialog scope, fix the component accessibility first instead of adding or relying on `data-testid`.
+- Remove production `data-testid` attributes when semantic selectors can cover the behavior. Keep them only for non-visual mocked boundaries, editor/browser shims such as Monaco, canvas/chart output, or third-party widgets with no accessible DOM in the test environment.
+- Do not assert decorative icons by test id. Assert the named control that contains them, or mark decorative icons `aria-hidden`.
 - Avoid testing internal state directly
 - **Prefer pattern matching over hardcoded strings** in assertions:
 

.agents/skills/frontend-testing/references/mocking.md

@@ -56,7 +56,7 @@ See [Zustand Store Testing](#zustand-store-testing) section for full details.
 
 | Location | Purpose |
 |----------|---------|
-| `web/vitest.setup.ts` | Global mocks shared by all tests (`react-i18next`, `next/image`, `zustand`) |
+| `web/vitest.setup.ts` | Global mocks shared by all tests (`react-i18next`, `zustand`, clipboard, FloatingPortal, Monaco, localStorage`) |
 | `web/__mocks__/zustand.ts` | Zustand mock implementation (auto-resets stores after each test) |
 | `web/__mocks__/` | Reusable mock factories shared across multiple test files |
 | Test file | Test-specific mocks, inline with `vi.mock()` |
@@ -216,28 +216,21 @@ describe('Component', () => {
 })
 ```
 
-### 5. HTTP Mocking with Nock
+### 5. HTTP and `fetch` Mocking
 
 ```typescript
-import nock from 'nock'
-
-const GITHUB_HOST = 'https://api.github.com'
-const GITHUB_PATH = '/repos/owner/repo'
-
-const mockGithubApi = (status: number, body: Record<string, unknown>, delayMs = 0) => {
-  return nock(GITHUB_HOST)
-    .get(GITHUB_PATH)
-    .delay(delayMs)
-    .reply(status, body)
-}
-
 describe('GithubComponent', () => {
-  afterEach(() => {
-    nock.cleanAll()
+  beforeEach(() => {
+    vi.clearAllMocks()
   })
 
   it('should display repo info', async () => {
-    mockGithubApi(200, { name: 'dify', stars: 1000 })
+    vi.mocked(globalThis.fetch).mockResolvedValueOnce(
+      new Response(JSON.stringify({ name: 'dify', stars: 1000 }), {
+        status: 200,
+        headers: { 'Content-Type': 'application/json' },
+      }),
+    )
     
     render(<GithubComponent />)
     
@@ -247,7 +240,12 @@ describe('GithubComponent', () => {
   })
 
   it('should handle API error', async () => {
-    mockGithubApi(500, { message: 'Server error' })
+    vi.mocked(globalThis.fetch).mockResolvedValueOnce(
+      new Response(JSON.stringify({ message: 'Server error' }), {
+        status: 500,
+        headers: { 'Content-Type': 'application/json' },
+      }),
+    )
     
     render(<GithubComponent />)
     
@@ -258,6 +256,8 @@ describe('GithubComponent', () => {
 })
 ```
 
+Prefer mocking `@/service/*` modules or spying on `global.fetch` / `ky` clients with deterministic responses. Do not introduce an HTTP interception dependency such as `nock` or MSW unless it is already declared in the workspace or adding it is part of the task.
+
 ### 6. Context Providers
 
 ```typescript
@@ -332,7 +332,7 @@ const renderWithQueryClient = (ui: React.ReactElement) => {
 1. **Don't mock Zustand store modules** - Use real stores with `setState()`
 1. Don't mock components you can import directly
 1. Don't create overly simplified mocks that miss conditional logic
-1. Don't forget to clean up nock after each test
+1. Don't leave HTTP mocks or service mock state leaking between tests
 1. Don't use `any` types in mocks without necessity
 
 ### Mock Decision Tree

.agents/skills/frontend-testing/references/workflow.md

@@ -227,12 +227,12 @@ Failing tests compound:
 
 **Fix failures immediately before proceeding.**
 
-## Integration with Claude's Todo Feature
+## Integration with Codex's Todo Feature
 
-When using Claude for multi-file testing:
+When using Codex for multi-file testing:
 
-1. **Ask Claude to create a todo list** before starting
-1. **Request one file at a time** or ensure Claude processes incrementally
+1. **Create a todo list** before starting
+1. **Process one file at a time**
 1. **Verify each test passes** before asking for the next
 1. **Mark todos complete** as you progress
 

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

@@ -0,0 +1,71 @@
+---
+name: how-to-write-component
+description: React/TypeScript component style guide. Use when writing, refactoring, or reviewing React components, especially around props typing, state boundaries, shared local state with Jotai atoms, API types, query/mutation contracts, navigation, memoization, wrappers, and empty-state handling.
+---
+
+# How To Write A Component
+
+Use this as the decision guide for React/TypeScript component structure. Existing code is reference material, not automatic precedent; when it conflicts with these rules, adapt the approach instead of reproducing the violation.
+
+## Core Defaults
+
+- Search before adding UI, hooks, helpers, or styling patterns. Reuse existing base components, feature components, hooks, utilities, and design styles when they fit.
+- Group code by feature workflow, route, or ownership area: components, hooks, local types, query helpers, atoms, constants, and small utilities should live near the code that changes with them.
+- Promote code to shared only when multiple verticals need the same stable primitive. Otherwise keep it local and compose shared primitives inside the owning feature.
+- Use Tailwind CSS v4.1+ rules via the `tailwind-css-rules` skill. Prefer v4 utilities, `gap`, `text-size/line-height`, `min-h-dvh`, and avoid deprecated utilities and `@apply`.
+
+## Ownership
+
+- Put local state, queries, mutations, handlers, and derived UI data in the lowest component that uses them. Extract a purpose-built owner component only when the logic has no natural home.
+- Repeated TanStack query calls in sibling components are acceptable when each component independently consumes the data. Do not hoist a query only because it is duplicated; TanStack Query handles deduplication and cache sharing.
+- Hoist state, queries, or callbacks to a parent only when the parent consumes the data, coordinates shared loading/error/empty UI, needs one consistent snapshot, or owns a workflow spanning children.
+- Avoid prop drilling. One pass-through layer is acceptable; repeated forwarding means ownership should move down or into feature-scoped Jotai UI state. Keep server/cache state in query and API data flow.
+- Keep callbacks in a parent only for workflow coordination such as form submission, shared selection, batch behavior, or navigation. Otherwise let the child or row own its action.
+- Prefer uncontrolled DOM state and CSS variables before adding controlled props.
+
+## Components, Props, And Types
+
+- Type component signatures directly; do not use `FC` or `React.FC`.
+- Prefer `function` for top-level components and module helpers. Use arrow functions for local callbacks, handlers, and lambda-style APIs.
+- Prefer named exports. Use default exports only where the framework requires them, such as Next.js route files.
+- Type simple one-off props inline. Use a named `Props` type only when reused, exported, complex, or clearer.
+- Use API-generated or API-returned types at component boundaries. Keep small UI conversion helpers beside the component that needs them.
+- Name values by their domain role and backend API contract, and keep that name stable across the call chain, especially IDs like `appInstanceId`. Normalize framework or route params at the boundary.
+- Keep fallback and invariant checks at the lowest component that already handles that state; callers should pass raw values through instead of duplicating checks.
+
+## Queries And Mutations
+
+- Keep `web/contract/*` as the single source of truth for API shape; follow existing domain/router patterns and the `{ params, query?, body? }` input shape.
+- Consume queries directly with `useQuery(consoleQuery.xxx.queryOptions(...))` or `useQuery(marketplaceQuery.xxx.queryOptions(...))`.
+- Avoid pass-through hooks and thin `web/service/use-*` wrappers that only rename `queryOptions()` or `mutationOptions()`. Extract a small `queryOptions` helper only when repeated call-site options justify it.
+- Keep feature hooks for real orchestration, workflow state, or shared domain behavior.
+- For missing required query input, use `input: skipToken`; use `enabled` only for extra business gating after the input is valid.
+- Consume mutations directly with `useMutation(consoleQuery.xxx.mutationOptions(...))` or `useMutation(marketplaceQuery.xxx.mutationOptions(...))`; use oRPC clients as `mutationFn` only for custom flows.
+- Put shared cache behavior in `createTanstackQueryUtils(...experimental_defaults...)`; components may add UI feedback callbacks, but should not own shared invalidation rules.
+- Do not use deprecated `useInvalid` or `useReset`.
+- Prefer `mutate(...)`; use `mutateAsync(...)` only when Promise semantics are required, and wrap awaited calls in `try/catch`.
+
+## Component Boundaries
+
+- Use the first level below a page or tab to organize independent page sections when it adds real structure. This layer is layout/semantic first, not automatically the data owner.
+- Split deeper components by the data and state each layer actually needs. Each component should access only necessary data, and ownership should stay at the lowest consumer.
+- Keep cohesive forms, menu bodies, and one-off helpers local unless they need their own state, reuse, or semantic boundary.
+- Separate hidden secondary surfaces from the trigger's main flow. For dialogs, dropdowns, popovers, and similar branches, extract a small local component that owns the trigger, open state, and hidden content when it would obscure the parent flow.
+- Preserve composability by separating behavior ownership from layout ownership. A dropdown action may own its trigger, open state, and menu content; the caller owns placement such as slots, offsets, and alignment.
+- Avoid unnecessary DOM hierarchy. Do not add wrapper elements unless they provide layout, semantics, accessibility, state ownership, or integration with a library API; prefer fragments or styling an existing element when possible.
+- Avoid shallow wrappers and prop renaming unless the wrapper adds validation, orchestration, error handling, state ownership, or a real semantic boundary.
+
+## You Might Not Need An Effect
+
+- Use Effects only to synchronize with external systems such as browser APIs, non-React widgets, subscriptions, timers, analytics that must run because the component was shown, or imperative DOM integration.
+- Do not use Effects to transform props or state for rendering. Calculate derived values during render, and use `useMemo` only when the calculation is actually expensive.
+- Do not use Effects to handle user actions. Put action-specific logic in the event handler where the cause is known.
+- Do not use Effects to copy one state value into another state value representing the same concept. Pick one source of truth and derive the rest during render.
+- Do not reset or adjust state from props with an Effect. Prefer a `key` reset, storing a stable ID and deriving the selected object, or guarded same-component render-time adjustment when truly necessary.
+- Prefer framework data APIs or TanStack Query for data fetching instead of writing request Effects in components.
+- If an Effect still seems necessary, first name the external system it synchronizes with. If there is no external system, remove the Effect and restructure the state or event flow.
+
+## Navigation And Performance
+
+- Prefer `Link` for normal navigation. Use router APIs only for command-flow side effects such as mutation success, guarded redirects, or form submission.
+- Avoid `memo`, `useMemo`, and `useCallback` unless there is a clear performance reason.

.agents/skills/tailwind-css-rules/SKILL.md

@@ -0,0 +1,367 @@
+---
+name: tailwind-css-rules
+description: Tailwind CSS v4.1+ rules and best practices. Use when writing, reviewing, refactoring, or upgrading Tailwind CSS classes and styles, especially v4 utility migrations, layout spacing, typography, responsive variants, dark mode, gradients, CSS variables, and component styling.
+---
+
+# Tailwind CSS Rules and Best Practices
+
+## Core Principles
+
+- **Always use Tailwind CSS v4.1+** - Ensure the codebase is using the latest version
+- **Do not use deprecated or removed utilities** - ALWAYS use the replacement
+- **Never use `@apply`** - Use CSS variables, the `--spacing()` function, or framework components instead
+- **Check for redundant classes** - Remove any classes that aren't necessary
+- **Group elements logically** to simplify responsive tweaks later
+
+## Upgrading to Tailwind CSS v4
+
+### Before Upgrading
+
+- **Always read the upgrade documentation first** - Read https://tailwindcss.com/docs/upgrade-guide and https://tailwindcss.com/blog/tailwindcss-v4 before starting an upgrade.
+- Ensure the git repository is in a clean state before starting
+
+### Upgrade Process
+
+1. Run the upgrade command: `npx @tailwindcss/upgrade@latest` for both major and minor updates
+2. The tool will convert JavaScript config files to the new CSS format
+3. Review all changes extensively to clean up any false positives
+4. Test thoroughly across your application
+
+## Breaking Changes Reference
+
+### Removed Utilities (NEVER use these in v4)
+
+| ❌ Deprecated           | ✅ Replacement                                    |
+| ----------------------- | ------------------------------------------------- |
+| `bg-opacity-*`          | Use opacity modifiers like `bg-black/50`          |
+| `text-opacity-*`        | Use opacity modifiers like `text-black/50`        |
+| `border-opacity-*`      | Use opacity modifiers like `border-black/50`      |
+| `divide-opacity-*`      | Use opacity modifiers like `divide-black/50`      |
+| `ring-opacity-*`        | Use opacity modifiers like `ring-black/50`        |
+| `placeholder-opacity-*` | Use opacity modifiers like `placeholder-black/50` |
+| `flex-shrink-*`         | `shrink-*`                                        |
+| `flex-grow-*`           | `grow-*`                                          |
+| `overflow-ellipsis`     | `text-ellipsis`                                   |
+| `decoration-slice`      | `box-decoration-slice`                            |
+| `decoration-clone`      | `box-decoration-clone`                            |
+
+### Renamed Utilities
+
+Use the v4 name when migrating code that still carries Tailwind v3 semantics. Do not blanket-replace existing v4 classes: classes such as `rounded-sm`, `shadow-sm`, `ring-1`, and `ring-2` are valid in this codebase when they intentionally represent the current design scale.
+
+| ❌ v3 pattern       | ✅ v4 pattern                                      |
+| ------------------- | -------------------------------------------------- |
+| `bg-gradient-*`     | `bg-linear-*`                                      |
+| old shadow scale    | verify against the current Tailwind/design scale   |
+| old blur scale      | verify against the current Tailwind/design scale   |
+| old radius scale    | use the Dify radius token mapping when applicable  |
+| `outline-none`      | `outline-hidden`                                   |
+| bare `ring` utility | use an explicit ring width such as `ring-1`/`ring-2`/`ring-3` |
+
+For Figma radius tokens, follow `packages/dify-ui/AGENTS.md`. For example, `--radius/xs` maps to `rounded-sm`; do not rewrite it to `rounded-xs`.
+
+## Layout and Spacing Rules
+
+### Flexbox and Grid Spacing
+
+#### Always use gap utilities for internal spacing
+
+Gap provides consistent spacing without edge cases (no extra space on last items). It's cleaner and more maintainable than margins on children.
+
+```html
+<!-- ❌ Don't do this -->
+<div class="flex">
+  <div class="mr-4">Item 1</div>
+  <div class="mr-4">Item 2</div>
+  <div>Item 3</div>
+  <!-- No margin on last -->
+</div>
+
+<!-- ✅ Do this instead -->
+<div class="flex gap-4">
+  <div>Item 1</div>
+  <div>Item 2</div>
+  <div>Item 3</div>
+</div>
+```
+
+#### Gap vs Space utilities
+
+- **Never use `space-x-*` or `space-y-*` in flex/grid layouts** - always use gap
+- Space utilities add margins to children and have issues with wrapped items
+- Gap works correctly with flex-wrap and all flex directions
+
+```html
+<!-- ❌ Avoid space utilities in flex containers -->
+<div class="flex flex-wrap space-x-4">
+  <!-- Space utilities break with wrapped items -->
+</div>
+
+<!-- ✅ Use gap for consistent spacing -->
+<div class="flex flex-wrap gap-4">
+  <!-- Gap works perfectly with wrapping -->
+</div>
+```
+
+### General Spacing Guidelines
+
+- **Prefer top and left margins** over bottom and right margins (unless conditionally rendered)
+- **Use padding on parent containers** instead of bottom margins on the last child
+- **Always use `min-h-dvh` instead of `min-h-screen`** - `min-h-screen` is buggy on mobile Safari
+- **Prefer `size-*` utilities** over separate `w-*` and `h-*` when setting equal dimensions
+- For max-widths, prefer the container scale (e.g., `max-w-2xs` over `max-w-72`)
+
+## Typography Rules
+
+### Line Heights
+
+- **Never use `leading-*` classes** - Always use line height modifiers with text size
+- **Always use fixed line heights from the spacing scale** - Don't use named values
+
+```html
+<!-- ❌ Don't do this -->
+<p class="text-base leading-7">Text with separate line height</p>
+<p class="text-lg leading-relaxed">Text with named line height</p>
+
+<!-- ✅ Do this instead -->
+<p class="text-base/7">Text with line height modifier</p>
+<p class="text-lg/8">Text with specific line height</p>
+```
+
+### Font Size Reference
+
+Be precise with font sizes - know the actual pixel values:
+
+- `text-xs` = 12px
+- `text-sm` = 14px
+- `text-base` = 16px
+- `text-lg` = 18px
+- `text-xl` = 20px
+
+## Color and Opacity
+
+### Opacity Modifiers
+
+**Never use `bg-opacity-*`, `text-opacity-*`, etc.** - use the opacity modifier syntax:
+
+```html
+<!-- ❌ Don't do this -->
+<div class="bg-red-500 bg-opacity-60">Old opacity syntax</div>
+
+<!-- ✅ Do this instead -->
+<div class="bg-red-500/60">Modern opacity syntax</div>
+```
+
+## Responsive Design
+
+### Breakpoint Optimization
+
+- **Check for redundant classes across breakpoints**
+- **Only add breakpoint variants when values change**
+
+```html
+<!-- ❌ Redundant breakpoint classes -->
+<div class="px-4 md:px-4 lg:px-4">
+  <!-- md:px-4 and lg:px-4 are redundant -->
+</div>
+
+<!-- ✅ Efficient breakpoint usage -->
+<div class="px-4 lg:px-8">
+  <!-- Only specify when value changes -->
+</div>
+```
+
+## Dark Mode
+
+### Dark Mode Best Practices
+
+- Use the plain `dark:` variant pattern
+- Put light mode styles first, then dark mode styles
+- Ensure `dark:` variant comes before other variants
+
+```html
+<!-- ✅ Correct dark mode pattern -->
+<div class="bg-white text-black dark:bg-black dark:text-white">
+  <button class="hover:bg-gray-100 dark:hover:bg-gray-800">Click me</button>
+</div>
+```
+
+## Gradient Utilities
+
+- **ALWAYS Use `bg-linear-*` instead of `bg-gradient-*` utilities** - The gradient utilities were renamed in v4
+- Use the new `bg-radial` or `bg-radial-[<position>]` to create radial gradients
+- Use the new `bg-conic` or `bg-conic-*` to create conic gradients
+
+```html
+<!-- ✅ Use the new gradient utilities -->
+<div class="h-14 bg-linear-to-br from-violet-500 to-fuchsia-500"></div>
+<div
+  class="size-18 bg-radial-[at_50%_75%] from-sky-200 via-blue-400 to-indigo-900 to-90%"
+></div>
+<div
+  class="size-24 bg-conic-180 from-indigo-600 via-indigo-50 to-indigo-600"
+></div>
+
+<!-- ❌ Do not use bg-gradient-* utilities -->
+<div class="h-14 bg-gradient-to-br from-violet-500 to-fuchsia-500"></div>
+```
+
+## Working with CSS Variables
+
+### Accessing Theme Values
+
+Tailwind CSS v4 exposes all theme values as CSS variables:
+
+```css
+/* Access colors, and other theme values */
+.custom-element {
+  background: var(--color-red-500);
+  border-radius: var(--radius-lg);
+}
+```
+
+### The `--spacing()` Function
+
+Use the dedicated `--spacing()` function for spacing calculations:
+
+```css
+.custom-class {
+  margin-top: calc(100vh - --spacing(16));
+}
+```
+
+### Extending theme values
+
+Use CSS to extend theme values:
+
+```css
+@import "tailwindcss";
+
+@theme {
+  --color-mint-500: oklch(0.72 0.11 178);
+}
+```
+
+```html
+<div class="bg-mint-500">
+  <!-- ... -->
+</div>
+```
+
+## New v4 Features
+
+### Container Queries
+
+Use the `@container` class and size variants:
+
+```html
+<article class="@container">
+  <div class="flex flex-col @md:flex-row @lg:gap-8">
+    <img class="w-full @md:w-48" />
+    <div class="mt-4 @md:mt-0">
+      <!-- Content adapts to container size -->
+    </div>
+  </div>
+</article>
+```
+
+### Container Query Units
+
+Use container-based units like `cqw` for responsive sizing:
+
+```html
+<div class="@container">
+  <h1 class="text-[50cqw]">Responsive to container width</h1>
+</div>
+```
+
+### Text Shadows (v4.1)
+
+Use text-shadow-\* utilities from text-shadow-2xs to text-shadow-lg:
+
+```html
+<!-- ✅ Text shadow examples -->
+<h1 class="text-shadow-lg">Large shadow</h1>
+<p class="text-shadow-sm/50">Small shadow with opacity</p>
+```
+
+### Masking (v4.1)
+
+Use the new composable mask utilities for image and gradient masks:
+
+```html
+<!-- ✅ Linear gradient masks on specific sides -->
+<div class="mask-t-from-50%">Top fade</div>
+<div class="mask-b-from-20% mask-b-to-80%">Bottom gradient</div>
+<div class="mask-linear-from-white mask-linear-to-black/60">
+  Fade from white to black
+</div>
+
+<!-- ✅ Radial gradient masks -->
+<div class="mask-radial-[100%_100%] mask-radial-from-75% mask-radial-at-left">
+  Radial mask
+</div>
+```
+
+## Component Patterns
+
+### Avoiding Utility Inheritance
+
+Don't add utilities to parents that you override in children:
+
+```html
+<!-- ❌ Avoid this pattern -->
+<div class="text-center">
+  <h1>Centered Heading</h1>
+  <div class="text-left">Left-aligned content</div>
+</div>
+
+<!-- ✅ Better approach -->
+<div>
+  <h1 class="text-center">Centered Heading</h1>
+  <div>Left-aligned content</div>
+</div>
+```
+
+### Component Extraction
+
+- Extract repeated patterns into framework components, not CSS classes
+- Keep utility classes in templates/JSX
+- Use data attributes for complex state-based styling
+
+## CSS Best Practices
+
+### Nesting Guidelines
+
+- Use nesting when styling both parent and children
+- Avoid empty parent selectors
+
+```css
+/* ✅ Good nesting - parent has styles */
+.card {
+  padding: --spacing(4);
+
+  > .card-title {
+    font-weight: bold;
+  }
+}
+
+/* ❌ Avoid empty parents */
+ul {
+  > li {
+    /* Parent has no styles */
+  }
+}
+```
+
+## Common Pitfalls to Avoid
+
+1. **Using old opacity utilities** - Always use `/opacity` syntax like `bg-red-500/60`
+2. **Redundant breakpoint classes** - Only specify changes
+3. **Space utilities in flex/grid** - Always use gap
+4. **Leading utilities** - Use line-height modifiers like `text-sm/6`
+5. **Arbitrary values** - Use the design scale
+6. **@apply directive** - Use components or CSS variables
+7. **min-h-screen on mobile** - Use min-h-dvh
+8. **Separate width/height** - Use size utilities when equal
+9. **Arbitrary values** - Always use Tailwind's predefined scale whenever possible (e.g., use `ml-4` over `ml-[16px]`)

.github/actions/setup-web/action.yml

@@ -1,8 +1,13 @@
 name: Setup Web Environment
+description: Set up Node.js, Vite+, pnpm, and web dependencies
 
 runs:
   using: composite
   steps:
+    - name: Setup pnpm
+      uses: pnpm/action-setup@8912a9102ac27614460f54aedde9e1e7f9aec20d # v6.0.5
+      with:
+        run_install: false
     - name: Setup Vite+
       uses: voidzero-dev/setup-vp@4f5aa3e38c781f1b01e78fb9255527cee8a6efa6 # v1.8.0
       with:

.github/workflows/autofix.yml

@@ -120,7 +120,11 @@ jobs:
         if: github.event_name != 'merge_group' && steps.api-changes.outputs.any_changed == 'true'
         run: |
           cd api
-          uv run dev/generate_swagger_markdown_docs.py --swagger-dir openapi --markdown-dir openapi/markdown
+          uv run dev/generate_swagger_markdown_docs.py --swagger-dir ../packages/contracts/openapi --markdown-dir openapi/markdown --keep-swagger-json
+
+      - name: Generate frontend contracts
+        if: github.event_name != 'merge_group' && steps.api-changes.outputs.any_changed == 'true'
+        run: pnpm --dir packages/contracts gen-api-contract-from-openapi
 
       - name: ESLint autofix
         if: github.event_name != 'merge_group' && steps.web-changes.outputs.any_changed == 'true'

.github/workflows/labeler.yml

@@ -9,6 +9,6 @@ jobs:
       pull-requests: write
     runs-on: depot-ubuntu-24.04
     steps:
-      - uses: actions/labeler@634933edcd8ababfe52f92936142cc22ac488b1b # v6.0.1
+      - uses: actions/labeler@f27b608878404679385c85cfa523b85ccb86e213 # v6.1.0
         with:
           sync-labels: true

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

@@ -158,7 +158,7 @@ jobs:
 
       - name: Run Claude Code for Translation Sync
         if: steps.context.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@fefa07e9c665b7320f08c3b525980457f22f58aa # v1.0.111
+        uses: anthropics/claude-code-action@476e359e6203e73dad705c8b322e333fabbd7416 # v1.0.119
         with:
           anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
           github_token: ${{ secrets.GITHUB_TOKEN }}

.gitignore

@@ -250,5 +250,5 @@ scripts/stress-test/reports/
 
 # Code Agent Folder
 .qoder/*
-
+.context/*
 .eslintcache

AGENTS.md

@@ -9,6 +9,7 @@ The codebase is split into:
 - **Backend API** (`/api`): Python Flask application organized with Domain-Driven Design
 - **Frontend Web** (`/web`): Next.js application using TypeScript and React
 - **Docker deployment** (`/docker`): Containerized deployment configurations
+- **Dify Agent Backend** (`/dify-agent`): Backend services for managing and executing agent
 
 ## Backend Workflow
 

Makefile

@@ -3,6 +3,10 @@ DOCKER_REGISTRY=langgenius
 WEB_IMAGE=$(DOCKER_REGISTRY)/dify-web
 API_IMAGE=$(DOCKER_REGISTRY)/dify-api
 VERSION=latest
+DOCKER_DIR=docker
+DOCKER_MIDDLEWARE_ENV=$(DOCKER_DIR)/middleware.env
+DOCKER_MIDDLEWARE_ENV_EXAMPLE=$(DOCKER_DIR)/envs/middleware.env.example
+DOCKER_MIDDLEWARE_PROJECT=dify-middlewares-dev
 
 # Default target - show help
 .DEFAULT_GOAL := help
@@ -17,8 +21,13 @@ dev-setup: prepare-docker prepare-web prepare-api
 # Step 1: Prepare Docker middleware
 prepare-docker:
 	@echo "🐳 Setting up Docker middleware..."
-	@cp -n docker/middleware.env.example docker/middleware.env 2>/dev/null || echo "Docker middleware.env already exists"
-	@cd docker && docker compose -f docker-compose.middleware.yaml --env-file middleware.env -p dify-middlewares-dev up -d
+	@if [ ! -f "$(DOCKER_MIDDLEWARE_ENV)" ]; then \
+		cp "$(DOCKER_MIDDLEWARE_ENV_EXAMPLE)" "$(DOCKER_MIDDLEWARE_ENV)"; \
+		echo "Docker middleware.env created"; \
+	else \
+		echo "Docker middleware.env already exists"; \
+	fi
+	@cd $(DOCKER_DIR) && docker compose -f docker-compose.middleware.yaml --env-file middleware.env -p $(DOCKER_MIDDLEWARE_PROJECT) up -d
 	@echo "✅ Docker middleware started"
 
 # Step 2: Prepare web environment
@@ -39,12 +48,18 @@ prepare-api:
 # Clean dev environment
 dev-clean:
 	@echo "⚠️  Stopping Docker containers..."
-	@cd docker && docker compose -f docker-compose.middleware.yaml --env-file middleware.env -p dify-middlewares-dev down
+	@if [ -f "$(DOCKER_MIDDLEWARE_ENV)" ]; then \
+		cd $(DOCKER_DIR) && docker compose -f docker-compose.middleware.yaml --env-file middleware.env -p $(DOCKER_MIDDLEWARE_PROJECT) down; \
+	else \
+		echo "Docker middleware.env does not exist, skipping compose down"; \
+	fi
 	@echo "🗑️  Removing volumes..."
 	@rm -rf docker/volumes/db
+	@rm -rf docker/volumes/mysql
 	@rm -rf docker/volumes/redis
 	@rm -rf docker/volumes/plugin_daemon
 	@rm -rf docker/volumes/weaviate
+	@rm -rf docker/volumes/sandbox/dependencies
 	@rm -rf api/storage
 	@echo "✅ Cleanup complete"
 
@@ -68,16 +83,15 @@ lint:
 	@echo "✅ Linting complete"
 
 type-check:
-	@echo "📝 Running type checks (basedpyright + pyrefly + mypy)..."
-	@./dev/basedpyright-check $(PATH_TO_CHECK)
-	@./dev/pyrefly-check-local
-	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --exclude 'dev/generate_swagger_specs.py' --check-untyped-defs --disable-error-code=import-untyped .
+	@echo "📝 Running type checks (pyrefly + mypy)..."
+	@./dev/pyrefly-check-local $(PATH_TO_CHECK)
+	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --exclude 'dev/generate_swagger_specs.py' --exclude 'dev/generate_fastopenapi_specs.py' --check-untyped-defs --disable-error-code=import-untyped .
 	@echo "✅ Type checks complete"
 
 type-check-core:
-	@echo "📝 Running core type checks (basedpyright + mypy)..."
-	@./dev/basedpyright-check $(PATH_TO_CHECK)
-	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --exclude 'dev/generate_swagger_specs.py'  --exclude 'dev/generate_fastopenapi_specs.py' --check-untyped-defs --disable-error-code=import-untyped .
+	@echo "📝 Running core type checks (pyrefly + mypy)..."
+	@./dev/pyrefly-check-local $(PATH_TO_CHECK)
+	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --exclude 'dev/generate_swagger_specs.py' --exclude 'dev/generate_fastopenapi_specs.py' --check-untyped-defs --disable-error-code=import-untyped .
 	@echo "✅ Core type checks complete"
 
 test:
@@ -132,14 +146,14 @@ help:
 	@echo "  make prepare-docker - Set up Docker middleware"
 	@echo "  make prepare-web    - Set up web environment"
 	@echo "  make prepare-api    - Set up API environment"
-	@echo "  make dev-clean      - Stop Docker middleware containers"
+	@echo "  make dev-clean      - Stop Docker middleware containers and remove dev data"
 	@echo ""
 	@echo "Backend Code Quality:"
 	@echo "  make format         - Format code with ruff"
 	@echo "  make check          - Check code with ruff"
 	@echo "  make lint           - Format, fix, and lint code (ruff, imports, dotenv)"
-	@echo "  make type-check     - Run type checks (basedpyright, pyrefly, mypy)"
-	@echo "  make type-check-core - Run core type checks (basedpyright, mypy)"
+	@echo "  make type-check     - Run type checks (pyrefly, mypy)"
+	@echo "  make type-check-core - Run core type checks (pyrefly, mypy)"
 	@echo "  make test           - Run backend unit tests (or TARGET_TESTS=./api/tests/<target_tests>)"
 	@echo ""
 	@echo "Docker Build Targets:"

api/.env.example

@@ -34,7 +34,7 @@ TRIGGER_URL=http://localhost:5001
 FILES_ACCESS_TIMEOUT=300
 
 # Collaboration mode toggle
-ENABLE_COLLABORATION_MODE=false
+ENABLE_COLLABORATION_MODE=true
 
 # Access token expiration time in minutes
 ACCESS_TOKEN_EXPIRE_MINUTES=60
@@ -88,6 +88,10 @@ REDIS_HEALTH_CHECK_INTERVAL=30
 CELERY_BROKER_URL=redis://:difyai123456@localhost:${REDIS_PORT}/1
 CELERY_BACKEND=redis
 
+# Ops trace retry configuration
+OPS_TRACE_RETRYABLE_DISPATCH_MAX_RETRIES=60
+OPS_TRACE_RETRYABLE_DISPATCH_DELAY_SECONDS=5
+
 # Database configuration
 DB_TYPE=postgresql
 DB_USERNAME=postgres
@@ -553,7 +557,7 @@ MAX_VARIABLE_SIZE=204800
 
 # GraphEngine Worker Pool Configuration
 # Minimum number of workers per GraphEngine instance (default: 1)
-GRAPH_ENGINE_MIN_WORKERS=1
+GRAPH_ENGINE_MIN_WORKERS=3
 # Maximum number of workers per GraphEngine instance (default: 10)
 GRAPH_ENGINE_MAX_WORKERS=10
 # Queue depth threshold that triggers worker scale up (default: 3)

api/AGENTS.md

@@ -195,7 +195,7 @@ Before opening a PR / submitting:
 - Document non-obvious behaviour with concise docstrings and comments.
 - For Flask-RESTX controller request, query, and response schemas, follow `controllers/API_SCHEMA_GUIDE.md`.
   In short: use Pydantic models, document GET query params with `query_params_from_model(...)`, register response
-  DTOs with `register_response_schema_models(...)`, serialize with `ResponseModel.model_validate(...).model_dump(...)`,
+  DTOs with `register_response_schema_models(...)`, serialize response DTOs with `dump_response(...)`,
   and avoid adding new legacy `ns.model(...)`, `@marshal_with(...)`, or GET `@ns.expect(...)` patterns.
 
 ### Miscellaneous

api/Dockerfile

@@ -24,7 +24,8 @@ RUN apt-get update \
 # Install Python dependencies (workspace members under providers/vdb/)
 COPY pyproject.toml uv.lock ./
 COPY providers ./providers
-RUN uv sync --locked --no-dev
+# Trust the checked-in lock during image builds; dev-only path sources live outside the api/ context.
+RUN uv sync --frozen --no-dev
 
 # production stage
 FROM base AS production

api/README.md

@@ -99,7 +99,7 @@ The scripts resolve paths relative to their location, so you can run them from a
    ./dev/reformat               # Run all formatters and linters
    uv run ruff check --fix ./   # Fix linting issues
    uv run ruff format ./        # Format code
-   uv run basedpyright .        # Type checking
+   uv run pyrefly check         # Type checking
    ```
 
 ## Generate TS stub

api/app_factory.py

@@ -117,7 +117,7 @@ def create_flask_app_with_configs() -> DifyApp:
             logger.warning("Failed to add trace headers to response", exc_info=True)
         return response
 
-    # Capture the decorator's return value to avoid pyright reportUnusedFunction
+    # Capture the decorator return values so static checkers do not treat the hooks as unused.
     _ = before_request
     _ = add_trace_headers
 
@@ -181,7 +181,6 @@ def initialize_extensions(app: DifyApp):
         ext_import_modules,
         ext_orjson,
         ext_forward_refs,
-        ext_set_secretkey,
         ext_compress,
         ext_code_based_extension,
         ext_database,
@@ -189,6 +188,7 @@ def initialize_extensions(app: DifyApp):
         ext_migrate,
         ext_redis,
         ext_storage,
+        ext_set_secretkey,
         ext_logstore,  # Initialize logstore after storage, before celery
         ext_celery,
         ext_login,

api/commands/plugin.py

@@ -185,9 +185,9 @@ def transform_datasource_credentials(environment: str):
         firecrawl_plugin_id = "langgenius/firecrawl_datasource"
         jina_plugin_id = "langgenius/jina_datasource"
         if environment == "online":
-            notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)  # pyright: ignore[reportPrivateUsage]
-            firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)  # pyright: ignore[reportPrivateUsage]
-            jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)  # pyright: ignore[reportPrivateUsage]
+            notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)
+            firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)
+            jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)
         else:
             notion_plugin_unique_identifier = None
             firecrawl_plugin_unique_identifier = None

api/commands/system.py

@@ -14,6 +14,7 @@ from libs.rsa import generate_key_pair
 from models import Tenant
 from models.model import App, AppMode, Conversation
 from models.provider import Provider, ProviderModel
+from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider
 
 logger = logging.getLogger(__name__)
 
@@ -23,13 +24,16 @@ DB_UPGRADE_LOCK_TTL_SECONDS = 60
 @click.command(
     "reset-encrypt-key-pair",
     help="Reset the asymmetric key pair of workspace for encrypt LLM credentials. "
-    "After the reset, all LLM credentials will become invalid, "
-    "requiring re-entry."
+    "After the reset, all LLM credentials and tool provider credentials "
+    "(builtin / API / MCP) will be purged, requiring re-entry. "
     "Only support SELF_HOSTED mode.",
 )
 @click.confirmation_option(
     prompt=click.style(
-        "Are you sure you want to reset encrypt key pair? This operation cannot be rolled back!", fg="red"
+        "Are you sure you want to reset encrypt key pair? "
+        "This will also purge builtin / API / MCP tool provider records for every tenant. "
+        "This operation cannot be rolled back!",
+        fg="red",
     )
 )
 def reset_encrypt_key_pair():
@@ -53,6 +57,13 @@ def reset_encrypt_key_pair():
             session.execute(delete(Provider).where(Provider.provider_type == "custom", Provider.tenant_id == tenant.id))
             session.execute(delete(ProviderModel).where(ProviderModel.tenant_id == tenant.id))
 
+            # Purge tool provider records that hold credentials encrypted under the
+            # tenant key. Leaving them in place causes /console/api/workspaces/current/
+            # tool-providers to 500 because decryption fails on stale ciphertext (#35396).
+            session.execute(delete(BuiltinToolProvider).where(BuiltinToolProvider.tenant_id == tenant.id))
+            session.execute(delete(ApiToolProvider).where(ApiToolProvider.tenant_id == tenant.id))
+            session.execute(delete(MCPToolProvider).where(MCPToolProvider.tenant_id == tenant.id))
+
             click.echo(
                 click.style(
                     f"Congratulations! The asymmetric key pair of workspace {tenant.id} has been reset.",

api/configs/enterprise/__init__.py

@@ -23,6 +23,12 @@ class EnterpriseFeatureConfig(BaseSettings):
         ge=1, description="Maximum timeout in seconds for enterprise requests", default=5
     )
 
+    ENTERPRISE_DISABLE_RUNTIME_CREDENTIAL_CHECK: bool = Field(
+        default=False,
+        description="If disabled, credential policy check is only performed when saving workflows."
+        "This helps gain runtime performance by trading off consistency.",
+    )
+
 
 class EnterpriseTelemetryConfig(BaseSettings):
     """

api/configs/feature/__init__.py

@@ -23,9 +23,9 @@ class SecurityConfig(BaseSettings):
     """
 
     SECRET_KEY: str = Field(
-        description="Secret key for secure session cookie signing."
-        "Make sure you are changing this key for your deployment with a strong key."
-        "Generate a strong key using `openssl rand -base64 42` or set via the `SECRET_KEY` environment variable.",
+        description="Secret key for secure session cookie signing. "
+        "Leave empty to let Dify generate a persistent key in the storage directory, "
+        "or set a strong value via the `SECRET_KEY` environment variable.",
         default="",
     )
 
@@ -761,7 +761,7 @@ class WorkflowConfig(BaseSettings):
     # GraphEngine Worker Pool Configuration
     GRAPH_ENGINE_MIN_WORKERS: PositiveInt = Field(
         description="Minimum number of workers per GraphEngine instance",
-        default=1,
+        default=3,
     )
 
     GRAPH_ENGINE_MAX_WORKERS: PositiveInt = Field(
@@ -1137,6 +1137,18 @@ class MultiModalTransferConfig(BaseSettings):
     )
 
 
+class OpsTraceConfig(BaseSettings):
+    OPS_TRACE_RETRYABLE_DISPATCH_MAX_RETRIES: PositiveInt = Field(
+        description="Maximum retry attempts for transient ops trace provider dispatch failures.",
+        default=60,
+    )
+
+    OPS_TRACE_RETRYABLE_DISPATCH_DELAY_SECONDS: PositiveInt = Field(
+        description="Delay in seconds between transient ops trace provider dispatch retry attempts.",
+        default=5,
+    )
+
+
 class CeleryBeatConfig(BaseSettings):
     CELERY_BEAT_SCHEDULER_TIME: int = Field(
         description="Interval in days for Celery Beat scheduler execution, default to 1 day",
@@ -1298,7 +1310,7 @@ class PositionConfig(BaseSettings):
 class CollaborationConfig(BaseSettings):
     ENABLE_COLLABORATION_MODE: bool = Field(
         description="Whether to enable collaboration mode features across the workspace",
-        default=False,
+        default=True,
     )
 
 
@@ -1417,6 +1429,7 @@ class FeatureConfig(
     ModelLoadBalanceConfig,
     ModerationConfig,
     MultiModalTransferConfig,
+    OpsTraceConfig,
     PositionConfig,
     RagEtlConfig,
     RepositoryConfig,

api/configs/middleware/__init__.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Literal, TypedDict
+from typing import Any, Literal, TypedDict, cast
 from urllib.parse import parse_qsl, quote_plus
 
 from pydantic import Field, NonNegativeFloat, NonNegativeInt, PositiveFloat, PositiveInt, computed_field
@@ -50,28 +50,30 @@ from .vdb.vastbase_vector_config import VastbaseVectorConfig
 from .vdb.vikingdb_config import VikingDBConfig
 from .vdb.weaviate_config import WeaviateConfig
 
+_VALID_STORAGE_TYPE = Literal[
+    "opendal",
+    "s3",
+    "aliyun-oss",
+    "azure-blob",
+    "baidu-obs",
+    "clickzetta-volume",
+    "google-storage",
+    "huawei-obs",
+    "oci-storage",
+    "tencent-cos",
+    "volcengine-tos",
+    "supabase",
+    "local",
+]
+
 
 class StorageConfig(BaseSettings):
-    STORAGE_TYPE: Literal[
-        "opendal",
-        "s3",
-        "aliyun-oss",
-        "azure-blob",
-        "baidu-obs",
-        "clickzetta-volume",
-        "google-storage",
-        "huawei-obs",
-        "oci-storage",
-        "tencent-cos",
-        "volcengine-tos",
-        "supabase",
-        "local",
-    ] = Field(
+    STORAGE_TYPE: _VALID_STORAGE_TYPE = Field(
         description="Type of storage to use."
         " Options: 'opendal', '(deprecated) local', 's3', 'aliyun-oss', 'azure-blob', 'baidu-obs', "
         "'clickzetta-volume', 'google-storage', 'huawei-obs', 'oci-storage', 'tencent-cos', "
         "'volcengine-tos', 'supabase'. Default is 'opendal'.",
-        default="opendal",
+        default=cast(_VALID_STORAGE_TYPE, "opendal"),
     )
 
     STORAGE_LOCAL_PATH: str = Field(

api/configs/secret_key.py

@@ -0,0 +1,38 @@
+"""SECRET_KEY persistence helpers for runtime setup."""
+
+from __future__ import annotations
+
+import secrets
+
+from extensions.ext_storage import storage
+
+GENERATED_SECRET_KEY_FILENAME = ".dify_secret_key"
+
+
+def resolve_secret_key(secret_key: str) -> str:
+    """Return an explicit SECRET_KEY or a generated key persisted in storage."""
+    if secret_key:
+        return secret_key
+
+    return _load_or_create_secret_key()
+
+
+def _load_or_create_secret_key() -> str:
+    try:
+        persisted_key = storage.load_once(GENERATED_SECRET_KEY_FILENAME).decode("utf-8").strip()
+        if persisted_key:
+            return persisted_key
+    except FileNotFoundError:
+        pass
+
+    generated_key = secrets.token_urlsafe(48)
+
+    try:
+        storage.save(GENERATED_SECRET_KEY_FILENAME, f"{generated_key}\n".encode())
+    except Exception as exc:
+        raise ValueError(
+            f"SECRET_KEY is not set and could not be generated at {GENERATED_SECRET_KEY_FILENAME}. "
+            "Set SECRET_KEY explicitly or make storage writable."
+        ) from exc
+
+    return generated_key

api/controllers/API_SCHEMA_GUIDE.md

@@ -34,6 +34,7 @@ from controllers.common.schema import (
     register_response_schema_models,
     register_schema_models,
 )
+from libs.helper import dump_response
 ```
 
 Register request payload and query models with `register_schema_models(...)`:
@@ -82,7 +83,7 @@ register_schema_models(console_ns, DraftWorkflowNodeRunPayload)
 def post(self, app_model: App, node_id: str):
     payload = DraftWorkflowNodeRunPayload.model_validate(console_ns.payload or {})
     result = service.run(..., inputs=payload.inputs, query=payload.query)
-    return WorkflowRunNodeExecutionResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
+    return dump_response(WorkflowRunNodeExecutionResponse, result)
 ```
 
 ## Query Parameters
@@ -105,7 +106,7 @@ class WorkflowRunListQuery(BaseModel):
 def get(self, app_model: App):
     query = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))
     result = service.list(..., limit=query.limit, last_id=query.last_id)
-    return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
+    return dump_response(WorkflowRunPaginationResponse, result)
 ```
 
 Do not do this for GET query parameters:
@@ -145,10 +146,25 @@ def post(...):
 Serialize explicitly:
 
 ```python
-return WorkflowRunNodeExecutionResponse.model_validate(
-    workflow_node_execution,
-    from_attributes=True,
-).model_dump(mode="json")
+return dump_response(WorkflowRunNodeExecutionResponse, workflow_node_execution)
+```
+
+`dump_response(...)` is the preferred response serialization helper for a single Pydantic response DTO. It validates
+with `from_attributes=True` and returns `model_dump(mode="json")`, so SQLAlchemy models, plain objects, dictionaries,
+Pydantic aliases, computed fields, and `datetime` values are serialized consistently.
+
+For wrapper responses, pass a dictionary with the public wrapper fields:
+
+```python
+return dump_response(
+    WorkflowRunPaginationResponse,
+    {
+        "data": workflow_runs,
+        "page": page,
+        "limit": limit,
+        "has_more": has_more,
+    },
+)
 ```
 
 If the service can return `None`, translate that into the expected HTTP error before validation:
@@ -158,9 +174,12 @@ workflow_run = service.get_workflow_run(...)
 if workflow_run is None:
     raise NotFound("Workflow run not found")
 
-return WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json")
+return dump_response(WorkflowRunDetailResponse, workflow_run)
 ```
 
+Use manual `model_validate(...).model_dump(...)` only when the endpoint needs behavior that `dump_response(...)` does
+not provide, such as returning a non-dict payload, intentionally excluding fields, or composing a `(body, status)` tuple.
+
 ## Legacy Flask-RESTX Patterns
 
 Avoid adding these patterns to new or migrated endpoints:
@@ -190,4 +209,3 @@ Inspect affected endpoints with `jq`. Check that:
 - Request bodies appear only where the endpoint has a body.
 - Responses reference the expected `*Response` schema.
 - Response schemas use public serialized names, not internal validation aliases like `inputs_dict`.
-

api/controllers/common/human_input.py

@@ -1,6 +1,21 @@
+import json
+
 from pydantic import BaseModel, JsonValue
 
 
 class HumanInputFormSubmitPayload(BaseModel):
     inputs: dict[str, JsonValue]
     action: str
+
+
+def stringify_form_default_values(values: dict[str, object]) -> dict[str, str]:
+    """Serialize default values into strings expected by human-input form clients."""
+    result: dict[str, str] = {}
+    for key, value in values.items():
+        if value is None:
+            result[key] = ""
+        elif isinstance(value, (dict, list)):
+            result[key] = json.dumps(value, ensure_ascii=False)
+        else:
+            result[key] = str(value)
+    return result

api/controllers/common/schema.py

@@ -39,6 +39,7 @@ QueryParamDoc = TypedDict(
 def _register_json_schema(namespace: Namespace, name: str, schema: dict) -> None:
     """Register a JSON schema and promote any nested Pydantic `$defs`."""
 
+    schema = _swagger_2_compatible_schema(schema)
     nested_definitions = schema.get("$defs")
     schema_to_register = dict(schema)
     if isinstance(nested_definitions, dict):
@@ -65,6 +66,35 @@ def _register_schema_model(namespace: Namespace, model: type[BaseModel], *, mode
     )
 
 
+def _swagger_2_compatible_schema(value: Any) -> Any:
+    if isinstance(value, list):
+        return [_swagger_2_compatible_schema(item) for item in value]
+
+    if not isinstance(value, dict):
+        return value
+
+    converted = {key: _swagger_2_compatible_schema(child) for key, child in value.items()}
+    any_of = value.get("anyOf")
+    if not isinstance(any_of, list):
+        return converted
+
+    non_null_candidates = [
+        candidate for candidate in any_of if isinstance(candidate, Mapping) and candidate.get("type") != "null"
+    ]
+    has_null_candidate = any(isinstance(candidate, Mapping) and candidate.get("type") == "null" for candidate in any_of)
+    if not has_null_candidate or len(non_null_candidates) != 1:
+        return converted
+
+    non_null_schema = _swagger_2_compatible_schema(dict(non_null_candidates[0]))
+    if not isinstance(non_null_schema, dict):
+        return converted
+
+    converted.pop("anyOf", None)
+    converted.update(non_null_schema)
+    converted["x-nullable"] = True
+    return converted
+
+
 def register_schema_model(namespace: Namespace, model: type[BaseModel]) -> None:
     """Register a BaseModel and its nested schema definitions for Swagger documentation."""
 

api/controllers/console/__init__.py

@@ -33,7 +33,6 @@ for module_name in RESOURCE_MODULES:
 # Ensure resource modules are imported so route decorators are evaluated.
 # Import other controllers
 from . import (
-    admin,
     apikey,
     extension,
     feature,
@@ -117,7 +116,7 @@ from .explore import (
     saved_message,
     trial,
 )
-from .socketio import workflow as socketio_workflow  # pyright: ignore[reportUnusedImport]
+from .socketio import workflow as socketio_workflow
 
 # Import tag controllers
 from .tag import tags
@@ -142,7 +141,6 @@ api.add_namespace(console_ns)
 __all__ = [
     "account",
     "activate",
-    "admin",
     "advanced_prompt_template",
     "agent",
     "agent_providers",

api/controllers/console/admin.py

@@ -1,64 +1,11 @@
-import csv
-import io
 from collections.abc import Callable
 from functools import wraps
-from typing import cast
-from uuid import UUID
 
 from flask import request
-from flask_restx import Resource
-from pydantic import BaseModel, Field, field_validator
-from sqlalchemy import select
-from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
+from werkzeug.exceptions import Unauthorized
 
 from configs import dify_config
-from constants.languages import supported_language
-from controllers.common.schema import register_schema_models
-from controllers.console import console_ns
-from controllers.console.wraps import only_edition_cloud
-from core.db.session_factory import session_factory
-from extensions.ext_database import db
 from libs.token import extract_access_token
-from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
-from services.billing_service import BillingService, LangContentDict
-
-
-class InsertExploreAppPayload(BaseModel):
-    app_id: str = Field(...)
-    desc: str | None = None
-    copyright: str | None = None
-    privacy_policy: str | None = None
-    custom_disclaimer: str | None = None
-    language: str = Field(...)
-    category: str = Field(...)
-    position: int = Field(...)
-    can_trial: bool = Field(default=False)
-    trial_limit: int = Field(default=0)
-
-    @field_validator("language")
-    @classmethod
-    def validate_language(cls, value: str) -> str:
-        return supported_language(value)
-
-
-class InsertExploreBannerPayload(BaseModel):
-    category: str = Field(...)
-    title: str = Field(...)
-    description: str = Field(...)
-    img_src: str = Field(..., alias="img-src")
-    language: str = Field(default="en-US")
-    link: str = Field(...)
-    sort: int = Field(...)
-
-    @field_validator("language")
-    @classmethod
-    def validate_language(cls, value: str) -> str:
-        return supported_language(value)
-
-    model_config = {"populate_by_name": True}
-
-
-register_schema_models(console_ns, InsertExploreAppPayload, InsertExploreBannerPayload)
 
 
 def admin_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
@@ -76,353 +23,3 @@ def admin_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
         return view(*args, **kwargs)
 
     return decorated
-
-
-@console_ns.route("/admin/insert-explore-apps")
-class InsertExploreAppListApi(Resource):
-    @console_ns.doc("insert_explore_app")
-    @console_ns.doc(description="Insert or update an app in the explore list")
-    @console_ns.expect(console_ns.models[InsertExploreAppPayload.__name__])
-    @console_ns.response(200, "App updated successfully")
-    @console_ns.response(201, "App inserted successfully")
-    @console_ns.response(404, "App not found")
-    @only_edition_cloud
-    @admin_required
-    def post(self):
-        payload = InsertExploreAppPayload.model_validate(console_ns.payload)
-
-        app = db.session.execute(select(App).where(App.id == payload.app_id)).scalar_one_or_none()
-        if not app:
-            raise NotFound(f"App '{payload.app_id}' is not found")
-
-        site = app.site
-        if not site:
-            desc = payload.desc or ""
-            copy_right = payload.copyright or ""
-            privacy_policy = payload.privacy_policy or ""
-            custom_disclaimer = payload.custom_disclaimer or ""
-        else:
-            desc = site.description or payload.desc or ""
-            copy_right = site.copyright or payload.copyright or ""
-            privacy_policy = site.privacy_policy or payload.privacy_policy or ""
-            custom_disclaimer = site.custom_disclaimer or payload.custom_disclaimer or ""
-
-        with session_factory.create_session() as session:
-            recommended_app = session.execute(
-                select(RecommendedApp).where(RecommendedApp.app_id == payload.app_id)
-            ).scalar_one_or_none()
-
-            if not recommended_app:
-                recommended_app = RecommendedApp(
-                    app_id=app.id,
-                    description=desc,
-                    copyright=copy_right,
-                    privacy_policy=privacy_policy,
-                    custom_disclaimer=custom_disclaimer,
-                    language=payload.language,
-                    category=payload.category,
-                    position=payload.position,
-                )
-
-                db.session.add(recommended_app)
-                if payload.can_trial:
-                    trial_app = db.session.execute(
-                        select(TrialApp).where(TrialApp.app_id == payload.app_id)
-                    ).scalar_one_or_none()
-                    if not trial_app:
-                        db.session.add(
-                            TrialApp(
-                                app_id=payload.app_id,
-                                tenant_id=app.tenant_id,
-                                trial_limit=payload.trial_limit,
-                            )
-                        )
-                    else:
-                        trial_app.trial_limit = payload.trial_limit
-
-                app.is_public = True
-                db.session.commit()
-
-                return {"result": "success"}, 201
-            else:
-                recommended_app.description = desc
-                recommended_app.copyright = copy_right
-                recommended_app.privacy_policy = privacy_policy
-                recommended_app.custom_disclaimer = custom_disclaimer
-                recommended_app.language = payload.language
-                recommended_app.category = payload.category
-                recommended_app.position = payload.position
-
-                if payload.can_trial:
-                    trial_app = db.session.execute(
-                        select(TrialApp).where(TrialApp.app_id == payload.app_id)
-                    ).scalar_one_or_none()
-                    if not trial_app:
-                        db.session.add(
-                            TrialApp(
-                                app_id=payload.app_id,
-                                tenant_id=app.tenant_id,
-                                trial_limit=payload.trial_limit,
-                            )
-                        )
-                    else:
-                        trial_app.trial_limit = payload.trial_limit
-                app.is_public = True
-
-                db.session.commit()
-
-                return {"result": "success"}, 200
-
-
-@console_ns.route("/admin/insert-explore-apps/<uuid:app_id>")
-class InsertExploreAppApi(Resource):
-    @console_ns.doc("delete_explore_app")
-    @console_ns.doc(description="Remove an app from the explore list")
-    @console_ns.doc(params={"app_id": "Application ID to remove"})
-    @console_ns.response(204, "App removed successfully")
-    @only_edition_cloud
-    @admin_required
-    def delete(self, app_id: UUID):
-        with session_factory.create_session() as session:
-            recommended_app = session.execute(
-                select(RecommendedApp).where(RecommendedApp.app_id == str(app_id))
-            ).scalar_one_or_none()
-
-        if not recommended_app:
-            return {"result": "success"}, 204
-
-        with session_factory.create_session() as session:
-            app = session.execute(select(App).where(App.id == recommended_app.app_id)).scalar_one_or_none()
-
-        if app:
-            app.is_public = False
-
-        with session_factory.create_session() as session:
-            installed_apps = (
-                session.execute(
-                    select(InstalledApp).where(
-                        InstalledApp.app_id == recommended_app.app_id,
-                        InstalledApp.tenant_id != InstalledApp.app_owner_tenant_id,
-                    )
-                )
-                .scalars()
-                .all()
-            )
-
-            for installed_app in installed_apps:
-                session.delete(installed_app)
-
-            trial_app = session.execute(
-                select(TrialApp).where(TrialApp.app_id == recommended_app.app_id)
-            ).scalar_one_or_none()
-            if trial_app:
-                session.delete(trial_app)
-
-        db.session.delete(recommended_app)
-        db.session.commit()
-
-        return {"result": "success"}, 204
-
-
-@console_ns.route("/admin/insert-explore-banner")
-class InsertExploreBannerApi(Resource):
-    @console_ns.doc("insert_explore_banner")
-    @console_ns.doc(description="Insert an explore banner")
-    @console_ns.expect(console_ns.models[InsertExploreBannerPayload.__name__])
-    @console_ns.response(201, "Banner inserted successfully")
-    @only_edition_cloud
-    @admin_required
-    def post(self):
-        payload = InsertExploreBannerPayload.model_validate(console_ns.payload)
-
-        banner = ExporleBanner(
-            content={
-                "category": payload.category,
-                "title": payload.title,
-                "description": payload.description,
-                "img-src": payload.img_src,
-            },
-            link=payload.link,
-            sort=payload.sort,
-            language=payload.language,
-        )
-        db.session.add(banner)
-        db.session.commit()
-
-        return {"result": "success"}, 201
-
-
-@console_ns.route("/admin/delete-explore-banner/<uuid:banner_id>")
-class DeleteExploreBannerApi(Resource):
-    @console_ns.doc("delete_explore_banner")
-    @console_ns.doc(description="Delete an explore banner")
-    @console_ns.doc(params={"banner_id": "Banner ID to delete"})
-    @console_ns.response(204, "Banner deleted successfully")
-    @only_edition_cloud
-    @admin_required
-    def delete(self, banner_id):
-        banner = db.session.execute(select(ExporleBanner).where(ExporleBanner.id == banner_id)).scalar_one_or_none()
-        if not banner:
-            raise NotFound(f"Banner '{banner_id}' is not found")
-
-        db.session.delete(banner)
-        db.session.commit()
-
-        return {"result": "success"}, 204
-
-
-class LangContentPayload(BaseModel):
-    lang: str = Field(..., description="Language tag: 'zh' | 'en' | 'jp'")
-    title: str = Field(...)
-    subtitle: str | None = Field(default=None)
-    body: str = Field(...)
-    title_pic_url: str | None = Field(default=None)
-
-
-class UpsertNotificationPayload(BaseModel):
-    notification_id: str | None = Field(default=None, description="Omit to create; supply UUID to update")
-    contents: list[LangContentPayload] = Field(..., min_length=1)
-    start_time: str | None = Field(default=None, description="RFC3339, e.g. 2026-03-01T00:00:00Z")
-    end_time: str | None = Field(default=None, description="RFC3339, e.g. 2026-03-20T23:59:59Z")
-    frequency: str = Field(default="once", description="'once' | 'every_page_load'")
-    status: str = Field(default="active", description="'active' | 'inactive'")
-
-
-class BatchAddNotificationAccountsPayload(BaseModel):
-    notification_id: str = Field(...)
-    user_email: list[str] = Field(..., description="List of account email addresses")
-
-
-register_schema_models(console_ns, UpsertNotificationPayload, BatchAddNotificationAccountsPayload)
-
-
-@console_ns.route("/admin/upsert_notification")
-class UpsertNotificationApi(Resource):
-    @console_ns.doc("upsert_notification")
-    @console_ns.doc(
-        description=(
-            "Create or update an in-product notification. "
-            "Supply notification_id to update an existing one; omit it to create a new one. "
-            "Pass at least one language variant in contents (zh / en / jp)."
-        )
-    )
-    @console_ns.expect(console_ns.models[UpsertNotificationPayload.__name__])
-    @console_ns.response(200, "Notification upserted successfully")
-    @only_edition_cloud
-    @admin_required
-    def post(self):
-        payload = UpsertNotificationPayload.model_validate(console_ns.payload)
-        result = BillingService.upsert_notification(
-            contents=[cast(LangContentDict, c.model_dump()) for c in payload.contents],
-            frequency=payload.frequency,
-            status=payload.status,
-            notification_id=payload.notification_id,
-            start_time=payload.start_time,
-            end_time=payload.end_time,
-        )
-        return {"result": "success", "notification_id": result.get("notificationId")}, 200
-
-
-@console_ns.route("/admin/batch_add_notification_accounts")
-class BatchAddNotificationAccountsApi(Resource):
-    @console_ns.doc("batch_add_notification_accounts")
-    @console_ns.doc(
-        description=(
-            "Register target accounts for a notification by email address. "
-            'JSON body: {"notification_id": "...", "user_email": ["a@example.com", ...]}. '
-            "File upload: multipart/form-data with a 'file' field (CSV or TXT, one email per line) "
-            "plus a 'notification_id' field. "
-            "Emails that do not match any account are silently skipped."
-        )
-    )
-    @console_ns.response(200, "Accounts added successfully")
-    @only_edition_cloud
-    @admin_required
-    def post(self):
-        from models.account import Account
-
-        if "file" in request.files:
-            notification_id = request.form.get("notification_id", "").strip()
-            if not notification_id:
-                raise BadRequest("notification_id is required.")
-            emails = self._parse_emails_from_file()
-        else:
-            payload = BatchAddNotificationAccountsPayload.model_validate(console_ns.payload)
-            notification_id = payload.notification_id
-            emails = payload.user_email
-
-        if not emails:
-            raise BadRequest("No valid email addresses provided.")
-
-        # Resolve emails → account IDs in chunks to avoid large IN-clause
-        account_ids: list[str] = []
-        chunk_size = 500
-        for i in range(0, len(emails), chunk_size):
-            chunk = emails[i : i + chunk_size]
-            rows = db.session.execute(select(Account.id, Account.email).where(Account.email.in_(chunk))).all()
-            account_ids.extend(str(row.id) for row in rows)
-
-        if not account_ids:
-            raise BadRequest("None of the provided emails matched an existing account.")
-
-        # Send to dify-saas in batches of 1000
-        total_count = 0
-        batch_size = 1000
-        for i in range(0, len(account_ids), batch_size):
-            batch = account_ids[i : i + batch_size]
-            result = BillingService.batch_add_notification_accounts(
-                notification_id=notification_id,
-                account_ids=batch,
-            )
-            total_count += result.get("count", 0)
-
-        return {
-            "result": "success",
-            "emails_provided": len(emails),
-            "accounts_matched": len(account_ids),
-            "count": total_count,
-        }, 200
-
-    @staticmethod
-    def _parse_emails_from_file() -> list[str]:
-        """Parse email addresses from an uploaded CSV or TXT file."""
-        file = request.files["file"]
-        if not file.filename:
-            raise BadRequest("Uploaded file has no filename.")
-
-        filename_lower = file.filename.lower()
-        if not filename_lower.endswith((".csv", ".txt")):
-            raise BadRequest("Invalid file type. Only CSV (.csv) and TXT (.txt) files are allowed.")
-
-        try:
-            content = file.stream.read().decode("utf-8")
-        except UnicodeDecodeError:
-            try:
-                file.stream.seek(0)
-                content = file.stream.read().decode("gbk")
-            except UnicodeDecodeError:
-                raise BadRequest("Unable to decode the file. Please use UTF-8 or GBK encoding.")
-
-        emails: list[str] = []
-        if filename_lower.endswith(".csv"):
-            reader = csv.reader(io.StringIO(content))
-            for row in reader:
-                for cell in row:
-                    cell = cell.strip()
-                    if cell:
-                        emails.append(cell)
-        else:
-            for line in content.splitlines():
-                line = line.strip()
-                if line:
-                    emails.append(line)
-
-        # Deduplicate while preserving order
-        seen: set[str] = set()
-        unique_emails: list[str] = []
-        for email in emails:
-            if email.lower() not in seen:
-                seen.add(email.lower())
-                unique_emails.append(email)
-
-        return unique_emails

api/controllers/console/apikey.py

@@ -11,6 +11,7 @@ from werkzeug.exceptions import Forbidden
 from controllers.common.schema import register_schema_models
 from extensions.ext_database import db
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import Dataset
 from models.enums import ApiTokenType
@@ -21,12 +22,6 @@ from . import console_ns
 from .wraps import account_initialization_required, edit_permission_required, setup_required
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class ApiKeyItem(ResponseModel):
     id: str
     type: str
@@ -37,7 +32,7 @@ class ApiKeyItem(ResponseModel):
     @field_validator("last_used_at", "created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class ApiKeyList(ResponseModel):

api/controllers/console/app/app.py

@@ -3,7 +3,6 @@ import re
 import uuid
 from datetime import datetime
 from typing import Any, Literal
-from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
@@ -34,12 +33,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
+from libs.helper import build_icon_url, to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
 from models.model import IconType
 from services.app_dsl_service import AppDslService
-from services.app_service import AppService
+from services.app_service import AppListParams, AppService, CreateAppParams
 from services.enterprise.enterprise_service import EnterpriseService
 from services.entities.dsl_entities import ImportMode, ImportStatus
 from services.entities.knowledge_entities.knowledge_entities import (
@@ -178,12 +177,6 @@ class AppTracePayload(BaseModel):
 type JSONValue = Any
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class Tag(ResponseModel):
     id: str
     name: str
@@ -200,7 +193,7 @@ class WorkflowPartial(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class ModelConfigPartial(ResponseModel):
@@ -214,7 +207,7 @@ class ModelConfigPartial(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class ModelConfig(ResponseModel):
@@ -275,7 +268,7 @@ class ModelConfig(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class Site(ResponseModel):
@@ -318,7 +311,7 @@ class Site(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class DeletedTool(ResponseModel):
@@ -361,7 +354,7 @@ class AppPartial(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AppDetail(ResponseModel):
@@ -391,7 +384,7 @@ class AppDetail(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AppDetailWithSite(AppDetail):
@@ -478,11 +471,18 @@ class AppListApi(Resource):
         current_user, current_tenant_id = current_account_with_tenant()
 
         args = AppListQuery.model_validate(_normalize_app_list_query_args(request.args))
-        args_dict = args.model_dump()
+        params = AppListParams(
+            page=args.page,
+            limit=args.limit,
+            mode=args.mode,
+            name=args.name,
+            tag_ids=args.tag_ids,
+            is_created_by_me=args.is_created_by_me,
+        )
 
         # get app list
         app_service = AppService()
-        app_pagination = app_service.get_paginate_apps(current_user.id, current_tenant_id, args_dict)
+        app_pagination = app_service.get_paginate_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
@@ -546,9 +546,17 @@ class AppListApi(Resource):
         """Create app"""
         current_user, current_tenant_id = current_account_with_tenant()
         args = CreateAppPayload.model_validate(console_ns.payload)
+        params = CreateAppParams(
+            name=args.name,
+            description=args.description,
+            mode=args.mode,
+            icon_type=args.icon_type,
+            icon=args.icon,
+            icon_background=args.icon_background,
+        )
 
         app_service = AppService()
-        app = app_service.create_app(current_tenant_id, args.model_dump(), current_user)
+        app = app_service.create_app(current_tenant_id, params, current_user)
         app_detail = AppDetail.model_validate(app, from_attributes=True)
         return app_detail.model_dump(mode="json"), 201
 
@@ -841,10 +849,11 @@ class AppTraceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def get(self, app_id: UUID):
+    @get_app_model
+    def get(self, app_model):
         """Get app trace"""
         with session_factory.create_session() as session:
-            app_trace_config = OpsTraceManager.get_app_tracing_config(str(app_id), session)
+            app_trace_config = OpsTraceManager.get_app_tracing_config(app_model.id, session)
 
         return app_trace_config
 
@@ -858,12 +867,13 @@ class AppTraceApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    def post(self, app_id: UUID):
+    @get_app_model
+    def post(self, app_model):
         # add app trace
         args = AppTracePayload.model_validate(console_ns.payload)
 
         OpsTraceManager.update_app_tracing_config(
-            app_id=str(app_id),
+            app_id=app_model.id,
             enabled=args.enabled,
             tracing_provider=args.tracing_provider,
         )

api/controllers/console/app/conversation_variables.py

@@ -16,6 +16,7 @@ from controllers.console.wraps import account_initialization_required, setup_req
 from extensions.ext_database import db
 from fields._value_type_serializer import serialize_value_type
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import login_required
 from models import ConversationVariable
 from models.model import AppMode
@@ -25,12 +26,6 @@ class ConversationVariablesQuery(BaseModel):
     conversation_id: str = Field(..., description="Conversation ID to filter variables")
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class ConversationVariableResponse(ResponseModel):
     id: str
     name: str
@@ -65,7 +60,7 @@ class ConversationVariableResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class PaginatedConversationVariableResponse(ResponseModel):

api/controllers/console/app/mcp_server.py

@@ -13,6 +13,7 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from extensions.ext_database import db
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.enums import AppMCPServerStatus
 from models.model import AppMCPServer
@@ -30,12 +31,6 @@ class MCPServerUpdatePayload(BaseModel):
     status: str | None = Field(default=None, description="Server status")
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class AppMCPServerResponse(ResponseModel):
     id: str
     name: str
@@ -59,7 +54,7 @@ class AppMCPServerResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 register_schema_models(console_ns, MCPServerCreatePayload, MCPServerUpdatePayload, AppMCPServerResponse)

api/controllers/console/app/message.py

@@ -37,10 +37,9 @@ from fields.conversation_fields import (
     JSONValue,
     MessageFile,
     format_files_contained,
-    to_timestamp,
 )
 from graphon.model_runtime.errors.invoke import InvokeError
-from libs.helper import uuid_value
+from libs.helper import to_timestamp, uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import current_account_with_tenant, login_required
 from models.enums import FeedbackFromSource, FeedbackRating
@@ -144,9 +143,7 @@ class MessageDetailResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class MessageInfiniteScrollPaginationResponse(ResponseModel):

api/controllers/console/app/ops_trace.py

@@ -1,5 +1,4 @@
 from typing import Any
-from uuid import UUID
 
 from flask import request
 from flask_restx import Resource, fields
@@ -9,8 +8,10 @@ from werkzeug.exceptions import BadRequest
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import TracingConfigCheckError, TracingConfigIsExist, TracingConfigNotExist
+from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from libs.login import login_required
+from models import App
 from services.ops_service import OpsService
 
 
@@ -43,11 +44,14 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def get(self, app_id: UUID):
-        args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))
+    @get_app_model
+    def get(self, app_model: App):
+        args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
 
         try:
-            trace_config = OpsService.get_tracing_app_config(app_id=str(app_id), tracing_provider=args.tracing_provider)
+            trace_config = OpsService.get_tracing_app_config(
+                app_id=app_model.id, tracing_provider=args.tracing_provider
+            )
             if not trace_config:
                 return {"has_not_configured": True}
             return trace_config
@@ -65,13 +69,14 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def post(self, app_id: UUID):
+    @get_app_model
+    def post(self, app_model: App):
         """Create a new trace app configuration"""
         args = TraceConfigPayload.model_validate(console_ns.payload)
 
         try:
             result = OpsService.create_tracing_app_config(
-                app_id=str(app_id), tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
+                app_id=app_model.id, tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
             )
             if not result:
                 raise TracingConfigIsExist()
@@ -90,13 +95,14 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def patch(self, app_id: UUID):
+    @get_app_model
+    def patch(self, app_model: App):
         """Update an existing trace app configuration"""
         args = TraceConfigPayload.model_validate(console_ns.payload)
 
         try:
             result = OpsService.update_tracing_app_config(
-                app_id=str(app_id), tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
+                app_id=app_model.id, tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
             )
             if not result:
                 raise TracingConfigNotExist()
@@ -113,12 +119,13 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def delete(self, app_id: UUID):
+    @get_app_model
+    def delete(self, app_model: App):
         """Delete an existing trace app configuration"""
         args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))
 
         try:
-            result = OpsService.delete_tracing_app_config(app_id=str(app_id), tracing_provider=args.tracing_provider)
+            result = OpsService.delete_tracing_app_config(app_id=app_model.id, tracing_provider=args.tracing_provider)
             if not result:
                 raise TracingConfigNotExist()
             return {"result": "success"}, 204

api/controllers/console/app/workflow.py

@@ -1,17 +1,22 @@
 import json
 import logging
 from collections.abc import Sequence
-from typing import Any
+from datetime import datetime
+from typing import Any, NotRequired, TypedDict
 
 from flask import abort, request
-from flask_restx import Resource, fields, marshal, marshal_with
-from pydantic import BaseModel, Field, ValidationError, field_validator
+from flask_restx import Resource, fields
+from pydantic import AliasChoices, BaseModel, Field, ValidationError, field_validator
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
 
 import services
 from controllers.common.controller_schemas import DefaultBlockConfigQuery, WorkflowListQuery, WorkflowUpdatePayload
-from controllers.common.schema import register_response_schema_model, register_schema_models
+from controllers.common.schema import (
+    register_response_schema_model,
+    register_response_schema_models,
+    register_schema_models,
+)
 from controllers.console import console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.wraps import get_app_model
@@ -22,6 +27,7 @@ from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.file_access import DatabaseFileAccessController
+from core.helper import encrypter
 from core.helper.trace_id_helper import get_external_trace_id
 from core.plugin.impl.exc import PluginInvokeError
 from core.trigger.constants import TRIGGER_SCHEDULE_NODE_TYPE
@@ -34,18 +40,18 @@ from core.trigger.debug.event_selectors import (
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from factories import file_factory, variable_factory
-from fields.member_fields import simple_account_fields
-from fields.online_user_fields import online_user_list_fields
-from fields.workflow_fields import workflow_fields, workflow_pagination_fields
+from fields.base import ResponseModel
+from fields.member_fields import SimpleAccount
 from fields.workflow_run_fields import WorkflowRunNodeExecutionResponse
 from graphon.enums import NodeType
 from graphon.file import File
 from graphon.file import helpers as file_helpers
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.utils.encoders import jsonable_encoder
+from graphon.variables import SecretVariable, SegmentType, VariableBase
 from libs import helper
 from libs.datetime_utils import naive_utc_now
-from libs.helper import TimestampField, uuid_value
+from libs.helper import TimestampField, dump_response, to_timestamp, uuid_value
 from libs.login import current_account_with_tenant, login_required
 from models import App
 from models.model import AppMode
@@ -64,42 +70,15 @@ LISTENING_RETRY_IN = 2000
 RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE = "source workflow must be published"
 MAX_WORKFLOW_ONLINE_USERS_REQUEST_IDS = 1000
 WORKFLOW_ONLINE_USERS_REDIS_BATCH_SIZE = 50
+ENVIRONMENT_VARIABLE_SUPPORTED_TYPES = (SegmentType.STRING, SegmentType.NUMBER, SegmentType.SECRET)
 
-# Register models for flask_restx to avoid dict type issues in Swagger
-# Register in dependency order: base models first, then dependent models
 
-# Base models
-simple_account_model = console_ns.model("SimpleAccount", simple_account_fields)
-
-from fields.workflow_fields import pipeline_variable_fields, serialize_value_type
-
-conversation_variable_model = console_ns.model(
-    "ConversationVariable",
-    {
-        "id": fields.String,
-        "name": fields.String,
-        "value_type": fields.String(attribute=serialize_value_type),
-        "value": fields.Raw,
-        "description": fields.String,
-    },
-)
-
-pipeline_variable_model = console_ns.model("PipelineVariable", pipeline_variable_fields)
-
-# Workflow model with nested dependencies
-workflow_fields_copy = workflow_fields.copy()
-workflow_fields_copy["created_by"] = fields.Nested(simple_account_model, attribute="created_by_account")
-workflow_fields_copy["updated_by"] = fields.Nested(
-    simple_account_model, attribute="updated_by_account", allow_null=True
-)
-workflow_fields_copy["conversation_variables"] = fields.List(fields.Nested(conversation_variable_model))
-workflow_fields_copy["rag_pipeline_variables"] = fields.List(fields.Nested(pipeline_variable_model))
-workflow_model = console_ns.model("Workflow", workflow_fields_copy)
-
-# Workflow pagination model
-workflow_pagination_fields_copy = workflow_pagination_fields.copy()
-workflow_pagination_fields_copy["items"] = fields.List(fields.Nested(workflow_model), attribute="items")
-workflow_pagination_model = console_ns.model("WorkflowPagination", workflow_pagination_fields_copy)
+class EnvironmentVariableResponseDict(TypedDict):
+    value_type: str
+    id: NotRequired[str]
+    name: NotRequired[str]
+    value: NotRequired[Any]
+    description: NotRequired[str | None]
 
 
 class SyncDraftWorkflowPayload(BaseModel):
@@ -170,6 +149,110 @@ class WorkflowOnlineUsersPayload(BaseModel):
         return list(dict.fromkeys(app_id.strip() for app_id in app_ids if app_id.strip()))
 
 
+class WorkflowConversationVariableResponse(ResponseModel):
+    id: str
+    name: str
+    value_type: str
+    value: Any = Field(json_schema_extra={"type": "object"})
+    description: str
+
+    @field_validator("value_type", mode="before")
+    @classmethod
+    def _serialize_value_type(cls, value: Any) -> str:
+        if hasattr(value, "exposed_type"):
+            return str(value.exposed_type())
+        return str(value)
+
+
+class PipelineVariableResponse(ResponseModel):
+    label: str
+    variable: str
+    type: str
+    belong_to_node_id: str
+    max_length: int | None = None
+    required: bool
+    unit: str | None = None
+    default_value: Any = Field(default=None, json_schema_extra={"type": "object"})
+    options: list[str] | None = None
+    placeholder: str | None = None
+    tooltips: str | None = None
+    allowed_file_types: list[str] | None = None
+    allowed_file_extensions: list[str] | None = Field(
+        default=None, validation_alias=AliasChoices("allowed_file_extensions", "allow_file_extension")
+    )
+    allowed_file_upload_methods: list[str] | None = Field(
+        default=None, validation_alias=AliasChoices("allowed_file_upload_methods", "allow_file_upload_methods")
+    )
+
+
+class WorkflowEnvironmentVariableResponse(ResponseModel):
+    value_type: str
+    id: str
+    name: str
+    value: Any = Field(json_schema_extra={"type": "object"})
+    description: str
+
+
+class WorkflowResponse(ResponseModel):
+    id: str
+    graph: dict[str, Any] = Field(validation_alias=AliasChoices("graph_dict", "graph"))
+    features: dict[str, Any] = Field(validation_alias=AliasChoices("features_dict", "features"))
+    hash: str = Field(validation_alias=AliasChoices("unique_hash", "hash"))
+    version: str
+    marked_name: str
+    marked_comment: str
+    created_by: SimpleAccount | None = Field(
+        default=None, validation_alias=AliasChoices("created_by_account", "created_by")
+    )
+    created_at: int
+    updated_by: SimpleAccount | None = Field(
+        default=None, validation_alias=AliasChoices("updated_by_account", "updated_by")
+    )
+    updated_at: int
+    tool_published: bool
+    environment_variables: list[WorkflowEnvironmentVariableResponse]
+    conversation_variables: list[WorkflowConversationVariableResponse]
+    rag_pipeline_variables: list[PipelineVariableResponse]
+
+    @field_validator("created_at", "updated_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int:
+        timestamp = to_timestamp(value)
+        if timestamp is None:
+            raise ValueError("timestamp is required")
+        return timestamp
+
+    @field_validator("environment_variables", mode="before")
+    @classmethod
+    def _serialize_environment_variables(cls, value: Any) -> list[Any]:
+        if value is None:
+            return []
+
+        return [_serialize_environment_variable(item) for item in value]
+
+
+class WorkflowPaginationResponse(ResponseModel):
+    items: list[WorkflowResponse]
+    page: int
+    limit: int
+    has_more: bool
+
+
+class WorkflowOnlineUser(ResponseModel):
+    user_id: str
+    username: str
+    avatar: str | None = None
+
+
+class WorkflowOnlineUsersByApp(ResponseModel):
+    app_id: str
+    users: list[WorkflowOnlineUser]
+
+
+class WorkflowOnlineUsersResponse(ResponseModel):
+    data: list[WorkflowOnlineUsersByApp]
+
+
 class DraftWorkflowTriggerRunPayload(BaseModel):
     node_id: str
 
@@ -197,6 +280,17 @@ register_schema_models(
     DraftWorkflowTriggerRunAllPayload,
 )
 register_response_schema_model(console_ns, WorkflowRunNodeExecutionResponse)
+register_response_schema_models(
+    console_ns,
+    WorkflowConversationVariableResponse,
+    PipelineVariableResponse,
+    WorkflowEnvironmentVariableResponse,
+    WorkflowResponse,
+    WorkflowPaginationResponse,
+    WorkflowOnlineUser,
+    WorkflowOnlineUsersByApp,
+    WorkflowOnlineUsersResponse,
+)
 
 
 # TODO(QuantumGhost): Refactor existing node run API to handle file parameter parsing
@@ -218,18 +312,56 @@ def _parse_file(workflow: Workflow, files: list[dict] | None = None) -> Sequence
     return file_objs
 
 
+def _serialize_environment_variable(value: Any) -> EnvironmentVariableResponseDict | Any:
+    match value:
+        case SecretVariable():
+            return {
+                "id": value.id,
+                "name": value.name,
+                "value": encrypter.full_mask_token(),
+                "value_type": value.value_type.value,
+                "description": value.description,
+            }
+
+        case VariableBase():
+            return {
+                "id": value.id,
+                "name": value.name,
+                "value": value.value,
+                "value_type": str(value.value_type.exposed_type()),
+                "description": value.description,
+            }
+
+        case dict():
+            value_type_str = value.get("value_type")
+            if not isinstance(value_type_str, str):
+                raise TypeError(
+                    f"unexpected type for value_type field, value={value_type_str}, type={type(value_type_str)}"
+                )
+            value_type = SegmentType(value_type_str).exposed_type()
+            if value_type not in ENVIRONMENT_VARIABLE_SUPPORTED_TYPES:
+                raise ValueError(f"Unsupported environment variable value type: {value_type}")
+            return value
+
+        case _:
+            return value
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft")
 class DraftWorkflowApi(Resource):
     @console_ns.doc("get_draft_workflow")
     @console_ns.doc(description="Get draft workflow for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Draft workflow retrieved successfully", workflow_model)
+    @console_ns.response(
+        200,
+        "Draft workflow retrieved successfully",
+        console_ns.models[WorkflowResponse.__name__],
+    )
     @console_ns.response(404, "Draft workflow not found")
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_model)
     @edit_permission_required
     def get(self, app_model: App):
         """
@@ -242,8 +374,8 @@ class DraftWorkflowApi(Resource):
         if not workflow:
             raise DraftWorkflowNotExist()
 
-        # return workflow, if not found, return None (initiate graph by frontend)
-        return workflow
+        # return workflow, if not found, return 404
+        return dump_response(WorkflowResponse, workflow)
 
     @setup_required
     @login_required
@@ -817,13 +949,15 @@ class PublishedWorkflowApi(Resource):
     @console_ns.doc("get_published_workflow")
     @console_ns.doc(description="Get published workflow for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Published workflow retrieved successfully", workflow_model)
-    @console_ns.response(404, "Published workflow not found")
+    @console_ns.response(
+        200,
+        "Published workflow retrieved successfully, or null if not found",
+        console_ns.models[WorkflowResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_model)
     @edit_permission_required
     def get(self, app_model: App):
         """
@@ -834,7 +968,10 @@ class PublishedWorkflowApi(Resource):
         workflow = workflow_service.get_published_workflow(app_model=app_model)
 
         # return workflow, if not found, return None
-        return workflow
+        if workflow is None:
+            return None
+
+        return dump_response(WorkflowResponse, workflow)
 
     @console_ns.expect(console_ns.models[PublishWorkflowPayload.__name__])
     @setup_required
@@ -993,7 +1130,11 @@ class PublishedAllWorkflowApi(Resource):
     @console_ns.doc("get_all_published_workflows")
     @console_ns.doc(description="Get all published workflows for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Published workflows retrieved successfully", workflow_pagination_model)
+    @console_ns.response(
+        200,
+        "Published workflows retrieved successfully",
+        console_ns.models[WorkflowPaginationResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
@@ -1025,14 +1166,14 @@ class PublishedAllWorkflowApi(Resource):
                 user_id=user_id,
                 named_only=named_only,
             )
-            serialized_workflows = marshal(workflows, workflow_fields_copy)
-
-            return {
-                "items": serialized_workflows,
-                "page": page,
-                "limit": limit,
-                "has_more": has_more,
-            }
+            return WorkflowPaginationResponse.model_validate(
+                {
+                    "items": workflows,
+                    "page": page,
+                    "limit": limit,
+                    "has_more": has_more,
+                }
+            ).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflows/<string:workflow_id>/restore")
@@ -1078,14 +1219,13 @@ class WorkflowByIdApi(Resource):
     @console_ns.doc(description="Update workflow by ID")
     @console_ns.doc(params={"app_id": "Application ID", "workflow_id": "Workflow ID"})
     @console_ns.expect(console_ns.models[WorkflowUpdatePayload.__name__])
-    @console_ns.response(200, "Workflow updated successfully", workflow_model)
+    @console_ns.response(200, "Workflow updated successfully", console_ns.models[WorkflowResponse.__name__])
     @console_ns.response(404, "Workflow not found")
     @console_ns.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_model)
     @edit_permission_required
     def patch(self, app_model: App, workflow_id: str):
         """
@@ -1119,7 +1259,7 @@ class WorkflowByIdApi(Resource):
             if not workflow:
                 raise NotFound("Workflow not found")
 
-        return workflow
+        return dump_response(WorkflowResponse, workflow)
 
     @setup_required
     @login_required
@@ -1404,12 +1544,16 @@ class DraftWorkflowTriggerRunAllApi(Resource):
 @console_ns.route("/apps/workflows/online-users")
 class WorkflowOnlineUsersApi(Resource):
     @console_ns.expect(console_ns.models[WorkflowOnlineUsersPayload.__name__])
+    @console_ns.response(
+        200,
+        "Workflow online users retrieved successfully",
+        console_ns.models[WorkflowOnlineUsersResponse.__name__],
+    )
     @console_ns.doc("get_workflow_online_users")
     @console_ns.doc(description="Get workflow online users")
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(online_user_list_fields)
     def post(self):
         args = WorkflowOnlineUsersPayload.model_validate(console_ns.payload or {})
 
@@ -1452,10 +1596,18 @@ class WorkflowOnlineUsersApi(Resource):
                 if not isinstance(user_info, dict):
                     continue
 
+                user_id = user_info.get("user_id")
+                username = user_info.get("username")
+                if not isinstance(user_id, str) or not isinstance(username, str):
+                    continue
+
                 avatar = user_info.get("avatar")
+                if avatar is not None and not isinstance(avatar, str):
+                    avatar = None
+
                 if isinstance(avatar, str) and avatar and not avatar.startswith(("http://", "https://")):
                     try:
-                        user_info["avatar"] = file_helpers.get_signed_file_url(avatar)
+                        avatar = file_helpers.get_signed_file_url(avatar)
                     except Exception as exc:
                         logger.warning(
                             "Failed to sign workflow online user avatar; using original value. "
@@ -1465,7 +1617,7 @@ class WorkflowOnlineUsersApi(Resource):
                             exc,
                         )
 
-                users.append(user_info)
+                users.append({"user_id": user_id, "username": username, "avatar": avatar})
             results.append({"app_id": app_id, "users": users})
 
-        return {"data": results}
+        return WorkflowOnlineUsersResponse.model_validate({"data": results}).model_dump(mode="json")

api/controllers/console/app/workflow_app_log.py

@@ -16,6 +16,7 @@ from fields.base import ResponseModel
 from fields.end_user_fields import SimpleEndUser
 from fields.member_fields import SimpleAccount
 from graphon.enums import WorkflowExecutionStatus
+from libs.helper import to_timestamp
 from libs.login import login_required
 from models import App
 from models.model import AppMode
@@ -82,9 +83,7 @@ class WorkflowRunForLogResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class WorkflowRunForArchivedLogResponse(ResponseModel):
@@ -117,9 +116,7 @@ class WorkflowAppLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class WorkflowArchivedLogPartialResponse(ResponseModel):
@@ -133,9 +130,7 @@ class WorkflowArchivedLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPaginationResponse(ResponseModel):

api/controllers/console/app/workflow_comment.py

@@ -1,22 +1,16 @@
 import logging
+from datetime import datetime
 
-from flask_restx import Resource, marshal_with
-from pydantic import BaseModel, Field, TypeAdapter
+from flask_restx import Resource
+from pydantic import BaseModel, Field, TypeAdapter, computed_field, field_validator
 
-from controllers.common.schema import register_schema_models
+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 fields.base import ResponseModel
 from fields.member_fields import AccountWithRole
-from fields.workflow_comment_fields import (
-    workflow_comment_basic_fields,
-    workflow_comment_create_fields,
-    workflow_comment_detail_fields,
-    workflow_comment_reply_create_fields,
-    workflow_comment_reply_update_fields,
-    workflow_comment_resolve_fields,
-    workflow_comment_update_fields,
-)
+from libs.helper import build_avatar_url, dump_response, to_timestamp
 from libs.login import current_user, login_required
 from models import App
 from services.account_service import TenantService
@@ -51,6 +45,138 @@ class WorkflowCommentMentionUsersPayload(BaseModel):
     users: list[AccountWithRole]
 
 
+class WorkflowCommentAccount(ResponseModel):
+    id: str
+    name: str
+    email: str
+    avatar: str | None = Field(default=None, exclude=True)
+
+    @computed_field(return_type=str | None)  # type: ignore[prop-decorator]
+    @property
+    def avatar_url(self) -> str | None:
+        return build_avatar_url(self.avatar)
+
+
+class WorkflowCommentReply(ResponseModel):
+    id: str
+    content: str
+    created_by: str
+    created_by_account: WorkflowCommentAccount | None = None
+    created_at: int | None = None
+
+    @field_validator("created_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentMention(ResponseModel):
+    mentioned_user_id: str
+    mentioned_user_account: WorkflowCommentAccount | None = None
+    reply_id: str | None = None
+
+
+class WorkflowCommentBasic(ResponseModel):
+    id: str
+    position_x: float
+    position_y: float
+    content: str
+    created_by: str
+    created_by_account: WorkflowCommentAccount | None = None
+    created_at: int | None = None
+    updated_at: int | None = None
+    resolved: bool
+    resolved_at: int | None = None
+    resolved_by: str | None = None
+    resolved_by_account: WorkflowCommentAccount | None = None
+    reply_count: int
+    mention_count: int
+    participants: list[WorkflowCommentAccount]
+
+    @field_validator("created_at", "updated_at", "resolved_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentBasicList(ResponseModel):
+    data: list[WorkflowCommentBasic]
+
+
+class WorkflowCommentDetail(ResponseModel):
+    id: str
+    position_x: float
+    position_y: float
+    content: str
+    created_by: str
+    created_by_account: WorkflowCommentAccount | None = None
+    created_at: int | None = None
+    updated_at: int | None = None
+    resolved: bool
+    resolved_at: int | None = None
+    resolved_by: str | None = None
+    resolved_by_account: WorkflowCommentAccount | None = None
+    replies: list[WorkflowCommentReply]
+    mentions: list[WorkflowCommentMention]
+
+    @field_validator("created_at", "updated_at", "resolved_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentCreate(ResponseModel):
+    id: str
+    created_at: int | None = None
+
+    @field_validator("created_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentUpdate(ResponseModel):
+    id: str
+    updated_at: int | None = None
+
+    @field_validator("updated_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentResolve(ResponseModel):
+    id: str
+    resolved: bool
+    resolved_at: int | None = None
+    resolved_by: str | None = None
+
+    @field_validator("resolved_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentReplyCreate(ResponseModel):
+    id: str
+    created_at: int | None = None
+
+    @field_validator("created_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentReplyUpdate(ResponseModel):
+    id: str
+    updated_at: int | None = None
+
+    @field_validator("updated_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
 register_schema_models(
     console_ns,
     AccountWithRole,
@@ -59,17 +185,19 @@ register_schema_models(
     WorkflowCommentUpdatePayload,
     WorkflowCommentReplyPayload,
 )
-
-workflow_comment_basic_model = console_ns.model("WorkflowCommentBasic", workflow_comment_basic_fields)
-workflow_comment_detail_model = console_ns.model("WorkflowCommentDetail", workflow_comment_detail_fields)
-workflow_comment_create_model = console_ns.model("WorkflowCommentCreate", workflow_comment_create_fields)
-workflow_comment_update_model = console_ns.model("WorkflowCommentUpdate", workflow_comment_update_fields)
-workflow_comment_resolve_model = console_ns.model("WorkflowCommentResolve", workflow_comment_resolve_fields)
-workflow_comment_reply_create_model = console_ns.model(
-    "WorkflowCommentReplyCreate", workflow_comment_reply_create_fields
-)
-workflow_comment_reply_update_model = console_ns.model(
-    "WorkflowCommentReplyUpdate", workflow_comment_reply_update_fields
+register_response_schema_models(
+    console_ns,
+    WorkflowCommentAccount,
+    WorkflowCommentReply,
+    WorkflowCommentMention,
+    WorkflowCommentBasic,
+    WorkflowCommentBasicList,
+    WorkflowCommentDetail,
+    WorkflowCommentCreate,
+    WorkflowCommentUpdate,
+    WorkflowCommentResolve,
+    WorkflowCommentReplyCreate,
+    WorkflowCommentReplyUpdate,
 )
 
 
@@ -80,28 +208,26 @@ class WorkflowCommentListApi(Resource):
     @console_ns.doc("list_workflow_comments")
     @console_ns.doc(description="Get all comments for a workflow")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Comments retrieved successfully", workflow_comment_basic_model)
+    @console_ns.response(200, "Comments retrieved successfully", console_ns.models[WorkflowCommentBasicList.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_basic_model, envelope="data")
     def get(self, 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)
 
-        return comments
+        return WorkflowCommentBasicList.model_validate({"data": comments}).model_dump(mode="json")
 
     @console_ns.doc("create_workflow_comment")
     @console_ns.doc(description="Create a new workflow comment")
     @console_ns.doc(params={"app_id": "Application ID"})
     @console_ns.expect(console_ns.models[WorkflowCommentCreatePayload.__name__])
-    @console_ns.response(201, "Comment created successfully", workflow_comment_create_model)
+    @console_ns.response(201, "Comment created successfully", console_ns.models[WorkflowCommentCreate.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_create_model)
     @edit_permission_required
     def post(self, app_model: App):
         """Create a new workflow comment."""
@@ -117,7 +243,7 @@ class WorkflowCommentListApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return result, 201
+        return dump_response(WorkflowCommentCreate, result), 201
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>")
@@ -127,30 +253,28 @@ class WorkflowCommentDetailApi(Resource):
     @console_ns.doc("get_workflow_comment")
     @console_ns.doc(description="Get a specific workflow comment")
     @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
-    @console_ns.response(200, "Comment retrieved successfully", workflow_comment_detail_model)
+    @console_ns.response(200, "Comment retrieved successfully", console_ns.models[WorkflowCommentDetail.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_detail_model)
     def get(self, 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
         )
 
-        return comment
+        return dump_response(WorkflowCommentDetail, comment)
 
     @console_ns.doc("update_workflow_comment")
     @console_ns.doc(description="Update a workflow comment")
     @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
     @console_ns.expect(console_ns.models[WorkflowCommentUpdatePayload.__name__])
-    @console_ns.response(200, "Comment updated successfully", workflow_comment_update_model)
+    @console_ns.response(200, "Comment updated successfully", console_ns.models[WorkflowCommentUpdate.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_update_model)
     @edit_permission_required
     def put(self, app_model: App, comment_id: str):
         """Update a workflow comment."""
@@ -167,7 +291,7 @@ class WorkflowCommentDetailApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return result
+        return dump_response(WorkflowCommentUpdate, result)
 
     @console_ns.doc("delete_workflow_comment")
     @console_ns.doc(description="Delete a workflow comment")
@@ -197,12 +321,11 @@ class WorkflowCommentResolveApi(Resource):
     @console_ns.doc("resolve_workflow_comment")
     @console_ns.doc(description="Resolve a workflow comment")
     @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
-    @console_ns.response(200, "Comment resolved successfully", workflow_comment_resolve_model)
+    @console_ns.response(200, "Comment resolved successfully", console_ns.models[WorkflowCommentResolve.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_resolve_model)
     @edit_permission_required
     def post(self, app_model: App, comment_id: str):
         """Resolve a workflow comment."""
@@ -213,7 +336,7 @@ class WorkflowCommentResolveApi(Resource):
             user_id=current_user.id,
         )
 
-        return comment
+        return dump_response(WorkflowCommentResolve, comment)
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies")
@@ -224,12 +347,11 @@ class WorkflowCommentReplyApi(Resource):
     @console_ns.doc(description="Add a reply to a workflow comment")
     @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
     @console_ns.expect(console_ns.models[WorkflowCommentReplyPayload.__name__])
-    @console_ns.response(201, "Reply created successfully", workflow_comment_reply_create_model)
+    @console_ns.response(201, "Reply created successfully", console_ns.models[WorkflowCommentReplyCreate.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_reply_create_model)
     @edit_permission_required
     def post(self, app_model: App, comment_id: str):
         """Add a reply to a workflow comment."""
@@ -247,7 +369,7 @@ class WorkflowCommentReplyApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return result, 201
+        return dump_response(WorkflowCommentReplyCreate, result), 201
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies/<string:reply_id>")
@@ -258,12 +380,11 @@ class WorkflowCommentReplyDetailApi(Resource):
     @console_ns.doc(description="Update a comment reply")
     @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID", "reply_id": "Reply ID"})
     @console_ns.expect(console_ns.models[WorkflowCommentReplyPayload.__name__])
-    @console_ns.response(200, "Reply updated successfully", workflow_comment_reply_update_model)
+    @console_ns.response(200, "Reply updated successfully", console_ns.models[WorkflowCommentReplyUpdate.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_reply_update_model)
     @edit_permission_required
     def put(self, app_model: App, comment_id: str, reply_id: str):
         """Update a comment reply."""
@@ -284,7 +405,7 @@ class WorkflowCommentReplyDetailApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return reply
+        return dump_response(WorkflowCommentReplyUpdate, reply)
 
     @console_ns.doc("delete_workflow_comment_reply")
     @console_ns.doc(description="Delete a comment reply")

api/controllers/console/auth/email_register.py

@@ -3,7 +3,7 @@ from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 
 from configs import dify_config
-from constants.languages import languages
+from constants.languages import get_valid_language, languages
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
@@ -15,11 +15,12 @@ from controllers.console.auth.error import (
     PasswordMismatchError,
 )
 from libs.helper import EmailStr, extract_remote_ip
+from libs.helper import timezone as validate_timezone_string
 from libs.password import valid_password
 from models import Account
 from services.account_service import AccountService
 from services.billing_service import BillingService
-from services.errors.account import AccountNotFoundError, AccountRegisterError
+from services.errors.account import AccountRegisterError
 
 from ..error import AccountInFreezeError, EmailSendIpLimitError
 from ..wraps import email_password_login_enabled, email_register_enabled, setup_required
@@ -40,12 +41,21 @@ class EmailRegisterResetPayload(BaseModel):
     token: str = Field(...)
     new_password: str = Field(...)
     password_confirm: str = Field(...)
+    language: str | None = Field(default=None)
+    timezone: str | None = Field(default=None)
 
     @field_validator("new_password", "password_confirm")
     @classmethod
     def validate_password(cls, value: str) -> str:
         return valid_password(value)
 
+    @field_validator("timezone")
+    @classmethod
+    def validate_timezone(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        return validate_timezone_string(value)
+
 
 register_schema_models(console_ns, EmailRegisterSendPayload, EmailRegisterValidityPayload, EmailRegisterResetPayload)
 
@@ -144,26 +154,32 @@ class EmailRegisterResetApi(Resource):
 
         if account:
             raise EmailAlreadyInUseError()
-        else:
-            account = self._create_new_account(normalized_email, args.password_confirm)
-            if not account:
-                raise AccountNotFoundError()
-            token_pair = AccountService.login(account=account, ip_address=extract_remote_ip(request))
-            AccountService.reset_login_error_rate_limit(normalized_email)
+
+        account = self._create_new_account(
+            email=normalized_email,
+            password=args.password_confirm,
+            timezone=args.timezone,
+            language=args.language,
+        )
+        token_pair = AccountService.login(account=account, ip_address=extract_remote_ip(request))
+        AccountService.reset_login_error_rate_limit(normalized_email)
 
         return {"result": "success", "data": token_pair.model_dump()}
 
-    def _create_new_account(self, email: str, password: str) -> Account | None:
-        # Create new account if allowed
-        account = None
+    def _create_new_account(
+        self,
+        email: str,
+        password: str,
+        timezone: str | None = None,
+        language: str | None = None,
+    ) -> Account:
         try:
-            account = AccountService.create_account_and_tenant(
+            return AccountService.create_account_and_tenant(
                 email=email,
                 name=email,
                 password=password,
-                interface_language=languages[0],
+                interface_language=get_valid_language(language),
+                timezone=timezone,
             )
         except AccountRegisterError:
             raise AccountInFreezeError()
-
-        return account

api/controllers/console/auth/login.py

@@ -3,7 +3,7 @@ import logging
 import flask_login
 from flask import make_response, request
 from flask_restx import Resource
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 from werkzeug.exceptions import Unauthorized
 
 import services
@@ -34,6 +34,7 @@ from controllers.console.wraps import (
 )
 from events.tenant_event import tenant_was_created
 from libs.helper import EmailStr, extract_remote_ip
+from libs.helper import timezone as validate_timezone_string
 from libs.login import current_account_with_tenant
 from libs.token import (
     clear_access_token_from_cookie,
@@ -69,6 +70,14 @@ class EmailCodeLoginPayload(BaseModel):
     code: str = Field(...)
     token: str = Field(...)
     language: str | None = Field(default=None)
+    timezone: str | None = Field(default=None)
+
+    @field_validator("timezone")
+    @classmethod
+    def validate_timezone(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        return validate_timezone_string(value)
 
 
 register_schema_models(console_ns, LoginPayload, EmailPayload, EmailCodeLoginPayload)
@@ -288,6 +297,7 @@ class EmailCodeLoginApi(Resource):
                     email=user_email,
                     name=user_email,
                     interface_language=get_valid_language(language),
+                    timezone=args.timezone,
                 )
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()

api/controllers/console/auth/oauth.py

@@ -12,7 +12,8 @@ from events.tenant_event import tenant_was_created
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from libs.helper import extract_remote_ip
-from libs.oauth import GitHubOAuth, GoogleOAuth, OAuthUserInfo
+from libs.helper import timezone as validate_timezone_string
+from libs.oauth import GitHubOAuth, GoogleOAuth, OAuthUserInfo, decode_oauth_state
 from libs.token import (
     set_access_token_to_cookie,
     set_csrf_token_to_cookie,
@@ -53,6 +54,31 @@ def get_oauth_providers():
         return OAUTH_PROVIDERS
 
 
+def _validated_timezone(value: str | None) -> str | None:
+    if not value:
+        return None
+    try:
+        return validate_timezone_string(value)
+    except ValueError:
+        return None
+
+
+def _validated_language(value: str | None) -> str | None:
+    if value and value in languages:
+        return value
+    return None
+
+
+def _preferred_interface_language(language: str | None = None) -> str:
+    if language:
+        return language
+
+    preferred_lang = request.accept_languages.best_match(languages)
+    if preferred_lang and preferred_lang in languages:
+        return preferred_lang
+    return languages[0]
+
+
 @console_ns.route("/oauth/login/<provider>")
 class OAuthLogin(Resource):
     @console_ns.doc("oauth_login")
@@ -64,13 +90,19 @@ class OAuthLogin(Resource):
     @console_ns.response(400, "Invalid provider")
     def get(self, provider: str):
         invite_token = request.args.get("invite_token") or None
+        timezone = _validated_timezone(request.args.get("timezone") or None)
+        language = _validated_language(request.args.get("language") or None)
         OAUTH_PROVIDERS = get_oauth_providers()
         with current_app.app_context():
             oauth_provider = OAUTH_PROVIDERS.get(provider)
         if not oauth_provider:
             return {"error": "Invalid provider"}, 400
 
-        auth_url = oauth_provider.get_authorization_url(invite_token=invite_token)
+        auth_url = oauth_provider.get_authorization_url(
+            invite_token=invite_token,
+            timezone=timezone,
+            language=language,
+        )
         return redirect(auth_url)
 
 
@@ -96,9 +128,10 @@ class OAuthCallback(Resource):
 
         code = request.args.get("code")
         state = request.args.get("state")
-        invite_token = None
-        if state:
-            invite_token = state
+        oauth_state = decode_oauth_state(state)
+        invite_token = oauth_state.get("invite_token")
+        timezone = _validated_timezone(oauth_state.get("timezone"))
+        language = _validated_language(oauth_state.get("language"))
 
         if not code:
             return {"error": "Authorization code is required"}, 400
@@ -129,7 +162,7 @@ class OAuthCallback(Resource):
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin/invite-settings?invite_token={invite_token}")
 
         try:
-            account, oauth_new_user = _generate_account(provider, user_info)
+            account, oauth_new_user = _generate_account(provider, user_info, timezone=timezone, language=language)
         except AccountNotFoundError:
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Account not found.")
         except (WorkSpaceNotFoundError, WorkSpaceNotAllowedCreateError):
@@ -184,7 +217,12 @@ def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) ->
     return account
 
 
-def _generate_account(provider: str, user_info: OAuthUserInfo) -> tuple[Account, bool]:
+def _generate_account(
+    provider: str,
+    user_info: OAuthUserInfo,
+    timezone: str | None = None,
+    language: str | None = None,
+) -> tuple[Account, bool]:
     # Get account by openid or email.
     account = _get_account_by_openid_or_email(provider, user_info)
     oauth_new_user = False
@@ -211,26 +249,19 @@ def _generate_account(provider: str, user_info: OAuthUserInfo) -> tuple[Account,
                         "30 days and is temporarily unavailable for new account registration"
                     )
                 )
-            else:
-                raise AccountRegisterError(description=("Invalid email or password"))
+            raise AccountRegisterError(description=("Invalid email or password"))
         account_name = user_info.name or "Dify"
+        interface_language = _preferred_interface_language(language)
         account = RegisterService.register(
             email=normalized_email,
             name=account_name,
             password=None,
             open_id=user_info.id,
             provider=provider,
+            language=interface_language,
+            timezone=timezone,
         )
 
-        # Set interface language
-        preferred_lang = request.accept_languages.best_match(languages)
-        if preferred_lang and preferred_lang in languages:
-            interface_language = preferred_lang
-        else:
-            interface_language = languages[0]
-        account.interface_language = interface_language
-        db.session.commit()
-
     # Link account
     AccountService.link_account_integrate(provider, user_info.id, account)
 

api/controllers/console/datasets/datasets.py

@@ -606,63 +606,63 @@ class DatasetIndexingEstimateApi(Resource):
         # validate args
         DocumentService.estimate_args_validate(args)
         extract_settings = []
-        if args["info_list"]["data_source_type"] == "upload_file":
-            file_ids = args["info_list"]["file_info_list"]["file_ids"]
-            file_details = db.session.scalars(
-                select(UploadFile).where(UploadFile.tenant_id == current_tenant_id, UploadFile.id.in_(file_ids))
-            ).all()
+        match args["info_list"]["data_source_type"]:
+            case "upload_file":
+                file_ids = args["info_list"]["file_info_list"]["file_ids"]
+                file_details = db.session.scalars(
+                    select(UploadFile).where(UploadFile.tenant_id == current_tenant_id, UploadFile.id.in_(file_ids))
+                ).all()
+                if file_details is None:
+                    raise NotFound("File not found.")
 
-            if file_details is None:
-                raise NotFound("File not found.")
-
-            if file_details:
-                for file_detail in file_details:
+                if file_details:
+                    for file_detail in file_details:
+                        extract_setting = ExtractSetting(
+                            datasource_type=DatasourceType.FILE,
+                            upload_file=file_detail,
+                            document_model=args["doc_form"],
+                        )
+                        extract_settings.append(extract_setting)
+            case "notion_import":
+                notion_info_list = args["info_list"]["notion_info_list"]
+                for notion_info in notion_info_list:
+                    workspace_id = notion_info["workspace_id"]
+                    credential_id = notion_info.get("credential_id")
+                    for page in notion_info["pages"]:
+                        extract_setting = ExtractSetting(
+                            datasource_type=DatasourceType.NOTION,
+                            notion_info=NotionInfo.model_validate(
+                                {
+                                    "credential_id": credential_id,
+                                    "notion_workspace_id": workspace_id,
+                                    "notion_obj_id": page["page_id"],
+                                    "notion_page_type": page["type"],
+                                    "tenant_id": current_tenant_id,
+                                }
+                            ),
+                            document_model=args["doc_form"],
+                        )
+                        extract_settings.append(extract_setting)
+            case "website_crawl":
+                website_info_list = args["info_list"]["website_info_list"]
+                for url in website_info_list["urls"]:
                     extract_setting = ExtractSetting(
-                        datasource_type=DatasourceType.FILE,
-                        upload_file=file_detail,
-                        document_model=args["doc_form"],
-                    )
-                    extract_settings.append(extract_setting)
-        elif args["info_list"]["data_source_type"] == "notion_import":
-            notion_info_list = args["info_list"]["notion_info_list"]
-            for notion_info in notion_info_list:
-                workspace_id = notion_info["workspace_id"]
-                credential_id = notion_info.get("credential_id")
-                for page in notion_info["pages"]:
-                    extract_setting = ExtractSetting(
-                        datasource_type=DatasourceType.NOTION,
-                        notion_info=NotionInfo.model_validate(
+                        datasource_type=DatasourceType.WEBSITE,
+                        website_info=WebsiteInfo.model_validate(
                             {
-                                "credential_id": credential_id,
-                                "notion_workspace_id": workspace_id,
-                                "notion_obj_id": page["page_id"],
-                                "notion_page_type": page["type"],
+                                "provider": website_info_list["provider"],
+                                "job_id": website_info_list["job_id"],
+                                "url": url,
                                 "tenant_id": current_tenant_id,
+                                "mode": "crawl",
+                                "only_main_content": website_info_list["only_main_content"],
                             }
                         ),
                         document_model=args["doc_form"],
                     )
                     extract_settings.append(extract_setting)
-        elif args["info_list"]["data_source_type"] == "website_crawl":
-            website_info_list = args["info_list"]["website_info_list"]
-            for url in website_info_list["urls"]:
-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.WEBSITE,
-                    website_info=WebsiteInfo.model_validate(
-                        {
-                            "provider": website_info_list["provider"],
-                            "job_id": website_info_list["job_id"],
-                            "url": url,
-                            "tenant_id": current_tenant_id,
-                            "mode": "crawl",
-                            "only_main_content": website_info_list["only_main_content"],
-                        }
-                    ),
-                    document_model=args["doc_form"],
-                )
-                extract_settings.append(extract_setting)
-        else:
-            raise ValueError("Data source type not support")
+            case _:
+                raise ValueError("Data source type not support")
         indexing_runner = IndexingRunner()
         try:
             response = indexing_runner.indexing_estimate(

api/controllers/console/datasets/datasets_document.py

@@ -39,6 +39,7 @@ from fields.document_fields import (
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from libs.datetime_utils import naive_utc_now
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import DatasetProcessRule, Document, DocumentSegment, UploadFile
 from models.dataset import DocumentPipelineExecutionLog
@@ -71,12 +72,6 @@ from ..wraps import (
 logger = logging.getLogger(__name__)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 def _normalize_enum(value: Any) -> Any:
     if isinstance(value, str) or value is None:
         return value
@@ -101,7 +96,7 @@ class DatasetResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class DocumentMetadataResponse(ResponseModel):
@@ -152,7 +147,7 @@ class DocumentResponse(ResponseModel):
     @field_validator("created_at", "disabled_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class DocumentWithSegmentsResponse(DocumentResponse):
@@ -369,28 +364,31 @@ class DatasetDocumentListApi(Resource):
         else:
             sort_logic = asc
 
-        if sort == "hit_count":
-            sub_query = (
-                sa.select(DocumentSegment.document_id, sa.func.sum(DocumentSegment.hit_count).label("total_hit_count"))
-                .where(DocumentSegment.dataset_id == str(dataset_id))
-                .group_by(DocumentSegment.document_id)
-                .subquery()
-            )
+        match sort:
+            case "hit_count":
+                sub_query = (
+                    sa.select(
+                        DocumentSegment.document_id, sa.func.sum(DocumentSegment.hit_count).label("total_hit_count")
+                    )
+                    .where(DocumentSegment.dataset_id == str(dataset_id))
+                    .group_by(DocumentSegment.document_id)
+                    .subquery()
+                )
 
-            query = query.outerjoin(sub_query, sub_query.c.document_id == Document.id).order_by(
-                sort_logic(sa.func.coalesce(sub_query.c.total_hit_count, 0)),
-                sort_logic(Document.position),
-            )
-        elif sort == "created_at":
-            query = query.order_by(
-                sort_logic(Document.created_at),
-                sort_logic(Document.position),
-            )
-        else:
-            query = query.order_by(
-                desc(Document.created_at),
-                desc(Document.position),
-            )
+                query = query.outerjoin(sub_query, sub_query.c.document_id == Document.id).order_by(
+                    sort_logic(sa.func.coalesce(sub_query.c.total_hit_count, 0)),
+                    sort_logic(Document.position),
+                )
+            case "created_at":
+                query = query.order_by(
+                    sort_logic(Document.created_at),
+                    sort_logic(Document.position),
+                )
+            case _:
+                query = query.order_by(
+                    desc(Document.created_at),
+                    desc(Document.position),
+                )
 
         paginated_documents = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
         documents = paginated_documents.items

api/controllers/console/datasets/hit_testing.py

@@ -8,6 +8,7 @@ from pydantic import Field, field_validator
 
 from controllers.common.schema import register_schema_models
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import login_required
 
 from .. import console_ns
@@ -19,12 +20,6 @@ from ..wraps import (
 )
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class HitTestingDocument(ResponseModel):
     id: str | None = None
     data_source_type: str | None = None
@@ -61,7 +56,7 @@ class HitTestingSegment(ResponseModel):
     @field_validator("disabled_at", "created_at", "indexing_at", "completed_at", "stopped_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class HitTestingChildChunk(ResponseModel):

api/controllers/console/datasets/hit_testing_base.py

@@ -39,11 +39,8 @@ class HitTestingPayload(BaseModel):
 
 class DatasetsHitTestingBase:
     @staticmethod
-    def _normalize_hit_testing_query(query: Any) -> str:
-        """Return the user-visible query string from legacy and current response shapes."""
-        if isinstance(query, str):
-            return query
-
+    def _extract_hit_testing_query(query: Any) -> str:
+        """Return the query string from the service response shape."""
         if isinstance(query, dict):
             content = query.get("content")
             if isinstance(content, str):
@@ -52,15 +49,15 @@ class DatasetsHitTestingBase:
         raise ValueError("Invalid hit testing query response")
 
     @staticmethod
-    def _normalize_hit_testing_records(records: Any) -> list[dict[str, Any]]:
-        """Coerce nullable collection fields into lists before response validation."""
+    def _prepare_hit_testing_records(records: Any) -> list[dict[str, Any]]:
+        """Ensure collection fields match the API schema before response validation."""
         if not isinstance(records, list):
-            return []
+            raise ValueError("Invalid hit testing records response")
 
         normalized_records: list[dict[str, Any]] = []
         for record in records:
             if not isinstance(record, dict):
-                continue
+                raise ValueError("Invalid hit testing record response")
 
             normalized_record = dict(record)
             segment = normalized_record.get("segment")
@@ -118,8 +115,8 @@ class DatasetsHitTestingBase:
                 limit=10,
             )
             return {
-                "query": DatasetsHitTestingBase._normalize_hit_testing_query(response.get("query")),
-                "records": DatasetsHitTestingBase._normalize_hit_testing_records(
+                "query": DatasetsHitTestingBase._extract_hit_testing_query(response.get("query")),
+                "records": DatasetsHitTestingBase._prepare_hit_testing_records(
                     marshal(response.get("records", []), hit_testing_record_fields)
                 ),
             }

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

@@ -3,7 +3,7 @@ import logging
 from typing import Any, Literal, cast
 
 from flask import abort, request
-from flask_restx import Resource, marshal_with  # type: ignore
+from flask_restx import Resource
 from pydantic import BaseModel, Field, ValidationError
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
@@ -19,8 +19,8 @@ from controllers.console.app.error import (
 )
 from controllers.console.app.workflow import (
     RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE,
-    workflow_model,
-    workflow_pagination_model,
+    WorkflowPaginationResponse,
+    WorkflowResponse,
 )
 from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import (
@@ -42,7 +42,7 @@ from fields.workflow_run_fields import (
 )
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs import helper
-from libs.helper import TimestampField, UUIDStrOrEmpty
+from libs.helper import TimestampField, UUIDStrOrEmpty, dump_response
 from libs.login import current_account_with_tenant, current_user, login_required
 from models import Account
 from models.dataset import Pipeline
@@ -142,12 +142,17 @@ register_response_schema_models(
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft")
 class DraftRagPipelineApi(Resource):
+    @console_ns.response(
+        200,
+        "Draft workflow retrieved successfully",
+        console_ns.models[WorkflowResponse.__name__],
+    )
+    @console_ns.response(404, "Draft workflow not found")
     @setup_required
     @login_required
     @account_initialization_required
     @get_rag_pipeline
     @edit_permission_required
-    @marshal_with(workflow_model)
     def get(self, pipeline: Pipeline):
         """
         Get draft rag pipeline's workflow
@@ -159,8 +164,8 @@ class DraftRagPipelineApi(Resource):
         if not workflow:
             raise DraftWorkflowNotExist()
 
-        # return workflow, if not found, return None (initiate graph by frontend)
-        return workflow
+        # return workflow, if not found, return 404
+        return dump_response(WorkflowResponse, workflow)
 
     @setup_required
     @login_required
@@ -476,12 +481,16 @@ class RagPipelineTaskStopApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/publish")
 class PublishedRagPipelineApi(Resource):
+    @console_ns.response(
+        200,
+        "Published workflow retrieved successfully, or null if not exist",
+        console_ns.models[WorkflowResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @edit_permission_required
     @get_rag_pipeline
-    @marshal_with(workflow_model)
     def get(self, pipeline: Pipeline):
         """
         Get published pipeline
@@ -494,7 +503,10 @@ class PublishedRagPipelineApi(Resource):
         workflow = rag_pipeline_service.get_published_workflow(pipeline=pipeline)
 
         # return workflow, if not found, return None
-        return workflow
+        if workflow is None:
+            return None
+
+        return dump_response(WorkflowResponse, workflow)
 
     @setup_required
     @login_required
@@ -567,12 +579,17 @@ class DefaultRagPipelineBlockConfigApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows")
 class PublishedAllRagPipelineApi(Resource):
+    @console_ns.response(
+        200,
+        "Published workflows retrieved successfully",
+        console_ns.models[WorkflowPaginationResponse.__name__],
+    )
+    @console_ns.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
     @edit_permission_required
     @get_rag_pipeline
-    @marshal_with(workflow_pagination_model)
     def get(self, pipeline: Pipeline):
         """
         Get published workflows
@@ -601,12 +618,14 @@ class PublishedAllRagPipelineApi(Resource):
                 named_only=named_only,
             )
 
-            return {
-                "items": workflows,
-                "page": page,
-                "limit": limit,
-                "has_more": has_more,
-            }
+            return WorkflowPaginationResponse.model_validate(
+                {
+                    "items": workflows,
+                    "page": page,
+                    "limit": limit,
+                    "has_more": has_more,
+                }
+            ).model_dump(mode="json")
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/<string:workflow_id>/restore")
@@ -641,12 +660,15 @@ class RagPipelineDraftWorkflowRestoreApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/<string:workflow_id>")
 class RagPipelineByIdApi(Resource):
+    @console_ns.response(200, "Workflow updated successfully", console_ns.models[WorkflowResponse.__name__])
+    @console_ns.response(400, "No valid fields to update")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Workflow not found")
     @setup_required
     @login_required
     @account_initialization_required
     @edit_permission_required
     @get_rag_pipeline
-    @marshal_with(workflow_model)
     def patch(self, pipeline: Pipeline, workflow_id: str):
         """
         Update workflow attributes
@@ -675,7 +697,7 @@ class RagPipelineByIdApi(Resource):
             if not workflow:
                 raise NotFound("Workflow not found")
 
-            return workflow
+            return dump_response(WorkflowResponse, workflow)
 
     @setup_required
     @login_required

api/controllers/console/explore/installed_app.py

@@ -16,6 +16,7 @@ from extensions.ext_database import db
 from fields.base import ResponseModel
 from graphon.file import helpers as file_helpers
 from libs.datetime_utils import naive_utc_now
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import App, InstalledApp, RecommendedApp
 from models.model import IconType
@@ -105,9 +106,7 @@ class InstalledAppResponse(ResponseModel):
     @field_validator("last_used_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class InstalledAppListResponse(ResponseModel):

api/controllers/console/explore/trial.py

@@ -106,7 +106,7 @@ app_detail_fields_with_site_copy["tags"] = fields.List(fields.Nested(tag_model))
 app_detail_fields_with_site_copy["site"] = fields.Nested(site_model)
 app_detail_with_site_model = get_or_create_model("TrialAppDetailWithSite", app_detail_fields_with_site_copy)
 
-simple_account_model = get_or_create_model("SimpleAccount", simple_account_fields)
+simple_account_model = get_or_create_model("TrialSimpleAccount", simple_account_fields)
 conversation_variable_model = get_or_create_model("TrialConversationVariable", conversation_variable_fields)
 pipeline_variable_model = get_or_create_model("TrialPipelineVariable", pipeline_variable_fields)
 

api/controllers/console/extension.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, Field, TypeAdapter, field_validator
 
 from constants import HIDDEN_VALUE
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.api_based_extension import APIBasedExtension
 from services.api_based_extension_service import APIBasedExtensionService
@@ -40,12 +41,6 @@ def _mask_api_key(api_key: str) -> str:
     return api_key[:3] + "******" + api_key[-3:]
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class APIBasedExtensionResponse(ResponseModel):
     id: str
     name: str
@@ -61,7 +56,7 @@ class APIBasedExtensionResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 register_schema_models(console_ns, APIBasedExtensionPayload, CodeBasedExtensionResponse, APIBasedExtensionResponse)

api/controllers/console/files.py

@@ -105,7 +105,8 @@ class FilePreviewApi(Resource):
     @account_initialization_required
     def get(self, file_id):
         file_id = str(file_id)
-        text = FileService(db.engine).get_file_preview(file_id)
+        _, tenant_id = current_account_with_tenant()
+        text = FileService(db.engine).get_file_preview(file_id, tenant_id)
         return {"content": text}
 
 

api/controllers/console/tag/tags.py

@@ -25,6 +25,10 @@ class TagBasePayload(BaseModel):
     type: TagType = Field(description="Tag type")
 
 
+class TagUpdateRequestPayload(BaseModel):
+    name: str = Field(description="Tag name", min_length=1, max_length=50)
+
+
 class TagBindingPayload(BaseModel):
     tag_ids: list[str] = Field(description="Tag IDs to bind")
     target_id: str = Field(description="Target ID to bind tags to")
@@ -68,6 +72,7 @@ class TagResponse(ResponseModel):
 register_schema_models(
     console_ns,
     TagBasePayload,
+    TagUpdateRequestPayload,
     TagBindingPayload,
     TagBindingRemovePayload,
     TagListQueryParam,
@@ -118,7 +123,7 @@ class TagListApi(Resource):
 
 @console_ns.route("/tags/<uuid:tag_id>")
 class TagUpdateDeleteApi(Resource):
-    @console_ns.expect(console_ns.models[TagBasePayload.__name__])
+    @console_ns.expect(console_ns.models[TagUpdateRequestPayload.__name__])
     @setup_required
     @login_required
     @account_initialization_required
@@ -129,8 +134,8 @@ class TagUpdateDeleteApi(Resource):
         if not (current_user.has_edit_permission or current_user.is_dataset_editor):
             raise Forbidden()
 
-        payload = TagBasePayload.model_validate(console_ns.payload or {})
-        tag = TagService.update_tags(UpdateTagPayload(name=payload.name, type=payload.type), tag_id)
+        payload = TagUpdateRequestPayload.model_validate(console_ns.payload or {})
+        tag = TagService.update_tags(UpdateTagPayload(name=payload.name), tag_id)
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 

api/controllers/console/workspace/account.py

@@ -42,7 +42,7 @@ from fields.base import ResponseModel
 from fields.member_fields import Account as AccountResponse
 from graphon.file import helpers as file_helpers
 from libs.datetime_utils import naive_utc_now
-from libs.helper import EmailStr, extract_remote_ip, timezone
+from libs.helper import EmailStr, extract_remote_ip, timezone, to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import AccountIntegrate, InvitationCode
 from models.account import AccountStatus, InvitationCodeStatus
@@ -185,12 +185,6 @@ def _serialize_account(account) -> dict[str, Any]:
     return AccountResponse.model_validate(account, from_attributes=True).model_dump(mode="json")
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class AccountIntegrateResponse(ResponseModel):
     provider: str
     created_at: int | None = None
@@ -200,7 +194,7 @@ class AccountIntegrateResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AccountIntegrateListResponse(ResponseModel):
@@ -220,7 +214,7 @@ class EducationStatusResponse(ResponseModel):
     @field_validator("expire_at", mode="before")
     @classmethod
     def _normalize_expire_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class EducationAutocompleteResponse(ResponseModel):

api/controllers/console/workspace/members.py

@@ -70,6 +70,12 @@ register_schema_models(
 )
 
 
+def _is_role_enabled(role: TenantAccountRole | str, tenant_id: str) -> bool:
+    if role != TenantAccountRole.DATASET_OPERATOR:
+        return True
+    return FeatureService.get_features(tenant_id=tenant_id).dataset_operator_enabled
+
+
 @console_ns.route("/workspaces/current/members")
 class MemberListApi(Resource):
     """List all members of current tenant."""
@@ -110,6 +116,8 @@ class MemberInviteEmailApi(Resource):
         inviter = current_user
         if not inviter.current_tenant:
             raise ValueError("No current tenant")
+        if not _is_role_enabled(invitee_role, inviter.current_tenant.id):
+            return {"code": "invalid-role", "message": "Invalid role"}, 400
 
         # Check workspace permission for member invitations
         from libs.workspace_permission import check_workspace_member_invite_permission
@@ -208,6 +216,8 @@ class MemberUpdateRoleApi(Resource):
         current_user, _ = current_account_with_tenant()
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
+        if not _is_role_enabled(new_role, current_user.current_tenant.id):
+            return {"code": "invalid-role", "message": "Invalid role"}, 400
         member = db.session.get(Account, str(member_id))
         if not member:
             abort(404)
@@ -215,11 +225,17 @@ class MemberUpdateRoleApi(Resource):
         try:
             assert member is not None, "Member not found"
             TenantService.update_member_role(current_user.current_tenant, member, new_role, current_user)
+        except services.errors.account.CannotOperateSelfError as e:
+            return {"code": "cannot-operate-self", "message": str(e)}, 400
+        except services.errors.account.NoPermissionError as e:
+            return {"code": "forbidden", "message": str(e)}, 403
+        except services.errors.account.MemberNotInTenantError as e:
+            return {"code": "member-not-found", "message": str(e)}, 404
+        except services.errors.account.RoleAlreadyAssignedError as e:
+            return {"code": "role-already-assigned", "message": str(e)}, 400
         except Exception as e:
             raise ValueError(str(e))
 
-        # todo: 403
-
         return {"result": "success"}
 
 

api/controllers/console/workspace/workspace.py

@@ -29,7 +29,7 @@ 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
+from libs.helper import TimestampField, to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.account import Tenant, TenantCustomConfigDict, TenantStatus
 from services.account_service import TenantService
@@ -86,9 +86,7 @@ class TenantInfoResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None):
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 register_schema_models(

api/controllers/service_api/app/conversation.py

@@ -22,7 +22,7 @@ from fields.conversation_fields import (
     SimpleConversation,
 )
 from graphon.variables.types import SegmentType
-from libs.helper import UUIDStrOrEmpty
+from libs.helper import UUIDStrOrEmpty, to_timestamp
 from models.model import App, AppMode, EndUser
 from services.conversation_service import ConversationService
 
@@ -115,9 +115,7 @@ class ConversationVariableResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class ConversationVariableInfiniteScrollPaginationResponse(ResponseModel):

api/controllers/service_api/app/human_input_form.py

@@ -7,18 +7,18 @@ paused human input forms in workflow/chatflow runs.
 
 import json
 import logging
-from datetime import datetime
 
 from flask import Response
 from flask_restx import Resource
 from werkzeug.exceptions import BadRequest, NotFound
 
-from controllers.common.human_input import HumanInputFormSubmitPayload
+from controllers.common.human_input import HumanInputFormSubmitPayload, stringify_form_default_values
 from controllers.common.schema import register_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
 from core.workflow.human_input_policy import HumanInputSurface, is_recipient_type_allowed_for_surface
 from extensions.ext_database import db
+from libs.helper import to_timestamp
 from models.model import App, EndUser
 from services.human_input_service import Form, FormNotFoundError, HumanInputService
 
@@ -28,30 +28,14 @@ logger = logging.getLogger(__name__)
 register_schema_models(service_api_ns, HumanInputFormSubmitPayload)
 
 
-def _stringify_default_values(values: dict[str, object]) -> dict[str, str]:
-    result: dict[str, str] = {}
-    for key, value in values.items():
-        if value is None:
-            result[key] = ""
-        elif isinstance(value, (dict, list)):
-            result[key] = json.dumps(value, ensure_ascii=False)
-        else:
-            result[key] = str(value)
-    return result
-
-
-def _to_timestamp(value: datetime) -> int:
-    return int(value.timestamp())
-
-
 def _jsonify_form_definition(form: Form) -> Response:
     definition_payload = form.get_definition().model_dump()
     payload = {
         "form_content": definition_payload["rendered_content"],
         "inputs": definition_payload["inputs"],
-        "resolved_default_values": _stringify_default_values(definition_payload["default_values"]),
+        "resolved_default_values": stringify_form_default_values(definition_payload["default_values"]),
         "user_actions": definition_payload["user_actions"],
-        "expiration_time": _to_timestamp(form.expiration_time),
+        "expiration_time": to_timestamp(form.expiration_time),
     }
     return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")
 

api/controllers/service_api/app/workflow.py

@@ -39,6 +39,7 @@ from graphon.enums import WorkflowExecutionStatus
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
+from libs.helper import to_timestamp
 from models.model import App, AppMode, EndUser
 from models.workflow import WorkflowRun
 from repositories.factory import DifyAPIRepositoryFactory
@@ -68,12 +69,6 @@ class WorkflowLogQuery(BaseModel):
 register_schema_models(service_api_ns, WorkflowRunPayload, WorkflowLogQuery)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 def _enum_value(value):
     return getattr(value, "value", value)
 
@@ -109,7 +104,7 @@ class WorkflowRunResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowRunForLogResponse(ResponseModel):
@@ -133,7 +128,7 @@ class WorkflowRunForLogResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPartialResponse(ResponseModel):
@@ -154,7 +149,7 @@ class WorkflowAppLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPaginationResponse(ResponseModel):

api/controllers/service_api/dataset/dataset.py

@@ -31,7 +31,9 @@ from services.tag_service import (
     TagBindingCreatePayload,
     TagBindingDeletePayload,
     TagService,
-    UpdateTagPayload,
+)
+from services.tag_service import (
+    UpdateTagPayload as UpdateTagServicePayload,
 )
 
 register_enum_models(service_api_ns, DatasetPermissionEnum)
@@ -556,7 +558,7 @@ class DatasetTagsApi(DatasetApiResource):
 
         payload = TagUpdatePayload.model_validate(service_api_ns.payload or {})
         tag_id = payload.tag_id
-        tag = TagService.update_tags(UpdateTagPayload(name=payload.name, type=TagType.KNOWLEDGE), tag_id)
+        tag = TagService.update_tags(UpdateTagServicePayload(name=payload.name), tag_id)
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 

api/controllers/service_api/dataset/document.py

@@ -136,7 +136,7 @@ def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Mapping[
     if not dataset:
         raise ValueError("Dataset does not exist.")
 
-    if not dataset.indexing_technique and not args["indexing_technique"]:
+    if not dataset.indexing_technique and not args.get("indexing_technique"):
         raise ValueError("indexing_technique is required.")
 
     embedding_model_provider = payload.embedding_model_provider

api/controllers/web/human_input_form.py

@@ -4,7 +4,6 @@ Web App Human Input Form APIs.
 
 import json
 import logging
-from datetime import datetime
 from typing import Any, NotRequired, TypedDict
 
 from flask import Response, request
@@ -13,12 +12,12 @@ from sqlalchemy import select
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
-from controllers.common.human_input import HumanInputFormSubmitPayload
+from controllers.common.human_input import HumanInputFormSubmitPayload, stringify_form_default_values
 from controllers.web import web_ns
 from controllers.web.error import NotFoundError, WebFormRateLimitExceededError
 from controllers.web.site import serialize_app_site_payload
 from extensions.ext_database import db
-from libs.helper import RateLimiter, extract_remote_ip
+from libs.helper import RateLimiter, extract_remote_ip, to_timestamp
 from models.account import TenantStatus
 from models.model import App, Site
 from services.human_input_service import Form, FormNotFoundError, HumanInputService
@@ -38,22 +37,6 @@ _FORM_ACCESS_RATE_LIMITER = RateLimiter(
 )
 
 
-def _stringify_default_values(values: dict[str, object]) -> dict[str, str]:
-    result: dict[str, str] = {}
-    for key, value in values.items():
-        if value is None:
-            result[key] = ""
-        elif isinstance(value, (dict, list)):
-            result[key] = json.dumps(value, ensure_ascii=False)
-        else:
-            result[key] = str(value)
-    return result
-
-
-def _to_timestamp(value: datetime) -> int:
-    return int(value.timestamp())
-
-
 class FormDefinitionPayload(TypedDict):
     form_content: Any
     inputs: Any
@@ -69,9 +52,9 @@ def _jsonify_form_definition(form: Form, site_payload: dict | None = None) -> Re
     payload: FormDefinitionPayload = {
         "form_content": definition_payload["rendered_content"],
         "inputs": definition_payload["inputs"],
-        "resolved_default_values": _stringify_default_values(definition_payload["default_values"]),
+        "resolved_default_values": stringify_form_default_values(definition_payload["default_values"]),
         "user_actions": definition_payload["user_actions"],
-        "expiration_time": _to_timestamp(form.expiration_time),
+        "expiration_time": to_timestamp(form.expiration_time),
     }
     if site_payload is not None:
         payload["site"] = site_payload

api/core/agent/base_agent_runner.py

@@ -532,7 +532,6 @@ class BaseAgentRunner(AppRunner):
         file_objs = file_factory.build_from_message_files(
             message_files=files,
             tenant_id=self.tenant_id,
-            config=file_extra_config,
             access_controller=_file_access_controller,
         )
         if not file_objs:

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

@@ -253,7 +253,20 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
     ):
         """
         Resume a paused advanced chat execution.
+
+        ``trace_manager`` is transient and excluded from generate-entity serialization,
+        so resumed executions rebuild it here before persistence layers receive the entity.
         """
+        if application_generate_entity.trace_manager is None:
+            application_generate_entity = application_generate_entity.model_copy(
+                update={
+                    "trace_manager": TraceQueueManager(
+                        app_id=app_model.id,
+                        user_id=user.id if isinstance(user, Account) else user.session_id,
+                    )
+                }
+            )
+
         return self._generate(
             workflow=workflow,
             user=user,

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

@@ -9,7 +9,7 @@ from datetime import datetime
 from threading import Thread
 from typing import Any, Union
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.orm import Session, sessionmaker
 
 from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
@@ -245,49 +245,50 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         """
         human_input_responses: list[HumanInputRequiredResponse] = []
         for stream_response in generator:
-            if isinstance(stream_response, ErrorStreamResponse):
-                raise stream_response.err
-            elif isinstance(stream_response, HumanInputRequiredResponse):
-                human_input_responses.append(stream_response)
-            elif isinstance(stream_response, WorkflowPauseStreamResponse):
-                return AdvancedChatPausedBlockingResponse(
-                    task_id=stream_response.task_id,
-                    data=AdvancedChatPausedBlockingResponse.Data(
-                        id=self._message_id,
-                        mode=self._conversation_mode,
-                        conversation_id=self._conversation_id,
-                        message_id=self._message_id,
-                        workflow_run_id=stream_response.data.workflow_run_id,
-                        answer=self._task_state.answer,
-                        metadata=self._message_end_to_stream_response().metadata,
-                        created_at=self._message_created_at,
-                        paused_nodes=stream_response.data.paused_nodes,
-                        reasons=stream_response.data.reasons,
-                        status=stream_response.data.status,
-                        elapsed_time=stream_response.data.elapsed_time,
-                        total_tokens=stream_response.data.total_tokens,
-                        total_steps=stream_response.data.total_steps,
-                    ),
-                )
-            elif isinstance(stream_response, MessageEndStreamResponse):
-                extras = {}
-                if stream_response.metadata:
-                    extras["metadata"] = stream_response.metadata
+            match stream_response:
+                case ErrorStreamResponse():
+                    raise stream_response.err
+                case HumanInputRequiredResponse():
+                    human_input_responses.append(stream_response)
+                case WorkflowPauseStreamResponse():
+                    return AdvancedChatPausedBlockingResponse(
+                        task_id=stream_response.task_id,
+                        data=AdvancedChatPausedBlockingResponse.Data(
+                            id=self._message_id,
+                            mode=self._conversation_mode,
+                            conversation_id=self._conversation_id,
+                            message_id=self._message_id,
+                            workflow_run_id=stream_response.data.workflow_run_id,
+                            answer=self._task_state.answer,
+                            metadata=self._message_end_to_stream_response().metadata,
+                            created_at=self._message_created_at,
+                            paused_nodes=stream_response.data.paused_nodes,
+                            reasons=stream_response.data.reasons,
+                            status=stream_response.data.status,
+                            elapsed_time=stream_response.data.elapsed_time,
+                            total_tokens=stream_response.data.total_tokens,
+                            total_steps=stream_response.data.total_steps,
+                        ),
+                    )
+                case MessageEndStreamResponse():
+                    extras = {}
+                    if stream_response.metadata:
+                        extras["metadata"] = stream_response.metadata
 
-                return ChatbotAppBlockingResponse(
-                    task_id=stream_response.task_id,
-                    data=ChatbotAppBlockingResponse.Data(
-                        id=self._message_id,
-                        mode=self._conversation_mode,
-                        conversation_id=self._conversation_id,
-                        message_id=self._message_id,
-                        answer=self._task_state.answer,
-                        created_at=self._message_created_at,
-                        **extras,
-                    ),
-                )
-            else:
-                continue
+                    return ChatbotAppBlockingResponse(
+                        task_id=stream_response.task_id,
+                        data=ChatbotAppBlockingResponse.Data(
+                            id=self._message_id,
+                            mode=self._conversation_mode,
+                            conversation_id=self._conversation_id,
+                            message_id=self._message_id,
+                            answer=self._task_state.answer,
+                            created_at=self._message_created_at,
+                            **extras,
+                        ),
+                    )
+                case _:
+                    continue
 
         if human_input_responses:
             return self._build_paused_blocking_response_from_human_input(human_input_responses)
@@ -425,11 +426,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         self._workflow_run_id = run_id
 
         with self._database_session() as session:
-            message = self._get_message(session=session)
-            if not message:
-                raise ValueError(f"Message not found: {self._message_id}")
-
-            message.workflow_run_id = run_id
+            session.execute(update(Message).where(Message.id == self._message_id).values(workflow_run_id=run_id))
 
         workflow_start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
             task_id=self._application_generate_entity.task_id,

api/core/app/apps/agent_chat/app_config_manager.py

@@ -178,7 +178,7 @@ class AgentChatAppConfigManager(BaseAppConfigManager):
         if not isinstance(agent_mode, dict):
             raise ValueError("agent_mode must be of object type")
 
-        # FIXME(-LAN-): Cast needed due to basedpyright limitation with dict type narrowing
+        # FIXME(-LAN-): Cast needed because static checkers do not narrow this dict value.
         agent_mode = cast(dict[str, Any], agent_mode)
 
         if "enabled" not in agent_mode or not agent_mode["enabled"]:

api/core/app/apps/base_app_generator.py

@@ -195,22 +195,23 @@ class BaseAppGenerator:
             )
 
         if variable_entity.type == VariableEntityType.NUMBER:
-            if isinstance(value, (int, float)):
-                return value
-            elif isinstance(value, str):
-                # handle empty string case
-                if not value.strip():
-                    return None
-                # may raise ValueError if user_input_value is not a valid number
-                try:
-                    if "." in value:
-                        return float(value)
-                    else:
-                        return int(value)
-                except ValueError:
-                    raise ValueError(f"{variable_entity.variable} in input form must be a valid number")
-            else:
-                raise TypeError(f"expected value type int, float or str, got {type(value)}, value: {value}")
+            match value:
+                case int() | float():
+                    return value
+                case str():
+                    # handle empty string case
+                    if not value.strip():
+                        return None
+                    # may raise ValueError if user_input_value is not a valid number
+                    try:
+                        if "." in value:
+                            return float(value)
+                        else:
+                            return int(value)
+                    except ValueError:
+                        raise ValueError(f"{variable_entity.variable} in input form must be a valid number")
+                case _:
+                    raise TypeError(f"expected value type int, float or str, got {type(value)}, value: {value}")
 
         match variable_entity.type:
             case VariableEntityType.SELECT:
@@ -241,17 +242,18 @@ class BaseAppGenerator:
                         f"{variable_entity.variable} in input form must be less than {variable_entity.max_length} files"
                     )
             case VariableEntityType.CHECKBOX:
-                if isinstance(value, str):
-                    normalized_value = value.strip().lower()
-                    if normalized_value in {"true", "1", "yes", "on"}:
-                        value = True
-                    elif normalized_value in {"false", "0", "no", "off"}:
-                        value = False
-                elif isinstance(value, (int, float)):
-                    if value == 1:
-                        value = True
-                    elif value == 0:
-                        value = False
+                match value:
+                    case str():
+                        normalized_value = value.strip().lower()
+                        if normalized_value in {"true", "1", "yes", "on"}:
+                            value = True
+                        elif normalized_value in {"false", "0", "no", "off"}:
+                            value = False
+                    case int() | float():
+                        if value == 1:
+                            value = True
+                        elif value == 0:
+                            value = False
             case VariableEntityType.JSON_OBJECT:
                 if value and not isinstance(value, dict):
                     raise ValueError(f"{variable_entity.variable} in input form must be a dict")

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

@@ -45,20 +45,24 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter[Workflow
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            if isinstance(sub_stream_response, PingStreamResponse):
-                yield "ping"
-                continue
+            match sub_stream_response:
+                case PingStreamResponse():
+                    yield "ping"
+                    continue
+                case ErrorStreamResponse():
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    data = cls._error_to_stream_response(sub_stream_response.err)
+                    response_chunk.update(cast(dict, data))
+                case _:
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(sub_stream_response.model_dump())
 
-            response_chunk = {
-                "event": sub_stream_response.event.value,
-                "workflow_run_id": chunk.workflow_run_id,
-            }
-
-            if isinstance(sub_stream_response, ErrorStreamResponse):
-                data = cls._error_to_stream_response(sub_stream_response.err)
-                response_chunk.update(cast(dict, data))
-            else:
-                response_chunk.update(sub_stream_response.model_dump())
             yield response_chunk
 
     @classmethod
@@ -74,20 +78,28 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter[Workflow
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            if isinstance(sub_stream_response, PingStreamResponse):
-                yield "ping"
-                continue
+            match sub_stream_response:
+                case PingStreamResponse():
+                    yield "ping"
+                    continue
+                case ErrorStreamResponse():
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    data = cls._error_to_stream_response(sub_stream_response.err)
+                    response_chunk.update(cast(dict, data))
+                case NodeStartStreamResponse() | NodeFinishStreamResponse():
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(cast(dict, sub_stream_response.to_ignore_detail_dict()))
+                case _:
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(sub_stream_response.model_dump())
 
-            response_chunk = {
-                "event": sub_stream_response.event.value,
-                "workflow_run_id": chunk.workflow_run_id,
-            }
-
-            if isinstance(sub_stream_response, ErrorStreamResponse):
-                data = cls._error_to_stream_response(sub_stream_response.err)
-                response_chunk.update(cast(dict, data))
-            elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
-                response_chunk.update(cast(dict, sub_stream_response.to_ignore_detail_dict()))
-            else:
-                response_chunk.update(sub_stream_response.model_dump())
             yield response_chunk

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

@@ -32,7 +32,7 @@ from core.app.entities.task_entities import (
 )
 from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig, PauseStatePersistenceLayer
 from core.db.session_factory import session_factory
-from core.helper.trace_id_helper import extract_external_trace_id_from_args
+from core.helper.trace_id_helper import extract_external_trace_id_from_args, extract_parent_trace_context_from_args
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories import DifyCoreRepositoryFactory
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
@@ -166,6 +166,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
 
             extras = {
                 **extract_external_trace_id_from_args(args),
+                **extract_parent_trace_context_from_args(args),
             }
             workflow_run_id = str(workflow_run_id or uuid.uuid4())
             # FIXME (Yeuoly): we need to remove the SKIP_PREPARE_USER_INPUTS_KEY from the args
@@ -252,7 +253,20 @@ class WorkflowAppGenerator(BaseAppGenerator):
     ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
         """
         Resume a paused workflow execution using the persisted runtime state.
+
+        ``trace_manager`` is transient and excluded from generate-entity serialization,
+        so resumed executions rebuild it here before persistence layers receive the entity.
         """
+        if application_generate_entity.trace_manager is None:
+            application_generate_entity = application_generate_entity.model_copy(
+                update={
+                    "trace_manager": TraceQueueManager(
+                        app_id=app_model.id,
+                        user_id=user.id if isinstance(user, Account) else user.session_id,
+                    )
+                }
+            )
+
         return self._generate(
             app_model=app_model,
             workflow=workflow,

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

@@ -52,20 +52,24 @@ class WorkflowAppGenerateResponseConverter(
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            if isinstance(sub_stream_response, PingStreamResponse):
-                yield "ping"
-                continue
+            match sub_stream_response:
+                case PingStreamResponse():
+                    yield "ping"
+                    continue
+                case ErrorStreamResponse():
+                    response_chunk: dict[str, object] = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    data = cls._error_to_stream_response(sub_stream_response.err)
+                    response_chunk.update(data)
+                case _:
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(sub_stream_response.model_dump(mode="json"))
 
-            response_chunk: dict[str, object] = {
-                "event": sub_stream_response.event.value,
-                "workflow_run_id": chunk.workflow_run_id,
-            }
-
-            if isinstance(sub_stream_response, ErrorStreamResponse):
-                data = cls._error_to_stream_response(sub_stream_response.err)
-                response_chunk.update(data)
-            else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk
 
     @classmethod
@@ -81,20 +85,28 @@ class WorkflowAppGenerateResponseConverter(
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            if isinstance(sub_stream_response, PingStreamResponse):
-                yield "ping"
-                continue
+            match sub_stream_response:
+                case PingStreamResponse():
+                    yield "ping"
+                    continue
+                case ErrorStreamResponse():
+                    response_chunk: dict[str, object] = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    data = cls._error_to_stream_response(sub_stream_response.err)
+                    response_chunk.update(data)
+                case NodeStartStreamResponse() | NodeFinishStreamResponse():
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(sub_stream_response.to_ignore_detail_dict())
+                case _:
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(sub_stream_response.model_dump(mode="json"))
 
-            response_chunk: dict[str, object] = {
-                "event": sub_stream_response.event.value,
-                "workflow_run_id": chunk.workflow_run_id,
-            }
-
-            if isinstance(sub_stream_response, ErrorStreamResponse):
-                data = cls._error_to_stream_response(sub_stream_response.err)
-                response_chunk.update(data)
-            elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
-                response_chunk.update(sub_stream_response.to_ignore_detail_dict())
-            else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk

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

@@ -145,50 +145,51 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         """
         human_input_responses: list[HumanInputRequiredResponse] = []
         for stream_response in generator:
-            if isinstance(stream_response, ErrorStreamResponse):
-                raise stream_response.err
-            elif isinstance(stream_response, HumanInputRequiredResponse):
-                human_input_responses.append(stream_response)
-            elif isinstance(stream_response, WorkflowPauseStreamResponse):
-                return WorkflowAppPausedBlockingResponse(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_run_id=stream_response.data.workflow_run_id,
-                    data=WorkflowAppPausedBlockingResponse.Data(
-                        id=stream_response.data.workflow_run_id,
-                        workflow_id=self._workflow.id,
-                        status=stream_response.data.status,
-                        outputs=stream_response.data.outputs or {},
-                        error=None,
-                        elapsed_time=stream_response.data.elapsed_time,
-                        total_tokens=stream_response.data.total_tokens,
-                        total_steps=stream_response.data.total_steps,
-                        created_at=stream_response.data.created_at,
-                        finished_at=None,
-                        paused_nodes=stream_response.data.paused_nodes,
-                        reasons=stream_response.data.reasons,
-                    ),
-                )
-
-            elif isinstance(stream_response, WorkflowFinishStreamResponse):
-                return WorkflowAppBlockingResponse(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_run_id=stream_response.data.id,
-                    data=WorkflowAppBlockingResponse.Data(
-                        id=stream_response.data.id,
-                        workflow_id=stream_response.data.workflow_id,
-                        status=stream_response.data.status,
-                        outputs=stream_response.data.outputs,
-                        error=stream_response.data.error,
-                        elapsed_time=stream_response.data.elapsed_time,
-                        total_tokens=stream_response.data.total_tokens,
-                        total_steps=stream_response.data.total_steps,
-                        created_at=int(stream_response.data.created_at),
-                        finished_at=int(stream_response.data.finished_at) if stream_response.data.finished_at else None,
-                    ),
-                )
-
-            else:
-                continue
+            match stream_response:
+                case ErrorStreamResponse():
+                    raise stream_response.err
+                case HumanInputRequiredResponse():
+                    human_input_responses.append(stream_response)
+                case WorkflowPauseStreamResponse():
+                    return WorkflowAppPausedBlockingResponse(
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_run_id=stream_response.data.workflow_run_id,
+                        data=WorkflowAppPausedBlockingResponse.Data(
+                            id=stream_response.data.workflow_run_id,
+                            workflow_id=self._workflow.id,
+                            status=stream_response.data.status,
+                            outputs=stream_response.data.outputs or {},
+                            error=None,
+                            elapsed_time=stream_response.data.elapsed_time,
+                            total_tokens=stream_response.data.total_tokens,
+                            total_steps=stream_response.data.total_steps,
+                            created_at=stream_response.data.created_at,
+                            finished_at=None,
+                            paused_nodes=stream_response.data.paused_nodes,
+                            reasons=stream_response.data.reasons,
+                        ),
+                    )
+                case WorkflowFinishStreamResponse():
+                    return WorkflowAppBlockingResponse(
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_run_id=stream_response.data.id,
+                        data=WorkflowAppBlockingResponse.Data(
+                            id=stream_response.data.id,
+                            workflow_id=stream_response.data.workflow_id,
+                            status=stream_response.data.status,
+                            outputs=stream_response.data.outputs,
+                            error=stream_response.data.error,
+                            elapsed_time=stream_response.data.elapsed_time,
+                            total_tokens=stream_response.data.total_tokens,
+                            total_steps=stream_response.data.total_steps,
+                            created_at=int(stream_response.data.created_at),
+                            finished_at=int(stream_response.data.finished_at)
+                            if stream_response.data.finished_at
+                            else None,
+                        ),
+                    )
+                case _:
+                    continue
 
         if human_input_responses:
             return self._build_paused_blocking_response_from_human_input(human_input_responses)

api/core/app/apps/workflow_app_runner.py

@@ -399,278 +399,281 @@ class WorkflowBasedAppRunner:
         :param workflow_entry: workflow entry
         :param event: event
         """
-        if isinstance(event, GraphRunStartedEvent):
-            self._publish_event(QueueWorkflowStartedEvent(reason=event.reason))
-        elif isinstance(event, GraphRunSucceededEvent):
-            self._publish_event(QueueWorkflowSucceededEvent(outputs=event.outputs))
-        elif isinstance(event, GraphRunPartialSucceededEvent):
-            self._publish_event(
-                QueueWorkflowPartialSuccessEvent(outputs=event.outputs, exceptions_count=event.exceptions_count)
-            )
-        elif isinstance(event, GraphRunFailedEvent):
-            self._publish_event(QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count))
-        elif isinstance(event, GraphRunAbortedEvent):
-            self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0))
-        elif isinstance(event, GraphRunPausedEvent):
-            runtime_state = workflow_entry.graph_engine.graph_runtime_state
-            paused_nodes = runtime_state.get_paused_nodes()
-            self._enqueue_human_input_notifications(event.reasons)
-            self._publish_event(
-                QueueWorkflowPausedEvent(
-                    reasons=event.reasons,
-                    outputs=event.outputs,
-                    paused_nodes=paused_nodes,
+        match event:
+            case GraphRunStartedEvent():
+                self._publish_event(QueueWorkflowStartedEvent(reason=event.reason))
+            case GraphRunSucceededEvent():
+                self._publish_event(QueueWorkflowSucceededEvent(outputs=event.outputs))
+            case GraphRunPartialSucceededEvent():
+                self._publish_event(
+                    QueueWorkflowPartialSuccessEvent(outputs=event.outputs, exceptions_count=event.exceptions_count)
                 )
-            )
-        elif isinstance(event, NodeRunHumanInputFormFilledEvent):
-            self._publish_event(
-                QueueHumanInputFormFilledEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    rendered_content=event.rendered_content,
-                    action_id=event.action_id,
-                    action_text=event.action_text,
+            case GraphRunFailedEvent():
+                self._publish_event(
+                    QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count)
                 )
-            )
-        elif isinstance(event, NodeRunHumanInputFormTimeoutEvent):
-            self._publish_event(
-                QueueHumanInputFormTimeoutEvent(
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    expiration_time=event.expiration_time,
+            case GraphRunAbortedEvent():
+                self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0))
+            case GraphRunPausedEvent():
+                runtime_state = workflow_entry.graph_engine.graph_runtime_state
+                paused_nodes = runtime_state.get_paused_nodes()
+                self._enqueue_human_input_notifications(event.reasons)
+                self._publish_event(
+                    QueueWorkflowPausedEvent(
+                        reasons=event.reasons,
+                        outputs=event.outputs,
+                        paused_nodes=paused_nodes,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunRetryEvent):
-            node_run_result = event.node_run_result
-            inputs = node_run_result.inputs
-            process_data = node_run_result.process_data
-            outputs = project_node_outputs_for_workflow_run(
-                node_type=event.node_type,
-                inputs=inputs,
-                outputs=node_run_result.outputs,
-            )
-            execution_metadata = node_run_result.metadata
-            self._publish_event(
-                QueueNodeRetryEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_title=event.node_title,
+            case NodeRunHumanInputFormFilledEvent():
+                self._publish_event(
+                    QueueHumanInputFormFilledEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        rendered_content=event.rendered_content,
+                        action_id=event.action_id,
+                        action_text=event.action_text,
+                    )
+                )
+            case NodeRunHumanInputFormTimeoutEvent():
+                self._publish_event(
+                    QueueHumanInputFormTimeoutEvent(
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        expiration_time=event.expiration_time,
+                    )
+                )
+            case NodeRunRetryEvent():
+                node_run_result = event.node_run_result
+                inputs = node_run_result.inputs
+                process_data = node_run_result.process_data
+                outputs = project_node_outputs_for_workflow_run(
                     node_type=event.node_type,
-                    start_at=event.start_at,
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
                     inputs=inputs,
-                    process_data=process_data,
-                    outputs=outputs,
-                    error=event.error,
-                    execution_metadata=execution_metadata,
-                    retry_index=event.retry_index,
-                    provider_type=event.provider_type,
-                    provider_id=event.provider_id,
+                    outputs=node_run_result.outputs,
                 )
-            )
-        elif isinstance(event, NodeRunStartedEvent):
-            self._publish_event(
-                QueueNodeStartedEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_title=event.node_title,
-                    node_type=event.node_type,
-                    start_at=event.start_at,
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
-                    agent_strategy=self._build_agent_strategy_info(event),
-                    provider_type=event.provider_type,
-                    provider_id=event.provider_id,
+                execution_metadata = node_run_result.metadata
+                self._publish_event(
+                    QueueNodeRetryEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_title=event.node_title,
+                        node_type=event.node_type,
+                        start_at=event.start_at,
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                        inputs=inputs,
+                        process_data=process_data,
+                        outputs=outputs,
+                        error=event.error,
+                        execution_metadata=execution_metadata,
+                        retry_index=event.retry_index,
+                        provider_type=event.provider_type,
+                        provider_id=event.provider_id,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunSucceededEvent):
-            node_run_result = event.node_run_result
-            inputs = node_run_result.inputs
-            process_data = node_run_result.process_data
-            outputs = project_node_outputs_for_workflow_run(
-                node_type=event.node_type,
-                inputs=inputs,
-                outputs=node_run_result.outputs,
-            )
-            execution_metadata = node_run_result.metadata
-            self._publish_event(
-                QueueNodeSucceededEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
+            case NodeRunStartedEvent():
+                self._publish_event(
+                    QueueNodeStartedEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_title=event.node_title,
+                        node_type=event.node_type,
+                        start_at=event.start_at,
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                        agent_strategy=self._build_agent_strategy_info(event),
+                        provider_type=event.provider_type,
+                        provider_id=event.provider_id,
+                    )
+                )
+            case NodeRunSucceededEvent():
+                node_run_result = event.node_run_result
+                inputs = node_run_result.inputs
+                process_data = node_run_result.process_data
+                outputs = project_node_outputs_for_workflow_run(
                     node_type=event.node_type,
-                    start_at=event.start_at,
-                    finished_at=event.finished_at,
                     inputs=inputs,
-                    process_data=process_data,
-                    outputs=outputs,
-                    execution_metadata=execution_metadata,
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
+                    outputs=node_run_result.outputs,
                 )
-            )
-        elif isinstance(event, NodeRunFailedEvent):
-            outputs = project_node_outputs_for_workflow_run(
-                node_type=event.node_type,
-                inputs=event.node_run_result.inputs,
-                outputs=event.node_run_result.outputs,
-            )
-            self._publish_event(
-                QueueNodeFailedEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
+                execution_metadata = node_run_result.metadata
+                self._publish_event(
+                    QueueNodeSucceededEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        start_at=event.start_at,
+                        finished_at=event.finished_at,
+                        inputs=inputs,
+                        process_data=process_data,
+                        outputs=outputs,
+                        execution_metadata=execution_metadata,
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                    )
+                )
+            case NodeRunFailedEvent():
+                outputs = project_node_outputs_for_workflow_run(
                     node_type=event.node_type,
-                    start_at=event.start_at,
-                    finished_at=event.finished_at,
                     inputs=event.node_run_result.inputs,
-                    process_data=event.node_run_result.process_data,
-                    outputs=outputs,
-                    error=event.node_run_result.error or "Unknown error",
-                    execution_metadata=event.node_run_result.metadata,
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
+                    outputs=event.node_run_result.outputs,
                 )
-            )
-        elif isinstance(event, NodeRunExceptionEvent):
-            outputs = project_node_outputs_for_workflow_run(
-                node_type=event.node_type,
-                inputs=event.node_run_result.inputs,
-                outputs=event.node_run_result.outputs,
-            )
-            self._publish_event(
-                QueueNodeExceptionEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
+                self._publish_event(
+                    QueueNodeFailedEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        start_at=event.start_at,
+                        finished_at=event.finished_at,
+                        inputs=event.node_run_result.inputs,
+                        process_data=event.node_run_result.process_data,
+                        outputs=outputs,
+                        error=event.node_run_result.error or "Unknown error",
+                        execution_metadata=event.node_run_result.metadata,
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                    )
+                )
+            case NodeRunExceptionEvent():
+                outputs = project_node_outputs_for_workflow_run(
                     node_type=event.node_type,
-                    start_at=event.start_at,
-                    finished_at=event.finished_at,
                     inputs=event.node_run_result.inputs,
-                    process_data=event.node_run_result.process_data,
-                    outputs=outputs,
-                    error=event.node_run_result.error or "Unknown error",
-                    execution_metadata=event.node_run_result.metadata,
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
+                    outputs=event.node_run_result.outputs,
                 )
-            )
-        elif isinstance(event, NodeRunStreamChunkEvent):
-            self._publish_event(
-                QueueTextChunkEvent(
-                    text=event.chunk,
-                    from_variable_selector=list(event.selector),
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
+                self._publish_event(
+                    QueueNodeExceptionEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        start_at=event.start_at,
+                        finished_at=event.finished_at,
+                        inputs=event.node_run_result.inputs,
+                        process_data=event.node_run_result.process_data,
+                        outputs=outputs,
+                        error=event.node_run_result.error or "Unknown error",
+                        execution_metadata=event.node_run_result.metadata,
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunRetrieverResourceEvent):
-            self._publish_event(
-                QueueRetrieverResourcesEvent(
-                    retriever_resources=[
-                        RetrievalSourceMetadata.model_validate(resource) for resource in event.retriever_resources
-                    ],
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
+            case NodeRunStreamChunkEvent():
+                self._publish_event(
+                    QueueTextChunkEvent(
+                        text=event.chunk,
+                        from_variable_selector=list(event.selector),
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunAgentLogEvent):
-            self._publish_event(
-                QueueAgentLogEvent(
-                    id=event.message_id,
-                    label=event.label,
-                    node_execution_id=event.node_execution_id,
-                    parent_id=event.parent_id,
-                    error=event.error,
-                    status=event.status,
-                    data=event.data,
-                    metadata=event.metadata,
-                    node_id=event.node_id,
+            case NodeRunRetrieverResourceEvent():
+                self._publish_event(
+                    QueueRetrieverResourcesEvent(
+                        retriever_resources=[
+                            RetrievalSourceMetadata.model_validate(resource) for resource in event.retriever_resources
+                        ],
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunIterationStartedEvent):
-            self._publish_event(
-                QueueIterationStartEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    start_at=event.start_at,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    inputs=event.inputs,
-                    metadata=event.metadata,
+            case NodeRunAgentLogEvent():
+                self._publish_event(
+                    QueueAgentLogEvent(
+                        id=event.message_id,
+                        label=event.label,
+                        node_execution_id=event.node_execution_id,
+                        parent_id=event.parent_id,
+                        error=event.error,
+                        status=event.status,
+                        data=event.data,
+                        metadata=event.metadata,
+                        node_id=event.node_id,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunIterationNextEvent):
-            self._publish_event(
-                QueueIterationNextEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    index=event.index,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    output=event.pre_iteration_output,
+            case NodeRunIterationStartedEvent():
+                self._publish_event(
+                    QueueIterationStartEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        start_at=event.start_at,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        inputs=event.inputs,
+                        metadata=event.metadata,
+                    )
                 )
-            )
-        elif isinstance(event, (NodeRunIterationSucceededEvent | NodeRunIterationFailedEvent)):
-            self._publish_event(
-                QueueIterationCompletedEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    start_at=event.start_at,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    inputs=event.inputs,
-                    outputs=event.outputs,
-                    metadata=event.metadata,
-                    steps=event.steps,
-                    error=event.error if isinstance(event, NodeRunIterationFailedEvent) else None,
+            case NodeRunIterationNextEvent():
+                self._publish_event(
+                    QueueIterationNextEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        index=event.index,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        output=event.pre_iteration_output,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunLoopStartedEvent):
-            self._publish_event(
-                QueueLoopStartEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    start_at=event.start_at,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    inputs=event.inputs,
-                    metadata=event.metadata,
+            case NodeRunIterationSucceededEvent() | NodeRunIterationFailedEvent():
+                self._publish_event(
+                    QueueIterationCompletedEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        start_at=event.start_at,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        inputs=event.inputs,
+                        outputs=event.outputs,
+                        metadata=event.metadata,
+                        steps=event.steps,
+                        error=event.error if isinstance(event, NodeRunIterationFailedEvent) else None,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunLoopNextEvent):
-            self._publish_event(
-                QueueLoopNextEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    index=event.index,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    output=event.pre_loop_output,
+            case NodeRunLoopStartedEvent():
+                self._publish_event(
+                    QueueLoopStartEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        start_at=event.start_at,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        inputs=event.inputs,
+                        metadata=event.metadata,
+                    )
                 )
-            )
-        elif isinstance(event, (NodeRunLoopSucceededEvent | NodeRunLoopFailedEvent)):
-            self._publish_event(
-                QueueLoopCompletedEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    start_at=event.start_at,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    inputs=event.inputs,
-                    outputs=event.outputs,
-                    metadata=event.metadata,
-                    steps=event.steps,
-                    error=event.error if isinstance(event, NodeRunLoopFailedEvent) else None,
+            case NodeRunLoopNextEvent():
+                self._publish_event(
+                    QueueLoopNextEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        index=event.index,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        output=event.pre_loop_output,
+                    )
+                )
+            case NodeRunLoopSucceededEvent() | NodeRunLoopFailedEvent():
+                self._publish_event(
+                    QueueLoopCompletedEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        start_at=event.start_at,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        inputs=event.inputs,
+                        outputs=event.outputs,
+                        metadata=event.metadata,
+                        steps=event.steps,
+                        error=event.error if isinstance(event, NodeRunLoopFailedEvent) else None,
+                    )
                 )
-            )
 
     def _enqueue_human_input_notifications(self, reasons: Sequence[object]) -> None:
         for reason in reasons:

api/core/app/entities/task_entities.py

@@ -10,7 +10,7 @@ from graphon.entities import WorkflowStartReason
 from graphon.entities.pause_reason import PauseReasonType
 from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMUsage
-from graphon.nodes.human_input.entities import FormInput, UserAction
+from graphon.nodes.human_input.entities import FormInputConfig, UserActionConfig
 
 
 class AnnotationReplyAccount(BaseModel):
@@ -284,8 +284,8 @@ class HumanInputRequiredResponse(StreamResponse):
         node_id: str
         node_title: str
         form_content: str
-        inputs: Sequence[FormInput] = Field(default_factory=list)
-        actions: Sequence[UserAction] = Field(default_factory=list)
+        inputs: Sequence[FormInputConfig] = Field(default_factory=list)
+        actions: Sequence[UserActionConfig] = Field(default_factory=list)
         display_in_ui: bool = False
         form_token: str | None = None
         resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)
@@ -307,8 +307,8 @@ class HumanInputRequiredPauseReasonPayload(BaseModel):
     node_id: str
     node_title: str
     form_content: str
-    inputs: Sequence[FormInput] = Field(default_factory=list)
-    actions: Sequence[UserAction] = Field(default_factory=list)
+    inputs: Sequence[FormInputConfig] = Field(default_factory=list)
+    actions: Sequence[UserActionConfig] = Field(default_factory=list)
     display_in_ui: bool = False
     form_token: str | None = None
     resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)

api/core/app/file_access/__init__.py

@@ -1,6 +1,13 @@
 from .controller import DatabaseFileAccessController
 from .protocols import FileAccessControllerProtocol
-from .scope import FileAccessScope, bind_file_access_scope, get_current_file_access_scope
+from .scope import (
+    FileAccessScope,
+    bind_file_access_scope,
+    get_current_file_access_scope,
+    grant_retriever_segment_access,
+    grant_upload_file_access,
+    is_retriever_segment_access_granted,
+)
 
 __all__ = [
     "DatabaseFileAccessController",
@@ -8,4 +15,7 @@ __all__ = [
     "FileAccessScope",
     "bind_file_access_scope",
     "get_current_file_access_scope",
+    "grant_retriever_segment_access",
+    "grant_upload_file_access",
+    "is_retriever_segment_access_granted",
 ]

api/core/app/file_access/controller.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Callable
 
-from sqlalchemy import select
+from sqlalchemy import and_, or_, select
 from sqlalchemy.orm import Session
 from sqlalchemy.sql import Select
 
@@ -18,7 +18,8 @@ class DatabaseFileAccessController(FileAccessControllerProtocol):
 
     Tenant scoping remains mandatory. When the current execution belongs to an
     end user, the lookup is additionally constrained to that end user's file
-    ownership markers.
+    ownership markers, plus upload files explicitly granted by the current
+    execution context.
     """
 
     _scope_getter: Callable[[], FileAccessScope | None]
@@ -47,10 +48,19 @@ class DatabaseFileAccessController(FileAccessControllerProtocol):
         if not resolved_scope.requires_user_ownership:
             return scoped_stmt
 
-        return scoped_stmt.where(
+        user_owned_filter = and_(
             UploadFile.created_by_role == CreatorUserRole.END_USER,
             UploadFile.created_by == resolved_scope.user_id,
         )
+        if not resolved_scope.granted_upload_file_ids:
+            return scoped_stmt.where(user_owned_filter)
+
+        return scoped_stmt.where(
+            or_(
+                user_owned_filter,
+                UploadFile.id.in_(resolved_scope.granted_upload_file_ids),
+            )
+        )
 
     def apply_tool_file_filters(
         self,

api/core/app/file_access/scope.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
-from collections.abc import Generator  # Changed from Iterator
+from collections.abc import Generator, Iterable
 from contextlib import contextmanager
 from contextvars import ContextVar
-from dataclasses import dataclass
+from dataclasses import dataclass, field, replace
 
 from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
 
@@ -15,12 +15,23 @@ _current_file_access_scope: ContextVar[FileAccessScope | None] = ContextVar(
 
 @dataclass(frozen=True, slots=True)
 class FileAccessScope:
-    """Request-scoped ownership context used by workflow-layer file lookups."""
+    """Request-scoped ownership context used by workflow-layer file lookups.
+
+    ``granted_upload_file_ids`` is execution-local: callers may add upload files
+    that were returned by trusted retrieval paths without changing persistent
+    ownership markers.
+
+    ``granted_retriever_segment_ids`` gates lazy attachment loading by segment
+    ID, so user-provided context cannot make a later LLM node load arbitrary
+    same-tenant knowledge attachments.
+    """
 
     tenant_id: str
     user_id: str
     user_from: UserFrom
     invoke_from: InvokeFrom
+    granted_upload_file_ids: frozenset[str] = field(default_factory=frozenset)
+    granted_retriever_segment_ids: frozenset[str] = field(default_factory=frozenset)
 
     @property
     def requires_user_ownership(self) -> bool:
@@ -31,8 +42,49 @@ def get_current_file_access_scope() -> FileAccessScope | None:
     return _current_file_access_scope.get()
 
 
+def grant_upload_file_access(upload_file_ids: Iterable[str]) -> None:
+    scope = _current_file_access_scope.get()
+    if scope is None:
+        return
+
+    granted_upload_file_ids = frozenset(str(file_id) for file_id in upload_file_ids if file_id)
+    if not granted_upload_file_ids:
+        return
+
+    _current_file_access_scope.set(
+        replace(
+            scope,
+            granted_upload_file_ids=scope.granted_upload_file_ids | granted_upload_file_ids,
+        )
+    )
+
+
+def grant_retriever_segment_access(segment_ids: Iterable[str]) -> None:
+    scope = _current_file_access_scope.get()
+    if scope is None:
+        return
+
+    granted_segment_ids = frozenset(str(segment_id) for segment_id in segment_ids if segment_id)
+    if not granted_segment_ids:
+        return
+
+    _current_file_access_scope.set(
+        replace(
+            scope,
+            granted_retriever_segment_ids=scope.granted_retriever_segment_ids | granted_segment_ids,
+        )
+    )
+
+
+def is_retriever_segment_access_granted(segment_id: str) -> bool:
+    scope = _current_file_access_scope.get()
+    if scope is None or not scope.requires_user_ownership:
+        return True
+    return str(segment_id) in scope.granted_retriever_segment_ids
+
+
 @contextmanager
-def bind_file_access_scope(scope: FileAccessScope) -> Generator[None, None, None]:  # Changed from Iterator[None]
+def bind_file_access_scope(scope: FileAccessScope) -> Generator[None, None, None]:
     token = _current_file_access_scope.set(scope)
     try:
         yield

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -140,42 +140,43 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         :return:
         """
         for stream_response in generator:
-            if isinstance(stream_response, ErrorStreamResponse):
-                raise stream_response.err
-            elif isinstance(stream_response, MessageEndStreamResponse):
-                extras = {"usage": self._task_state.llm_result.usage.model_dump()}
-                if self._task_state.metadata:
-                    extras["metadata"] = self._task_state.metadata.model_dump()
-                response: ChatbotAppBlockingResponse | CompletionAppBlockingResponse
-                if self._conversation_mode == AppMode.COMPLETION:
-                    response = CompletionAppBlockingResponse(
-                        task_id=self._application_generate_entity.task_id,
-                        data=CompletionAppBlockingResponse.Data(
-                            id=self._message_id,
-                            mode=self._conversation_mode,
-                            message_id=self._message_id,
-                            answer=self._task_state.llm_result.message.get_text_content(),
-                            created_at=self._message_created_at,
-                            **extras,
-                        ),
-                    )
-                else:
-                    response = ChatbotAppBlockingResponse(
-                        task_id=self._application_generate_entity.task_id,
-                        data=ChatbotAppBlockingResponse.Data(
-                            id=self._message_id,
-                            mode=self._conversation_mode,
-                            conversation_id=self._conversation_id,
-                            message_id=self._message_id,
-                            answer=self._task_state.llm_result.message.get_text_content(),
-                            created_at=self._message_created_at,
-                            **extras,
-                        ),
-                    )
+            match stream_response:
+                case ErrorStreamResponse():
+                    raise stream_response.err
+                case MessageEndStreamResponse():
+                    extras = {"usage": self._task_state.llm_result.usage.model_dump()}
+                    if self._task_state.metadata:
+                        extras["metadata"] = self._task_state.metadata.model_dump()
+                    response: ChatbotAppBlockingResponse | CompletionAppBlockingResponse
+                    if self._conversation_mode == AppMode.COMPLETION:
+                        response = CompletionAppBlockingResponse(
+                            task_id=self._application_generate_entity.task_id,
+                            data=CompletionAppBlockingResponse.Data(
+                                id=self._message_id,
+                                mode=self._conversation_mode,
+                                message_id=self._message_id,
+                                answer=self._task_state.llm_result.message.get_text_content(),
+                                created_at=self._message_created_at,
+                                **extras,
+                            ),
+                        )
+                    else:
+                        response = ChatbotAppBlockingResponse(
+                            task_id=self._application_generate_entity.task_id,
+                            data=ChatbotAppBlockingResponse.Data(
+                                id=self._message_id,
+                                mode=self._conversation_mode,
+                                conversation_id=self._conversation_id,
+                                message_id=self._message_id,
+                                answer=self._task_state.llm_result.message.get_text_content(),
+                                created_at=self._message_created_at,
+                                **extras,
+                            ),
+                        )
 
-                return response
-            else:
-                continue
+                    return response
+                case _:
+                    continue
 
         raise RuntimeError("queue listening stopped unexpectedly.")
 
@@ -265,104 +266,107 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
                 publisher.publish(message)
             event = message.event
 
-            if isinstance(event, QueueErrorEvent):
-                with sessionmaker(bind=db.engine).begin() as session:
-                    err = self.handle_error(event=event, session=session, message_id=self._message_id)
-                yield self.error_to_stream_response(err)
-                break
-            elif isinstance(event, QueueStopEvent | QueueMessageEndEvent):
-                if isinstance(event, QueueMessageEndEvent):
-                    if event.llm_result:
-                        self._task_state.llm_result = event.llm_result
-                else:
-                    self._handle_stop(event)
+            match event:
+                case QueueErrorEvent():
+                    with sessionmaker(bind=db.engine).begin() as session:
+                        err = self.handle_error(event=event, session=session, message_id=self._message_id)
+                    yield self.error_to_stream_response(err)
+                    break
+                case QueueStopEvent() | QueueMessageEndEvent():
+                    if isinstance(event, QueueMessageEndEvent):
+                        if event.llm_result:
+                            self._task_state.llm_result = event.llm_result
+                    else:
+                        self._handle_stop(event)
 
-                # handle output moderation
-                output_moderation_answer = self.handle_output_moderation_when_task_finished(
-                    self._task_state.llm_result.message.get_text_content()
-                )
-                if output_moderation_answer:
-                    self._task_state.llm_result.message.content = output_moderation_answer
-                    yield self._message_cycle_manager.message_replace_to_stream_response(
-                        answer=output_moderation_answer
+                    # handle output moderation
+                    output_moderation_answer = self.handle_output_moderation_when_task_finished(
+                        self._task_state.llm_result.message.get_text_content()
                     )
-
-                with sessionmaker(bind=db.engine).begin() as session:
-                    # Save message
-                    self._save_message(session=session, trace_manager=trace_manager)
-                message_end_resp = self._message_end_to_stream_response()
-                yield message_end_resp
-            elif isinstance(event, QueueRetrieverResourcesEvent):
-                self._message_cycle_manager.handle_retriever_resources(event)
-            elif isinstance(event, QueueAnnotationReplyEvent):
-                annotation = self._message_cycle_manager.handle_annotation_reply(event)
-                if annotation:
-                    self._task_state.llm_result.message.content = annotation.content
-            elif isinstance(event, QueueAgentThoughtEvent):
-                agent_thought_response = self._agent_thought_to_stream_response(event)
-                if agent_thought_response is not None:
-                    yield agent_thought_response
-            elif isinstance(event, QueueMessageFileEvent):
-                response = self._message_cycle_manager.message_file_to_stream_response(event)
-                if response:
-                    yield response
-            elif isinstance(event, QueueLLMChunkEvent | QueueAgentMessageEvent):
-                chunk = event.chunk
-                delta_text = chunk.delta.message.content
-                if delta_text is None:
-                    continue
-                if isinstance(chunk.delta.message.content, list):
-                    delta_text = ""
-                    for content in chunk.delta.message.content:
-                        logger.debug(
-                            "The content type %s in LLM chunk delta message content.: %r", type(content), content
+                    if output_moderation_answer:
+                        self._task_state.llm_result.message.content = output_moderation_answer
+                        yield self._message_cycle_manager.message_replace_to_stream_response(
+                            answer=output_moderation_answer
                         )
-                        if isinstance(content, TextPromptMessageContent):
-                            delta_text += content.data
-                        elif isinstance(content, str):
-                            delta_text += content  # failback to str
-                        else:
-                            logger.warning(
-                                "Unsupported content type %s in LLM chunk delta message content.: %r",
-                                type(content),
-                                content,
+
+                    with sessionmaker(bind=db.engine).begin() as session:
+                        # Save message
+                        self._save_message(session=session, trace_manager=trace_manager)
+                    message_end_resp = self._message_end_to_stream_response()
+                    yield message_end_resp
+                case QueueRetrieverResourcesEvent():
+                    self._message_cycle_manager.handle_retriever_resources(event)
+                case QueueAnnotationReplyEvent():
+                    annotation = self._message_cycle_manager.handle_annotation_reply(event)
+                    if annotation:
+                        self._task_state.llm_result.message.content = annotation.content
+                case QueueAgentThoughtEvent():
+                    agent_thought_response = self._agent_thought_to_stream_response(event)
+                    if agent_thought_response is not None:
+                        yield agent_thought_response
+                case QueueMessageFileEvent():
+                    response = self._message_cycle_manager.message_file_to_stream_response(event)
+                    if response:
+                        yield response
+                case QueueLLMChunkEvent() | QueueAgentMessageEvent():
+                    chunk = event.chunk
+                    delta_text = chunk.delta.message.content
+                    if delta_text is None:
+                        continue
+                    if isinstance(chunk.delta.message.content, list):
+                        delta_text = ""
+                        for content in chunk.delta.message.content:
+                            logger.debug(
+                                "The content type %s in LLM chunk delta message content.: %r", type(content), content
                             )
-                            continue
+                            match content:
+                                case TextPromptMessageContent():
+                                    delta_text += content.data
+                                case str():
+                                    delta_text += content  # failback to str
+                                case _:
+                                    logger.warning(
+                                        "Unsupported content type %s in LLM chunk delta message content.: %r",
+                                        type(content),
+                                        content,
+                                    )
+                                    continue
 
-                if not self._task_state.llm_result.prompt_messages:
-                    self._task_state.llm_result.prompt_messages = chunk.prompt_messages
+                    if not self._task_state.llm_result.prompt_messages:
+                        self._task_state.llm_result.prompt_messages = chunk.prompt_messages
 
-                # handle output moderation chunk
-                should_direct_answer = self._handle_output_moderation_chunk(cast(str, delta_text))
-                if should_direct_answer:
+                    # handle output moderation chunk
+                    should_direct_answer = self._handle_output_moderation_chunk(cast(str, delta_text))
+                    if should_direct_answer:
+                        continue
+
+                    current_content = cast(str, self._task_state.llm_result.message.content)
+                    current_content += cast(str, delta_text)
+                    self._task_state.llm_result.message.content = current_content
+
+                    match event:
+                        case QueueLLMChunkEvent():
+                            # Determine the event type once, on first LLM chunk, and reuse for subsequent chunks
+                            if not hasattr(self, "_precomputed_event_type") or self._precomputed_event_type is None:
+                                self._precomputed_event_type = self._message_cycle_manager.get_message_event_type(
+                                    message_id=self._message_id
+                                )
+                            yield self._message_cycle_manager.message_to_stream_response(
+                                answer=cast(str, delta_text),
+                                message_id=self._message_id,
+                                event_type=self._precomputed_event_type,
+                            )
+                        case _:
+                            yield self._agent_message_to_stream_response(
+                                answer=cast(str, delta_text),
+                                message_id=self._message_id,
+                            )
+                case QueueMessageReplaceEvent():
+                    yield self._message_cycle_manager.message_replace_to_stream_response(answer=event.text)
+                case QueuePingEvent():
+                    yield self.ping_stream_response()
+                case _:
                     continue
-
-                current_content = cast(str, self._task_state.llm_result.message.content)
-                current_content += cast(str, delta_text)
-                self._task_state.llm_result.message.content = current_content
-
-                if isinstance(event, QueueLLMChunkEvent):
-                    # Determine the event type once, on first LLM chunk, and reuse for subsequent chunks
-                    if not hasattr(self, "_precomputed_event_type") or self._precomputed_event_type is None:
-                        self._precomputed_event_type = self._message_cycle_manager.get_message_event_type(
-                            message_id=self._message_id
-                        )
-                    yield self._message_cycle_manager.message_to_stream_response(
-                        answer=cast(str, delta_text),
-                        message_id=self._message_id,
-                        event_type=self._precomputed_event_type,
-                    )
-                else:
-                    yield self._agent_message_to_stream_response(
-                        answer=cast(str, delta_text),
-                        message_id=self._message_id,
-                    )
-            elif isinstance(event, QueueMessageReplaceEvent):
-                yield self._message_cycle_manager.message_replace_to_stream_response(answer=event.text)
-            elif isinstance(event, QueuePingEvent):
-                yield self.ping_stream_response()
-            else:
-                continue
         if publisher:
             publisher.publish(None)
         if self._conversation_name_generate_thread:

api/core/app/workflow/file_runtime.py

@@ -128,7 +128,7 @@ class DifyWorkflowFileRuntime(WorkflowFileRuntimeProtocol):
 
     @staticmethod
     def _secret_key() -> bytes:
-        return dify_config.SECRET_KEY.encode() if dify_config.SECRET_KEY else b""
+        return dify_config.SECRET_KEY.encode()
 
     def _sign_query(self, *, payload: str) -> dict[str, str]:
         timestamp = str(int(time.time()))

api/core/app/workflow/layers/persistence.py

@@ -15,6 +15,7 @@ from datetime import datetime
 from typing import Any, Union
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
+from core.helper.trace_id_helper import ParentTraceContext
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
@@ -104,52 +105,31 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         self._node_sequence = 0
 
     def on_event(self, event: GraphEngineEvent) -> None:
-        if isinstance(event, GraphRunStartedEvent):
-            self._handle_graph_run_started()
-            return
-
-        if isinstance(event, GraphRunSucceededEvent):
-            self._handle_graph_run_succeeded(event)
-            return
-
-        if isinstance(event, GraphRunPartialSucceededEvent):
-            self._handle_graph_run_partial_succeeded(event)
-            return
-
-        if isinstance(event, GraphRunFailedEvent):
-            self._handle_graph_run_failed(event)
-            return
-
-        if isinstance(event, GraphRunAbortedEvent):
-            self._handle_graph_run_aborted(event)
-            return
-
-        if isinstance(event, GraphRunPausedEvent):
-            self._handle_graph_run_paused(event)
-            return
-
-        if isinstance(event, NodeRunRetryEvent):
-            self._handle_node_retry(event)
-            return
-
-        if isinstance(event, NodeRunStartedEvent):
-            self._handle_node_started(event)
-            return
-
-        if isinstance(event, NodeRunSucceededEvent):
-            self._handle_node_succeeded(event)
-            return
-
-        if isinstance(event, NodeRunFailedEvent):
-            self._handle_node_failed(event)
-            return
-
-        if isinstance(event, NodeRunExceptionEvent):
-            self._handle_node_exception(event)
-            return
-
-        if isinstance(event, NodeRunPauseRequestedEvent):
-            self._handle_node_pause_requested(event)
+        match event:
+            case GraphRunStartedEvent():
+                self._handle_graph_run_started()
+            case GraphRunSucceededEvent():
+                self._handle_graph_run_succeeded(event)
+            case GraphRunPartialSucceededEvent():
+                self._handle_graph_run_partial_succeeded(event)
+            case GraphRunFailedEvent():
+                self._handle_graph_run_failed(event)
+            case GraphRunAbortedEvent():
+                self._handle_graph_run_aborted(event)
+            case GraphRunPausedEvent():
+                self._handle_graph_run_paused(event)
+            case NodeRunRetryEvent():
+                self._handle_node_retry(event)
+            case NodeRunStartedEvent():
+                self._handle_node_started(event)
+            case NodeRunSucceededEvent():
+                self._handle_node_succeeded(event)
+            case NodeRunFailedEvent():
+                self._handle_node_failed(event)
+            case NodeRunExceptionEvent():
+                self._handle_node_exception(event)
+            case NodeRunPauseRequestedEvent():
+                self._handle_node_pause_requested(event)
 
     def on_graph_end(self, error: Exception | None) -> None:
         return
@@ -403,8 +383,13 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
 
         conversation_id = self._system_variables().get(SystemVariableKey.CONVERSATION_ID.value)
         external_trace_id = None
+        parent_trace_context = None
         if isinstance(self._application_generate_entity, (WorkflowAppGenerateEntity, AdvancedChatAppGenerateEntity)):
-            external_trace_id = self._application_generate_entity.extras.get("external_trace_id")
+            extras = self._application_generate_entity.extras
+            external_trace_id = extras.get("external_trace_id")
+            parent_trace_context = extras.get("parent_trace_context")
+            if isinstance(parent_trace_context, ParentTraceContext):
+                parent_trace_context = parent_trace_context.model_dump(exclude_none=True)
 
         trace_task = TraceTask(
             TraceTaskName.WORKFLOW_TRACE,
@@ -412,6 +397,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
             conversation_id=conversation_id,
             user_id=self._trace_manager.user_id,
             external_trace_id=external_trace_id,
+            parent_trace_context=parent_trace_context,
         )
         self._trace_manager.add_trace_task(trace_task)
 

api/core/base/tts/app_generator_tts_publisher.py

@@ -95,13 +95,14 @@ class AppGeneratorTTSPublisher:
                     message_content = message.event.chunk.delta.message.content
                     if not message_content:
                         continue
-                    if isinstance(message_content, str):
-                        self.msg_text += message_content
-                    elif isinstance(message_content, list):
-                        for content in message_content:
-                            if not isinstance(content, TextPromptMessageContent):
-                                continue
-                            self.msg_text += content.data
+                    match message_content:
+                        case str():
+                            self.msg_text += message_content
+                        case list():
+                            for content in message_content:
+                                if not isinstance(content, TextPromptMessageContent):
+                                    continue
+                                self.msg_text += content.data
                 elif isinstance(message.event, QueueTextChunkEvent):
                     self.msg_text += message.event.text
                 elif isinstance(message.event, QueueNodeSucceededEvent):

api/core/datasource/datasource_file_manager.py

@@ -35,8 +35,11 @@ class DatasourceFileManager:
         timestamp = str(int(time.time()))
         nonce = os.urandom(16).hex()
         data_to_sign = f"file-preview|{datasource_file_id}|{timestamp}|{nonce}"
-        secret_key = dify_config.SECRET_KEY.encode() if dify_config.SECRET_KEY else b""
-        sign = hmac.new(secret_key, data_to_sign.encode(), hashlib.sha256).digest()
+        sign = hmac.new(
+            dify_config.SECRET_KEY.encode(),
+            data_to_sign.encode(),
+            hashlib.sha256,
+        ).digest()
         encoded_sign = base64.urlsafe_b64encode(sign).decode()
 
         return f"{file_preview_url}?timestamp={timestamp}&nonce={nonce}&sign={encoded_sign}"
@@ -47,8 +50,11 @@ class DatasourceFileManager:
         verify signature
         """
         data_to_sign = f"file-preview|{datasource_file_id}|{timestamp}|{nonce}"
-        secret_key = dify_config.SECRET_KEY.encode() if dify_config.SECRET_KEY else b""
-        recalculated_sign = hmac.new(secret_key, data_to_sign.encode(), hashlib.sha256).digest()
+        recalculated_sign = hmac.new(
+            dify_config.SECRET_KEY.encode(),
+            data_to_sign.encode(),
+            hashlib.sha256,
+        ).digest()
         recalculated_encoded_sign = base64.urlsafe_b64encode(recalculated_sign).decode()
 
         # verify signature

api/core/entities/execution_extra_content.py

@@ -5,7 +5,7 @@ from typing import Any, TypeAlias
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from graphon.nodes.human_input.entities import FormInput, UserAction
+from graphon.nodes.human_input.entities import FormInputConfig, UserActionConfig
 from models.execution_extra_content import ExecutionContentType
 
 
@@ -16,8 +16,8 @@ class HumanInputFormDefinition(BaseModel):
     node_id: str
     node_title: str
     form_content: str
-    inputs: Sequence[FormInput] = Field(default_factory=list)
-    actions: Sequence[UserAction] = Field(default_factory=list)
+    inputs: Sequence[FormInputConfig] = Field(default_factory=list)
+    actions: Sequence[UserActionConfig] = Field(default_factory=list)
     display_in_ui: bool = False
     form_token: str | None = None
     resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)

api/core/entities/provider_configuration.py

@@ -171,9 +171,9 @@ class ProviderConfiguration(BaseModel):
                 current_credential_id = self.custom_configuration.provider.current_credential_id
 
             if current_credential_id:
-                from core.helper.credential_utils import check_credential_policy_compliance
+                from core.helper.credential_utils import runtime_check_credential_policy_compliance
 
-                check_credential_policy_compliance(
+                runtime_check_credential_policy_compliance(
                     credential_id=current_credential_id,
                     provider=self.provider.provider,
                     credential_type=PluginCredentialType.MODEL,
@@ -182,9 +182,9 @@ class ProviderConfiguration(BaseModel):
                 # no current credential id, check all available credentials
                 if self.custom_configuration.provider:
                     for credential_configuration in self.custom_configuration.provider.available_credentials:
-                        from core.helper.credential_utils import check_credential_policy_compliance
+                        from core.helper.credential_utils import runtime_check_credential_policy_compliance
 
-                        check_credential_policy_compliance(
+                        runtime_check_credential_policy_compliance(
                             credential_id=credential_configuration.credential_id,
                             provider=self.provider.provider,
                             credential_type=PluginCredentialType.MODEL,

api/core/errors/error.py

@@ -1,9 +1,9 @@
 class LLMError(ValueError):
     """Base class for all LLM exceptions."""
 
-    description: str | None = None
+    description: str = ""
 
-    def __init__(self, description: str | None = None):
+    def __init__(self, description: str = ""):
         self.description = description
 
 

api/core/helper/credential_utils.py

@@ -2,6 +2,7 @@
 Credential utility functions for checking credential existence and policy compliance.
 """
 
+from configs import dify_config
 from core.entities import PluginCredentialType
 
 
@@ -39,6 +40,16 @@ def is_credential_exists(credential_id: str, credential_type: "PluginCredentialT
         return False
 
 
+def runtime_check_credential_policy_compliance(
+    credential_id: str, provider: str, credential_type: "PluginCredentialType", check_existence: bool = True
+):
+    if dify_config.ENTERPRISE_DISABLE_RUNTIME_CREDENTIAL_CHECK:
+        return
+    check_credential_policy_compliance(
+        credential_id=credential_id, provider=provider, credential_type=credential_type, check_existence=check_existence
+    )
+
+
 def check_credential_policy_compliance(
     credential_id: str, provider: str, credential_type: "PluginCredentialType", check_existence: bool = True
 ) -> None:

api/core/helper/trace_id_helper.py

@@ -3,6 +3,17 @@ import re
 from collections.abc import Mapping
 from typing import Any
 
+from pydantic import BaseModel, ConfigDict, StrictStr, ValidationError
+
+
+class ParentTraceContext(BaseModel):
+    """Typed parent trace context propagated from an outer workflow tool node."""
+
+    parent_workflow_run_id: StrictStr
+    parent_node_execution_id: StrictStr | None = None
+
+    model_config = ConfigDict(extra="forbid")
+
 
 def is_valid_trace_id(trace_id: str) -> bool:
     """
@@ -61,6 +72,30 @@ def extract_external_trace_id_from_args(args: Mapping[str, Any]):
     return {}
 
 
+def extract_parent_trace_context_from_args(args: Mapping[str, Any]) -> dict[str, ParentTraceContext]:
+    """
+    Extract 'parent_trace_context' from args.
+
+    Returns a dict suitable for use in extras when both parent identifiers exist.
+    Returns an empty dict if the context is missing or incomplete.
+    """
+    parent_trace_context = args.get("parent_trace_context")
+    if isinstance(parent_trace_context, ParentTraceContext):
+        context = parent_trace_context
+    elif isinstance(parent_trace_context, Mapping):
+        try:
+            context = ParentTraceContext.model_validate(parent_trace_context)
+        except ValidationError:
+            return {}
+    else:
+        return {}
+
+    if context.parent_node_execution_id is None:
+        return {}
+
+    return {"parent_trace_context": context}
+
+
 def get_trace_id_from_otel_context() -> str | None:
     """
     Retrieve the current trace ID from the active OpenTelemetry trace context.

api/core/indexing_runner.py

@@ -324,9 +324,10 @@ class IndexingRunner:
         # one extract_setting is one source document
         for extract_setting in extract_settings:
             # extract
-            processing_rule = DatasetProcessRule(
-                mode=tmp_processing_rule["mode"], rules=json.dumps(tmp_processing_rule["rules"])
-            )
+            processing_rule = {
+                "mode": tmp_processing_rule["mode"],
+                "rules": tmp_processing_rule.get("rules"),
+            }
             # Extract document content
             text_docs = index_processor.extract(extract_setting, process_rule_mode=tmp_processing_rule["mode"])
             # Cleaning and segmentation
@@ -334,7 +335,7 @@ class IndexingRunner:
                 text_docs,
                 current_user=None,
                 embedding_model_instance=embedding_model_instance,
-                process_rule=processing_rule.to_dict(),
+                process_rule=processing_rule,
                 tenant_id=tenant_id,
                 doc_language=doc_language,
                 preview=True,

api/core/llm_generator/llm_generator.py

@@ -435,7 +435,7 @@ class LLMGenerator:
             stream=False,
         )
 
-        # Runtime type check since pyright has issues with the overload
+        # Runtime type check for overload narrowing.
         if not isinstance(result, LLMResult):
             raise TypeError("Expected LLMResult when stream=False")
         response = result

api/core/llm_generator/output_parser/structured_output.py

@@ -288,11 +288,13 @@ def _parse_structured_output(result_text: str) -> Mapping[str, Any]:
     except ValidationError:
         # if the result_text is not a valid json, try to repair it
         temp_parsed = json_repair.loads(result_text)
-        if not isinstance(temp_parsed, dict):
-            # handle reasoning model like deepseek-r1 got '<think>\n\n</think>\n' prefix
-            if isinstance(temp_parsed, list):
+        match temp_parsed:
+            case dict():
+                pass
+            case list():
+                # handle reasoning model like deepseek-r1 got '<think>\n\n</think>\n' prefix
                 temp_parsed = next((item for item in temp_parsed if isinstance(item, dict)), {})
-            else:
+            case _:
                 raise OutputParserError(f"Failed to parse structured output: {result_text}")
         structured_output = cast(dict, temp_parsed)
     return structured_output
@@ -341,12 +343,13 @@ def remove_additional_properties(schema: dict[str, Any]) -> None:
 
     # Process nested structures recursively
     for value in schema.values():
-        if isinstance(value, dict):
-            remove_additional_properties(value)
-        elif isinstance(value, list):
-            for item in value:
-                if isinstance(item, dict):
-                    remove_additional_properties(item)
+        match value:
+            case dict():
+                remove_additional_properties(value)
+            case list():
+                for item in value:
+                    if isinstance(item, dict):
+                        remove_additional_properties(item)
 
 
 def convert_boolean_to_string(schema: dict[str, Any]) -> None:
@@ -364,9 +367,10 @@ def convert_boolean_to_string(schema: dict[str, Any]) -> None:
 
     # Process nested dictionaries and lists recursively
     for value in schema.values():
-        if isinstance(value, dict):
-            convert_boolean_to_string(value)
-        elif isinstance(value, list):
-            for item in value:
-                if isinstance(item, dict):
-                    convert_boolean_to_string(item)
+        match value:
+            case dict():
+                convert_boolean_to_string(value)
+            case list():
+                for item in value:
+                    if isinstance(item, dict):
+                        convert_boolean_to_string(item)

api/core/mcp/types.py

@@ -2,7 +2,7 @@ from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Annotated, Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, FileUrl, RootModel
+from pydantic import BaseModel, ConfigDict, Field, FileUrl, RootModel, model_validator
 from pydantic.networks import AnyUrl, UrlConstraints
 
 """
@@ -173,7 +173,21 @@ class JSONRPCError(BaseModel):
 
 
 class JSONRPCMessage(RootModel[JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError]):
-    pass
+    @model_validator(mode="before")
+    @classmethod
+    def _select_message_type(
+        cls, value: Any
+    ) -> JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError | Any:
+        if isinstance(value, dict):
+            if "result" in value:
+                return JSONRPCResponse.model_validate(value)
+            if "error" in value:
+                return JSONRPCError.model_validate(value)
+            if "method" in value:
+                if "id" in value:
+                    return JSONRPCRequest.model_validate(value)
+                return JSONRPCNotification.model_validate(value)
+        return value
 
 
 class EmptyResult(Result):

api/core/memory/token_buffer_memory.py

@@ -86,12 +86,10 @@ class TokenBufferMemory:
 
         detail = ImagePromptMessageContent.DETAIL.HIGH
         if file_extra_config and app_record:
-            # Build files directly without filtering by belongs_to
             file_objs = [
                 file_factory.build_from_message_file(
                     message_file=message_file,
                     tenant_id=app_record.tenant_id,
-                    config=file_extra_config,
                     access_controller=_file_access_controller,
                 )
                 for message_file in message_files

api/core/model_manager.py

@@ -391,10 +391,10 @@ class ModelInstance:
 
             # Additional policy compliance check as fallback (in case fetch_next didn't catch it)
             try:
-                from core.helper.credential_utils import check_credential_policy_compliance
+                from core.helper.credential_utils import runtime_check_credential_policy_compliance
 
                 if lb_config.credential_id:
-                    check_credential_policy_compliance(
+                    runtime_check_credential_policy_compliance(
                         credential_id=lb_config.credential_id,
                         provider=self.provider,
                         credential_type=PluginCredentialType.MODEL,
@@ -630,10 +630,10 @@ class LBModelManager:
 
             # Check policy compliance for the selected configuration
             try:
-                from core.helper.credential_utils import check_credential_policy_compliance
+                from core.helper.credential_utils import runtime_check_credential_policy_compliance
 
                 if config.credential_id:
-                    check_credential_policy_compliance(
+                    runtime_check_credential_policy_compliance(
                         credential_id=config.credential_id,
                         provider=self._provider,
                         credential_type=PluginCredentialType.MODEL,