[autofix.ci] apply automated fixes

# Conflicts:
#	api/tests/unit_tests/core/rag/extractor/test_word_extractor.py

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

@@ -204,6 +204,16 @@ When assigned to test a directory/path, test **ALL content** within that path:
 
 > See [Test Structure Template](#test-structure-template) for correct import/mock patterns.
 
+### `nuqs` Query State Testing (Required for URL State Hooks)
+
+When a component or hook uses `useQueryState` / `useQueryStates`:
+
+- ✅ Use `NuqsTestingAdapter` (prefer shared helpers in `web/test/nuqs-testing.tsx`)
+- ✅ Assert URL synchronization via `onUrlUpdate` (`searchParams`, `options.history`)
+- ✅ For custom parsers (`createParser`), keep `parse` and `serialize` bijective and add round-trip edge cases (`%2F`, `%25`, spaces, legacy encoded values)
+- ✅ Verify default-clearing behavior (default values should be removed from URL when applicable)
+- ⚠️ Only mock `nuqs` directly when URL behavior is explicitly out of scope for the test
+
 ## Core Principles
 
 ### 1. AAA Pattern (Arrange-Act-Assert)

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

@@ -80,6 +80,9 @@ Use this checklist when generating or reviewing tests for Dify frontend componen
 - [ ] Router mocks match actual Next.js API
 - [ ] Mocks reflect actual component conditional behavior
 - [ ] Only mock: API services, complex context providers, third-party libs
+- [ ] For `nuqs` URL-state tests, wrap with `NuqsTestingAdapter` (prefer `web/test/nuqs-testing.tsx`)
+- [ ] For `nuqs` URL-state tests, assert `onUrlUpdate` payload (`searchParams`, `options.history`)
+- [ ] If custom `nuqs` parser exists, add round-trip tests for encoded edge cases (`%2F`, `%25`, spaces, legacy encoded values)
 
 ### Queries
 

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

@@ -125,6 +125,31 @@ describe('Component', () => {
 })
 ```
 
+### 2.1 `nuqs` Query State (Preferred: Testing Adapter)
+
+For tests that validate URL query behavior, use `NuqsTestingAdapter` instead of mocking `nuqs` directly.
+
+```typescript
+import { renderHookWithNuqs } from '@/test/nuqs-testing'
+
+it('should sync query to URL with push history', async () => {
+  const { result, onUrlUpdate } = renderHookWithNuqs(() => useMyQueryState(), {
+    searchParams: '?page=1',
+  })
+
+  act(() => {
+    result.current.setQuery({ page: 2 })
+  })
+
+  await waitFor(() => expect(onUrlUpdate).toHaveBeenCalled())
+  const update = onUrlUpdate.mock.calls[onUrlUpdate.mock.calls.length - 1][0]
+  expect(update.options.history).toBe('push')
+  expect(update.searchParams.get('page')).toBe('2')
+})
+```
+
+Use direct `vi.mock('nuqs')` only when URL synchronization is intentionally out of scope.
+
 ### 3. Portal Components (with Shared State)
 
 ```typescript

.agents/skills/orpc-contract-first/SKILL.md

@@ -1,43 +1,100 @@
 ---
 name: orpc-contract-first
-description: Guide for implementing oRPC contract-first API patterns in Dify frontend. Triggers when creating new API contracts, adding service endpoints, integrating TanStack Query with typed contracts, or migrating legacy service calls to oRPC. Use for all API layer work in web/contract and web/service directories.
+description: Guide for implementing oRPC contract-first API patterns in Dify frontend. Trigger when creating or updating contracts in web/contract, wiring router composition, integrating TanStack Query with typed contracts, migrating legacy service calls to oRPC, or deciding whether to call queryOptions directly vs extracting a helper or use-* hook in web/service.
 ---
 
 # oRPC Contract-First Development
 
-## Project Structure
+## Intent
 
-```
+- Keep contract as single source of truth in `web/contract/*`.
+- Default query usage: 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           # Base contract (inputStructure: 'detailed')
-├── router.ts         # Router composition & type exports
-├── marketplace.ts    # Marketplace contracts
-└── console/          # Console contracts by domain
-    ├── system.ts
-    └── billing.ts
+├── base.ts
+├── router.ts
+├── marketplace.ts
+└── console/
+    ├── billing.ts
+    └── ...other domains
+web/service/client.ts
 ```
 
-## Workflow
+## Core Workflow
 
-1. **Create contract** in `web/contract/console/{domain}.ts`
-   - Import `base` from `../base` and `type` from `@orpc/contract`
-   - Define route with `path`, `method`, `input`, `output`
+1. Define contract in `web/contract/console/{domain}.ts` or `web/contract/marketplace.ts`
+   - Use `base.route({...}).output(type<...>())` as baseline.
+   - Add `.input(type<...>())` only when request has `params/query/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 utils.
 
-2. **Register in router** at `web/contract/router.ts`
-   - Import directly from domain file (no barrel files)
-   - Nest by API prefix: `billing: { invoices, bindPartnerStack }`
+```typescript
+import { useQuery } from '@tanstack/react-query'
+import { consoleQuery } from '@/service/client'
 
-3. **Create hooks** in `web/service/use-{domain}.ts`
-   - Use `consoleQuery.{group}.{contract}.queryKey()` for query keys
-   - Use `consoleClient.{group}.{contract}()` for API calls
+const invoiceQuery = useQuery(consoleQuery.billing.invoices.queryOptions({
+  staleTime: 5 * 60 * 1000,
+  throwOnError: true,
+  select: invoice => invoice.url,
+}))
+```
 
-## Key Rules
+## Query Usage Decision Rule
+
+1. Default: call site directly uses `*.queryOptions(...)`.
+2. If 3+ call sites share the same extra options (for example `retry: false`), extract a small queryOptions helper, not a `use-*` passthrough hook.
+3. Create `web/service/use-{domain}.ts` only for orchestration:
+   - Combine multiple queries/mutations.
+   - Share domain-level derived state or invalidation helpers.
+
+```typescript
+const invoicesBaseQueryOptions = () =>
+  consoleQuery.billing.invoices.queryOptions({ retry: false })
+
+const invoiceQuery = useQuery({
+  ...invoicesBaseQueryOptions(),
+  throwOnError: true,
+})
+```
+
+## Mutation Usage Decision Rule
+
+1. Default: call mutation helpers from `consoleQuery` / `marketplaceQuery`, for example `useMutation(consoleQuery.billing.bindPartnerStack.mutationOptions(...))`.
+2. If mutation flow is heavily custom, use oRPC clients as `mutationFn` (for example `consoleClient.xxx` / `marketplaceClient.xxx`), instead of generic handwritten non-oRPC mutation logic.
+
+## Key API Guide (`.key` vs `.queryKey` vs `.mutationKey`)
+
+- `.key(...)`:
+  - Use for partial matching operations (recommended for invalidation/refetch/cancel patterns).
+  - Example: `queryClient.invalidateQueries({ queryKey: consoleQuery.billing.key() })`
+- `.queryKey(...)`:
+  - Use for a specific query's full key (exact query identity / direct cache addressing).
+- `.mutationKey(...)`:
+  - Use for a specific mutation's full key.
+  - Typical use cases: mutation defaults registration, mutation-status filtering (`useIsMutating`, `queryClient.isMutating`), or explicit devtools grouping.
+
+## Anti-Patterns
+
+- Do not wrap `useQuery` with `options?: Partial<UseQueryOptions>`.
+- Do not split local `queryKey/queryFn` when oRPC `queryOptions` already exists and fits the use case.
+- Do not create thin `use-*` passthrough hooks for a single endpoint.
+- Reason: these patterns can degrade inference (`data` may become `unknown`, especially around `throwOnError`/`select`) and add unnecessary indirection.
+
+## Contract Rules
 
 - **Input structure**: Always use `{ params, query?, body? }` format
+- **No-input GET**: Omit `.input(...)`; do not use `.input(type<unknown>())`
 - **Path params**: Use `{paramName}` in path, match in `params` object
-- **Router nesting**: Group by API prefix (e.g., `/billing/*` → `billing: {}`)
+- **Router nesting**: Group by API prefix (e.g., `/billing/*` -> `billing: {}`)
 - **No barrel files**: Import directly from specific files
 - **Types**: Import from `@/types/`, use `type<T>()` helper
+- **Mutations**: Prefer `mutationOptions`; use explicit `mutationKey` mainly for defaults/filtering/devtools
 
 ## Type Export
 

.devcontainer/post_create_command.sh

@@ -7,7 +7,7 @@ cd web && pnpm install
 pipx install uv
 
 echo "alias start-api=\"cd $WORKSPACE_ROOT/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug\"" >> ~/.bashrc
-echo "alias start-worker=\"cd $WORKSPACE_ROOT/api && uv run python -m celery -A app.celery worker -P threads -c 1 --loglevel INFO -Q dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention\"" >> ~/.bashrc
+echo "alias start-worker=\"cd $WORKSPACE_ROOT/api && uv run python -m celery -A app.celery worker -P threads -c 1 --loglevel INFO -Q dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention\"" >> ~/.bashrc
 echo "alias start-web=\"cd $WORKSPACE_ROOT/web && pnpm dev:inspect\"" >> ~/.bashrc
 echo "alias start-web-prod=\"cd $WORKSPACE_ROOT/web && pnpm build && pnpm start\"" >> ~/.bashrc
 echo "alias start-containers=\"cd $WORKSPACE_ROOT/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d\"" >> ~/.bashrc

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

@@ -0,0 +1,33 @@
+name: Setup Web Environment
+description: Setup pnpm, Node.js, and install web dependencies.
+
+inputs:
+  node-version:
+    description: Node.js version to use
+    required: false
+    default: "22"
+  install-dependencies:
+    description: Whether to install web dependencies after setting up Node.js
+    required: false
+    default: "true"
+
+runs:
+  using: composite
+  steps:
+    - name: Install pnpm
+      uses: pnpm/action-setup@41ff72655975bd51cab0327fa583b6e92b6d3061 # v4.2.0
+      with:
+        package_json_file: web/package.json
+        run_install: false
+
+    - name: Setup Node.js
+      uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v6.3.0
+      with:
+        node-version: ${{ inputs.node-version }}
+        cache: pnpm
+        cache-dependency-path: ./web/pnpm-lock.yaml
+
+    - name: Install dependencies
+      if: ${{ inputs.install-dependencies == 'true' }}
+      shell: bash
+      run: pnpm --dir web install --frozen-lockfile

.github/dependabot.yml

@@ -1,25 +1,57 @@
 version: 2
 
-multi-ecosystem-groups:
-  python:
-    schedule:
-      interval: "weekly"  # or whatever schedule you want
-
 updates:
   - package-ecosystem: "pip"
     directory: "/api"
     open-pull-requests-limit: 2
-    patterns: ["*"]
     schedule:
       interval: "weekly"
+    groups:
+      python-dependencies:
+        patterns:
+          - "*"
   - package-ecosystem: "uv"
     directory: "/api"
     open-pull-requests-limit: 2
-    patterns: ["*"]
     schedule:
       interval: "weekly"
+    groups:
+      uv-dependencies:
+        patterns:
+          - "*"
   - package-ecosystem: "npm"
     directory: "/web"
     schedule:
       interval: "weekly"
     open-pull-requests-limit: 2
+    ignore:
+      - dependency-name: "ky"
+      - dependency-name: "tailwind-merge"
+        update-types: ["version-update:semver-major"]
+      - dependency-name: "tailwindcss"
+        update-types: ["version-update:semver-major"]
+      - dependency-name: "react-syntax-highlighter"
+        update-types: ["version-update:semver-major"]
+      - dependency-name: "react-window"
+        update-types: ["version-update:semver-major"]
+    groups:
+      lexical:
+        patterns:
+          - "lexical"
+          - "@lexical/*"
+      storybook:
+        patterns:
+          - "storybook"
+          - "@storybook/*"
+      eslint-group:
+        patterns:
+          - "*eslint*"
+      npm-dependencies:
+        patterns:
+          - "*"
+        exclude-patterns:
+          - "lexical"
+          - "@lexical/*"
+          - "storybook"
+          - "@storybook/*"
+          - "*eslint*"

.github/workflows/anti-slop.yml

@@ -0,0 +1,17 @@
+name: Anti-Slop PR Check
+
+on:
+  pull_request_target:
+    types: [opened, edited, synchronize]
+
+permissions:
+  pull-requests: write
+  contents: read
+
+jobs:
+  anti-slop:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: peakoss/anti-slop@v0
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/api-tests.yml

@@ -22,12 +22,12 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@v7
+        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
         with:
           enable-cache: true
           python-version: ${{ matrix.python-version }}
@@ -51,7 +51,7 @@ jobs:
         run: sh .github/workflows/expose_service_ports.sh
 
       - name: Set up Sandbox
-        uses: hoverkraft-tech/compose-action@v2
+        uses: hoverkraft-tech/compose-action@4894d2492015c1774ee5a13a95b1072093087ec3 # v2.5.0
         with:
           compose-file: |
             docker/docker-compose.middleware.yaml

.github/workflows/autofix.yml

@@ -12,22 +12,22 @@ jobs:
     if: github.repository == 'langgenius/dify'
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
 
       - name: Check Docker Compose inputs
         id: docker-compose-changes
-        uses: tj-actions/changed-files@v47
+        uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # v47.0.5
         with:
           files: |
             docker/generate_docker_compose
             docker/.env.example
             docker/docker-compose-template.yaml
             docker/docker-compose.yaml
-      - uses: actions/setup-python@v6
+      - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
         with:
           python-version: "3.11"
 
-      - uses: astral-sh/setup-uv@v7
+      - uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
 
       - name: Generate Docker Compose
         if: steps.docker-compose-changes.outputs.any_changed == 'true'
@@ -84,4 +84,14 @@ jobs:
         run: |
           uvx --python 3.13 mdformat . --exclude ".agents/skills/**"
 
-      - uses: autofix-ci/action@635ffb0c9798bd160680f18fd73371e355b85f27
+      - name: Setup web environment
+        uses: ./.github/actions/setup-web
+        with:
+          node-version: "24"
+
+      - name: ESLint autofix
+        run: |
+          cd web
+          pnpm eslint --concurrency=2 --prune-suppressions
+
+      - uses: autofix-ci/action@7a166d7532b277f34e16238930461bf77f9d7ed8 # v1.3.3

.github/workflows/build-push.yml

@@ -53,26 +53,26 @@ jobs:
           echo "PLATFORM_PAIR=${platform//\//-}" >> $GITHUB_ENV
 
       - name: Login to Docker Hub
-        uses: docker/login-action@v3
+        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v4.0.0
         with:
           username: ${{ env.DOCKERHUB_USER }}
           password: ${{ env.DOCKERHUB_TOKEN }}
 
       - name: Set up QEMU
-        uses: docker/setup-qemu-action@v3
+        uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v4.0.0
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4.0.0
 
       - name: Extract metadata for Docker
         id: meta
-        uses: docker/metadata-action@v5
+        uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v6.0.0
         with:
           images: ${{ env[matrix.image_name_env] }}
 
       - name: Build Docker image
         id: build
-        uses: docker/build-push-action@v6
+        uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v7.0.0
         with:
           context: "{{defaultContext}}:${{ matrix.context }}"
           platforms: ${{ matrix.platform }}
@@ -91,7 +91,7 @@ jobs:
           touch "/tmp/digests/${sanitized_digest}"
 
       - name: Upload digest
-        uses: actions/upload-artifact@v6
+        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
         with:
           name: digests-${{ matrix.context }}-${{ env.PLATFORM_PAIR }}
           path: /tmp/digests/*
@@ -113,21 +113,21 @@ jobs:
             context: "web"
     steps:
       - name: Download digests
-        uses: actions/download-artifact@v7
+        uses: actions/download-artifact@70fc10c6e5e1ce46ad2ea6f2b72d43f7d47b13c3 # v8.0.0
         with:
           path: /tmp/digests
           pattern: digests-${{ matrix.context }}-*
           merge-multiple: true
 
       - name: Login to Docker Hub
-        uses: docker/login-action@v3
+        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v4.0.0
         with:
           username: ${{ env.DOCKERHUB_USER }}
           password: ${{ env.DOCKERHUB_TOKEN }}
 
       - name: Extract metadata for Docker
         id: meta
-        uses: docker/metadata-action@v5
+        uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v6.0.0
         with:
           images: ${{ env[matrix.image_name_env] }}
           tags: |

.github/workflows/db-migration-test.yml

@@ -13,13 +13,13 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           fetch-depth: 0
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@v7
+        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
         with:
           enable-cache: true
           python-version: "3.12"
@@ -40,7 +40,7 @@ jobs:
           cp middleware.env.example middleware.env
 
       - name: Set up Middlewares
-        uses: hoverkraft-tech/compose-action@v2.0.2
+        uses: hoverkraft-tech/compose-action@4894d2492015c1774ee5a13a95b1072093087ec3 # v2.5.0
         with:
           compose-file: |
             docker/docker-compose.middleware.yaml
@@ -63,13 +63,13 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           fetch-depth: 0
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@v7
+        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
         with:
           enable-cache: true
           python-version: "3.12"
@@ -94,7 +94,7 @@ jobs:
           sed -i 's/DB_USERNAME=postgres/DB_USERNAME=mysql/' middleware.env
 
       - name: Set up Middlewares
-        uses: hoverkraft-tech/compose-action@v2.0.2
+        uses: hoverkraft-tech/compose-action@4894d2492015c1774ee5a13a95b1072093087ec3 # v2.5.0
         with:
           compose-file: |
             docker/docker-compose.middleware.yaml

.github/workflows/deploy-agent-dev.yml

@@ -19,7 +19,7 @@ jobs:
       github.event.workflow_run.head_branch == 'deploy/agent-dev'
     steps:
       - name: Deploy to server
-        uses: appleboy/ssh-action@v1
+        uses: appleboy/ssh-action@0ff4204d59e8e51228ff73bce53f80d53301dee2 # v1.2.5
         with:
           host: ${{ secrets.AGENT_DEV_SSH_HOST }}
           username: ${{ secrets.SSH_USER }}

.github/workflows/deploy-dev.yml

@@ -16,7 +16,7 @@ jobs:
       github.event.workflow_run.head_branch == 'deploy/dev'
     steps:
       - name: Deploy to server
-        uses: appleboy/ssh-action@v1
+        uses: appleboy/ssh-action@0ff4204d59e8e51228ff73bce53f80d53301dee2 # v1.2.5
         with:
           host: ${{ secrets.SSH_HOST }}
           username: ${{ secrets.SSH_USER }}

.github/workflows/deploy-hitl.yml

@@ -16,7 +16,7 @@ jobs:
       github.event.workflow_run.head_branch == 'build/feat/hitl'
     steps:
       - name: Deploy to server
-        uses: appleboy/ssh-action@v1
+        uses: appleboy/ssh-action@0ff4204d59e8e51228ff73bce53f80d53301dee2 # v1.2.5
         with:
           host: ${{ secrets.HITL_SSH_HOST }}
           username: ${{ secrets.SSH_USER }}

.github/workflows/docker-build.yml

@@ -32,13 +32,13 @@ jobs:
             context: "web"
     steps:
       - name: Set up QEMU
-        uses: docker/setup-qemu-action@v3
+        uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v4.0.0
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4.0.0
 
       - name: Build Docker Image
-        uses: docker/build-push-action@v6
+        uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v7.0.0
         with:
           push: false
           context: "{{defaultContext}}:${{ matrix.context }}"

.github/workflows/labeler.yml

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

.github/workflows/main-ci.yml

@@ -27,8 +27,8 @@ jobs:
       vdb-changed: ${{ steps.changes.outputs.vdb }}
       migration-changed: ${{ steps.changes.outputs.migration }}
     steps:
-      - uses: actions/checkout@v6
-      - uses: dorny/paths-filter@v3
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      - uses: dorny/paths-filter@de90cc6fb38fc0963ad72b210f1f284cd68cea36 # v3.0.2
         id: changes
         with:
           filters: |
@@ -39,6 +39,7 @@ jobs:
             web:
               - 'web/**'
               - '.github/workflows/web-tests.yml'
+              - '.github/actions/setup-web/**'
             vdb:
               - 'api/core/rag/datasource/**'
               - 'docker/**'

.github/workflows/pyrefly-diff-comment.yml

@@ -21,7 +21,7 @@ jobs:
     if: ${{ github.event.workflow_run.conclusion == 'success' && github.event.workflow_run.pull_requests[0].head.repo.full_name != github.repository }}
     steps:
       - name: Download pyrefly diff artifact
-        uses: actions/github-script@v8
+        uses: actions/github-script@ed597411d8f924073f98dfc5c65a23a2325f34cd # v8.0.0
         with:
           github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
@@ -49,7 +49,7 @@ jobs:
         run: unzip -o pyrefly_diff.zip
 
       - name: Post comment
-        uses: actions/github-script@v8
+        uses: actions/github-script@ed597411d8f924073f98dfc5c65a23a2325f34cd # v8.0.0
         with:
           github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |

.github/workflows/pyrefly-diff.yml

@@ -17,12 +17,12 @@ jobs:
       pull-requests: write
     steps:
       - name: Checkout PR branch
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           fetch-depth: 0
 
       - name: Setup Python & UV
-        uses: astral-sh/setup-uv@v5
+        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
         with:
           enable-cache: true
 
@@ -55,7 +55,7 @@ jobs:
           echo ${{ github.event.pull_request.number }} > pr_number.txt
 
       - name: Upload pyrefly diff
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
         with:
           name: pyrefly_diff
           path: |
@@ -64,7 +64,7 @@ jobs:
 
       - name: Comment PR with pyrefly diff
         if: ${{ github.event.pull_request.head.repo.full_name == github.repository }}
-        uses: actions/github-script@v8
+        uses: actions/github-script@ed597411d8f924073f98dfc5c65a23a2325f34cd # v8.0.0
         with:
           github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |

.github/workflows/semantic-pull-request.yml

@@ -16,6 +16,6 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Check title
-        uses: amannn/action-semantic-pull-request@v6.1.1
+        uses: amannn/action-semantic-pull-request@48f256284bd46cdaab1048c3721360e808335d50 # v6.1.1
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/stale.yml

@@ -18,7 +18,7 @@ jobs:
       pull-requests: write
 
     steps:
-      - uses: actions/stale@v10
+      - uses: actions/stale@b5d41d4e1d5dceea10e7104786b73624c18a190f # v10.2.0
         with:
           days-before-issue-stale: 15
           days-before-issue-close: 3

.github/workflows/style.yml

@@ -19,13 +19,13 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           persist-credentials: false
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v47
+        uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # v47.0.5
         with:
           files: |
             api/**
@@ -33,7 +33,7 @@ jobs:
 
       - name: Setup UV and Python
         if: steps.changed-files.outputs.any_changed == 'true'
-        uses: astral-sh/setup-uv@v7
+        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
         with:
           enable-cache: false
           python-version: "3.12"
@@ -67,36 +67,22 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           persist-credentials: false
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v47
+        uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # v47.0.5
         with:
           files: |
             web/**
             .github/workflows/style.yml
+            .github/actions/setup-web/**
 
-      - name: Install pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          package_json_file: web/package.json
-          run_install: false
-
-      - name: Setup NodeJS
-        uses: actions/setup-node@v6
+      - name: Setup web environment
         if: steps.changed-files.outputs.any_changed == 'true'
-        with:
-          node-version: 24
-          cache: pnpm
-          cache-dependency-path: ./web/pnpm-lock.yaml
-
-      - name: Web dependencies
-        if: steps.changed-files.outputs.any_changed == 'true'
-        working-directory: ./web
-        run: pnpm install --frozen-lockfile
+        uses: ./.github/actions/setup-web
 
       - name: Web style check
         if: steps.changed-files.outputs.any_changed == 'true'
@@ -134,14 +120,14 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           fetch-depth: 0
           persist-credentials: false
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v47
+        uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # v47.0.5
         with:
           files: |
             **.sh
@@ -152,7 +138,7 @@ jobs:
             .editorconfig
 
       - name: Super-linter
-        uses: super-linter/super-linter/slim@v8
+        uses: super-linter/super-linter/slim@61abc07d755095a68f4987d1c2c3d1d64408f1f9 # v8.5.0
         if: steps.changed-files.outputs.any_changed == 'true'
         env:
           BASH_SEVERITY: warning

.github/workflows/tool-test-sdks.yaml

@@ -21,14 +21,14 @@ jobs:
         working-directory: sdks/nodejs-client
 
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           persist-credentials: false
 
       - name: Use Node.js
-        uses: actions/setup-node@v6
+        uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v6.3.0
         with:
-          node-version: 24
+          node-version: 22
           cache: ''
           cache-dependency-path: 'pnpm-lock.yaml'
 

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

@@ -38,7 +38,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           fetch-depth: 0
           token: ${{ secrets.GITHUB_TOKEN }}
@@ -48,18 +48,10 @@ jobs:
           git config --global user.name "github-actions[bot]"
           git config --global user.email "github-actions[bot]@users.noreply.github.com"
 
-      - name: Install pnpm
-        uses: pnpm/action-setup@v4
+      - name: Setup web environment
+        uses: ./.github/actions/setup-web
         with:
-          package_json_file: web/package.json
-          run_install: false
-
-      - name: Set up Node.js
-        uses: actions/setup-node@v6
-        with:
-          node-version: 24
-          cache: pnpm
-          cache-dependency-path: ./web/pnpm-lock.yaml
+          install-dependencies: "false"
 
       - name: Detect changed files and generate diff
         id: detect_changes
@@ -130,7 +122,7 @@ jobs:
 
       - name: Run Claude Code for Translation Sync
         if: steps.detect_changes.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@v1
+        uses: anthropics/claude-code-action@26ec041249acb0a944c0a47b6c0c13f05dbc5b44 # v1.0.70
         with:
           anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
           github_token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/trigger-i18n-sync.yml

@@ -21,7 +21,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           fetch-depth: 0
 
@@ -59,7 +59,7 @@ jobs:
 
       - name: Trigger i18n sync workflow
         if: steps.detect.outputs.has_changes == 'true'
-        uses: peter-evans/repository-dispatch@v3
+        uses: peter-evans/repository-dispatch@28959ce8df70de7be546dd1250a005dd32156697 # v4.0.1
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
           event-type: i18n-sync

.github/workflows/vdb-tests.yml

@@ -19,19 +19,19 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           persist-credentials: false
 
       - name: Free Disk Space
-        uses: endersonmenezes/free-disk-space@v3
+        uses: endersonmenezes/free-disk-space@7901478139cff6e9d44df5972fd8ab8fcade4db1 # v3.2.2
         with:
           remove_dotnet: true
           remove_haskell: true
           remove_tool_cache: true
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@v7
+        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
         with:
           enable-cache: true
           python-version: ${{ matrix.python-version }}
@@ -60,7 +60,7 @@ jobs:
 #            tiflash
 
       - name: Set up Vector Stores (Weaviate, Qdrant, PGVector, Milvus, PgVecto-RS, Chroma, MyScale, ElasticSearch, Couchbase, OceanBase)
-        uses: hoverkraft-tech/compose-action@v2.0.2
+        uses: hoverkraft-tech/compose-action@4894d2492015c1774ee5a13a95b1072093087ec3 # v2.5.0
         with:
           compose-file: |
             docker/docker-compose.yaml

.github/workflows/web-tests.yml

@@ -26,32 +26,19 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           persist-credentials: false
 
-      - name: Install pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          package_json_file: web/package.json
-          run_install: false
-
-      - name: Setup Node.js
-        uses: actions/setup-node@v6
-        with:
-          node-version: 24
-          cache: pnpm
-          cache-dependency-path: ./web/pnpm-lock.yaml
-
-      - name: Install dependencies
-        run: pnpm install --frozen-lockfile
+      - name: Setup web environment
+        uses: ./.github/actions/setup-web
 
       - name: Run tests
         run: pnpm vitest run --reporter=blob --shard=${{ matrix.shardIndex }}/${{ matrix.shardTotal }} --coverage
 
       - name: Upload blob report
         if: ${{ !cancelled() }}
-        uses: actions/upload-artifact@v6
+        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
         with:
           name: blob-report-${{ matrix.shardIndex }}
           path: web/.vitest-reports/*
@@ -70,28 +57,15 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           persist-credentials: false
 
-      - name: Install pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          package_json_file: web/package.json
-          run_install: false
-
-      - name: Setup Node.js
-        uses: actions/setup-node@v6
-        with:
-          node-version: 24
-          cache: pnpm
-          cache-dependency-path: ./web/pnpm-lock.yaml
-
-      - name: Install dependencies
-        run: pnpm install --frozen-lockfile
+      - name: Setup web environment
+        uses: ./.github/actions/setup-web
 
       - name: Download blob reports
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@70fc10c6e5e1ce46ad2ea6f2b72d43f7d47b13c3 # v8.0.0
         with:
           path: web/.vitest-reports
           pattern: blob-report-*
@@ -419,7 +393,7 @@ jobs:
 
       - name: Upload Coverage Artifact
         if: steps.coverage-summary.outputs.has_coverage == 'true'
-        uses: actions/upload-artifact@v6
+        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
         with:
           name: web-coverage-report
           path: web/coverage
@@ -435,36 +409,22 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v6
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
         with:
           persist-credentials: false
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v47
+        uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # v47.0.5
         with:
           files: |
             web/**
             .github/workflows/web-tests.yml
+            .github/actions/setup-web/**
 
-      - name: Install pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          package_json_file: web/package.json
-          run_install: false
-
-      - name: Setup NodeJS
-        uses: actions/setup-node@v6
+      - name: Setup web environment
         if: steps.changed-files.outputs.any_changed == 'true'
-        with:
-          node-version: 24
-          cache: pnpm
-          cache-dependency-path: ./web/pnpm-lock.yaml
-
-      - name: Web dependencies
-        if: steps.changed-files.outputs.any_changed == 'true'
-        working-directory: ./web
-        run: pnpm install --frozen-lockfile
+        uses: ./.github/actions/setup-web
 
       - name: Web build check
         if: steps.changed-files.outputs.any_changed == 'true'

.gitignore

@@ -222,6 +222,7 @@ mise.toml
 
 # AI Assistant
 .roo/
+/.claude/worktrees/
 api/.env.backup
 /clickzetta
 

.vscode/launch.json.template

@@ -37,7 +37,7 @@
                 "-c",
                 "1",
                 "-Q",
-                "dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention,workflow_based_app_execution",
+                "dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention,workflow_based_app_execution",
                 "--loglevel",
                 "INFO"
             ],

AGENTS.md

@@ -29,7 +29,7 @@ The codebase is split into:
 
 ## Language Style
 
-- **Python**: Keep type hints on functions and attributes, and implement relevant special methods (e.g., `__repr__`, `__str__`).
+- **Python**: Keep type hints on functions and attributes, and implement relevant special methods (e.g., `__repr__`, `__str__`). Prefer `TypedDict` over `dict` or `Mapping` for type safety and better code documentation.
 - **TypeScript**: Use the strict config, rely on ESLint (`pnpm lint:fix` preferred) plus `pnpm type-check:tsgo`, and avoid `any` types.
 
 ## General Practices

Makefile

@@ -68,8 +68,9 @@ lint:
 	@echo "✅ Linting complete"
 
 type-check:
-	@echo "📝 Running type checks (basedpyright + mypy)..."
+	@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/' --check-untyped-defs --disable-error-code=import-untyped .
 	@echo "✅ Type checks complete"
 
@@ -131,7 +132,7 @@ help:
 	@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, mypy)"
+	@echo "  make type-check     - Run type checks (basedpyright, pyrefly, mypy)"
 	@echo "  make test           - Run backend unit tests (or TARGET_TESTS=./api/tests/<target_tests>)"
 	@echo ""
 	@echo "Docker Build Targets:"

README.md

@@ -56,7 +56,7 @@
   <a href="./docs/bn-BD/README.md"><img alt="README in বাংলা" src="https://img.shields.io/badge/বাংলা-d9d9d9"></a>
 </p>
 
-Dify is an open-source platform for developing LLM applications. Its intuitive interface combines agentic AI workflows, RAG pipelines, agent capabilities, model management, observability features, and more—allowing you to quickly move from prototype to production.
+Dify is an open-source LLM app development platform. Its intuitive interface combines AI workflow, RAG pipeline, agent capabilities, model management, observability features (including [Opik](https://www.comet.com/docs/opik/integrations/dify), [Langfuse](https://docs.langfuse.com), and [Arize Phoenix](https://docs.arize.com/phoenix)) and more, letting you quickly go from prototype to production. Here's a list of the core features:
 
 ## Quick start
 
@@ -133,7 +133,7 @@ Star Dify on GitHub and be instantly notified of new releases.
 
 ### Custom configurations
 
-If you need to customize the configuration, please refer to the comments in our [.env.example](docker/.env.example) file and update the corresponding values in your `.env` file. Additionally, you might need to make adjustments to the `docker-compose.yaml` file itself, such as changing image versions, port mappings, or volume mounts, based on your specific deployment environment and requirements. After making any changes, please re-run `docker-compose up -d`. You can find the full list of available environment variables [here](https://docs.dify.ai/getting-started/install-self-hosted/environments).
+If you need to customize the configuration, please refer to the comments in our [.env.example](docker/.env.example) file and update the corresponding values in your `.env` file. Additionally, you might need to make adjustments to the `docker-compose.yaml` file itself, such as changing image versions, port mappings, or volume mounts, based on your specific deployment environment and requirements. After making any changes, please re-run `docker compose up -d`. You can find the full list of available environment variables [here](https://docs.dify.ai/getting-started/install-self-hosted/environments).
 
 #### Customizing Suggested Questions
 

api/.env.example

@@ -42,6 +42,8 @@ REFRESH_TOKEN_EXPIRE_DAYS=30
 # redis configuration
 REDIS_HOST=localhost
 REDIS_PORT=6379
+# Optional: limit total connections in connection pool (unset for default)
+# REDIS_MAX_CONNECTIONS=200
 REDIS_USERNAME=
 REDIS_PASSWORD=difyai123456
 REDIS_USE_SSL=false

api/.importlinter

@@ -28,17 +28,8 @@ ignore_imports =
     dify_graph.nodes.iteration.iteration_node -> dify_graph.graph_events
     dify_graph.nodes.loop.loop_node -> dify_graph.graph_events
 
-    dify_graph.nodes.iteration.iteration_node -> core.workflow.node_factory
-    dify_graph.nodes.loop.loop_node -> core.workflow.node_factory
-    dify_graph.nodes.iteration.iteration_node -> core.app.workflow.layers.llm_quota
-    dify_graph.nodes.loop.loop_node -> core.app.workflow.layers.llm_quota
-
     dify_graph.nodes.iteration.iteration_node -> dify_graph.graph_engine
-    dify_graph.nodes.iteration.iteration_node -> dify_graph.graph
-    dify_graph.nodes.iteration.iteration_node -> dify_graph.graph_engine.command_channels
     dify_graph.nodes.loop.loop_node -> dify_graph.graph_engine
-    dify_graph.nodes.loop.loop_node -> dify_graph.graph
-    dify_graph.nodes.loop.loop_node -> dify_graph.graph_engine.command_channels
     # TODO(QuantumGhost): fix the import violation later
     dify_graph.entities.pause_reason -> dify_graph.nodes.human_input.entities
 
@@ -53,13 +44,9 @@ forbidden_modules =
 allow_indirect_imports = True
 ignore_imports =
     dify_graph.nodes.agent.agent_node -> extensions.ext_database
-    dify_graph.nodes.llm.file_saver -> extensions.ext_database
     dify_graph.nodes.llm.node -> extensions.ext_database
-    dify_graph.nodes.tool.tool_node -> extensions.ext_database
     dify_graph.model_runtime.model_providers.__base.ai_model -> extensions.ext_redis
     dify_graph.model_runtime.model_providers.model_provider_factory -> extensions.ext_redis
-    # TODO(QuantumGhost): use DI to avoid depending on global DB.
-    dify_graph.nodes.human_input.human_input_node -> extensions.ext_database
 
 [importlinter:contract:workflow-external-imports]
 name = Workflow External Imports
@@ -103,13 +90,9 @@ forbidden_modules =
     core.trigger
     core.variables
 ignore_imports =
-    dify_graph.nodes.loop.loop_node -> core.workflow.node_factory
     dify_graph.nodes.agent.agent_node -> core.model_manager
     dify_graph.nodes.agent.agent_node -> core.provider_manager
     dify_graph.nodes.agent.agent_node -> core.tools.tool_manager
-    dify_graph.nodes.document_extractor.node -> core.helper.ssrf_proxy
-    dify_graph.nodes.iteration.iteration_node -> core.workflow.node_factory
-    dify_graph.nodes.iteration.iteration_node -> core.app.workflow.layers.llm_quota
     dify_graph.nodes.llm.llm_utils -> core.model_manager
     dify_graph.nodes.llm.protocols -> core.model_manager
     dify_graph.nodes.llm.llm_utils -> dify_graph.model_runtime.model_providers.__base.large_language_model
@@ -127,9 +110,7 @@ ignore_imports =
     dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.model_manager
     dify_graph.nodes.question_classifier.question_classifier_node -> core.model_manager
     dify_graph.nodes.tool.tool_node -> core.tools.utils.message_transformer
-    dify_graph.nodes.tool.tool_node -> models
     dify_graph.nodes.agent.agent_node -> models.model
-    dify_graph.nodes.llm.file_saver -> core.helper.ssrf_proxy
     dify_graph.nodes.llm.node -> core.helper.code_executor
     dify_graph.nodes.llm.node -> core.llm_generator.output_parser.errors
     dify_graph.nodes.llm.node -> core.llm_generator.output_parser.structured_output
@@ -150,13 +131,8 @@ ignore_imports =
     dify_graph.nodes.llm.file_saver -> core.tools.tool_file_manager
     dify_graph.nodes.tool.tool_node -> core.tools.errors
     dify_graph.nodes.agent.agent_node -> extensions.ext_database
-    dify_graph.nodes.llm.file_saver -> extensions.ext_database
     dify_graph.nodes.llm.node -> extensions.ext_database
-    dify_graph.nodes.tool.tool_node -> extensions.ext_database
-    dify_graph.nodes.human_input.human_input_node -> extensions.ext_database
-    dify_graph.nodes.human_input.human_input_node -> core.repositories.human_input_repository
     dify_graph.nodes.agent.agent_node -> models
-    dify_graph.nodes.loop.loop_node -> core.app.workflow.layers.llm_quota
     dify_graph.nodes.llm.node -> models.model
     dify_graph.nodes.agent.agent_node -> services
     dify_graph.nodes.tool.tool_node -> services

api/AGENTS.md

@@ -62,6 +62,22 @@ This is the default standard for backend code in this repo. Follow it for new co
 
 - Code should usually include type annotations that match the repo’s current Python version (avoid untyped public APIs and “mystery” values).
 - Prefer modern typing forms (e.g. `list[str]`, `dict[str, int]`) and avoid `Any` unless there’s a strong reason.
+- For dictionary-like data with known keys and value types, prefer `TypedDict` over `dict[...]` or `Mapping[...]`.
+- For optional keys in typed payloads, use `NotRequired[...]` (or `total=False` when most fields are optional).
+- Keep `dict[...]` / `Mapping[...]` for truly dynamic key spaces where the key set is unknown.
+
+```python
+from datetime import datetime
+from typing import NotRequired, TypedDict
+
+
+class UserProfile(TypedDict):
+    user_id: str
+    email: str
+    created_at: datetime
+    nickname: NotRequired[str]
+```
+
 - For classes, declare member variables at the top of the class body (before `__init__`) so the class shape is obvious at a glance:
 
 ```python

api/commands.py

@@ -30,6 +30,7 @@ from extensions.ext_redis import redis_client
 from extensions.ext_storage import storage
 from extensions.storage.opendal_storage import OpenDALStorage
 from extensions.storage.storage_type import StorageType
+from libs.datetime_utils import naive_utc_now
 from libs.db_migration_lock import DbMigrationAutoRenewLock
 from libs.helper import email as email_validate
 from libs.password import hash_password, password_pattern, valid_password
@@ -2598,15 +2599,29 @@ def migrate_oss(
 @click.option(
     "--start-from",
     type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
-    required=True,
+    required=False,
+    default=None,
     help="Lower bound (inclusive) for created_at.",
 )
 @click.option(
     "--end-before",
     type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
-    required=True,
+    required=False,
+    default=None,
     help="Upper bound (exclusive) for created_at.",
 )
+@click.option(
+    "--from-days-ago",
+    type=int,
+    default=None,
+    help="Relative lower bound in days ago (inclusive). Must be used with --before-days.",
+)
+@click.option(
+    "--before-days",
+    type=int,
+    default=None,
+    help="Relative upper bound in days ago (exclusive). Required for relative mode.",
+)
 @click.option("--batch-size", default=1000, show_default=True, help="Batch size for selecting messages.")
 @click.option(
     "--graceful-period",
@@ -2618,8 +2633,10 @@ def migrate_oss(
 def clean_expired_messages(
     batch_size: int,
     graceful_period: int,
-    start_from: datetime.datetime,
-    end_before: datetime.datetime,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime | None,
+    from_days_ago: int | None,
+    before_days: int | None,
     dry_run: bool,
 ):
     """
@@ -2630,18 +2647,70 @@ def clean_expired_messages(
     start_at = time.perf_counter()
 
     try:
+        abs_mode = start_from is not None and end_before is not None
+        rel_mode = before_days is not None
+
+        if abs_mode and rel_mode:
+            raise click.UsageError(
+                "Options are mutually exclusive: use either (--start-from,--end-before) "
+                "or (--from-days-ago,--before-days)."
+            )
+
+        if from_days_ago is not None and before_days is None:
+            raise click.UsageError("--from-days-ago must be used together with --before-days.")
+
+        if (start_from is None) ^ (end_before is None):
+            raise click.UsageError("Both --start-from and --end-before are required when using absolute time range.")
+
+        if not abs_mode and not rel_mode:
+            raise click.UsageError(
+                "You must provide either (--start-from,--end-before) or (--before-days [--from-days-ago])."
+            )
+
+        if rel_mode:
+            assert before_days is not None
+            if before_days < 0:
+                raise click.UsageError("--before-days must be >= 0.")
+            if from_days_ago is not None:
+                if from_days_ago < 0:
+                    raise click.UsageError("--from-days-ago must be >= 0.")
+                if from_days_ago <= before_days:
+                    raise click.UsageError("--from-days-ago must be greater than --before-days.")
+
         # Create policy based on billing configuration
         # NOTE: graceful_period will be ignored when billing is disabled.
         policy = create_message_clean_policy(graceful_period_days=graceful_period)
 
         # Create and run the cleanup service
-        service = MessagesCleanService.from_time_range(
-            policy=policy,
-            start_from=start_from,
-            end_before=end_before,
-            batch_size=batch_size,
-            dry_run=dry_run,
-        )
+        if abs_mode:
+            assert start_from is not None
+            assert end_before is not None
+            service = MessagesCleanService.from_time_range(
+                policy=policy,
+                start_from=start_from,
+                end_before=end_before,
+                batch_size=batch_size,
+                dry_run=dry_run,
+            )
+        elif from_days_ago is None:
+            assert before_days is not None
+            service = MessagesCleanService.from_days(
+                policy=policy,
+                days=before_days,
+                batch_size=batch_size,
+                dry_run=dry_run,
+            )
+        else:
+            assert before_days is not None
+            assert from_days_ago is not None
+            now = naive_utc_now()
+            service = MessagesCleanService.from_time_range(
+                policy=policy,
+                start_from=now - datetime.timedelta(days=from_days_ago),
+                end_before=now - datetime.timedelta(days=before_days),
+                batch_size=batch_size,
+                dry_run=dry_run,
+            )
         stats = service.run()
 
         end_at = time.perf_counter()
@@ -2668,3 +2737,77 @@ def clean_expired_messages(
         raise
 
     click.echo(click.style("messages cleanup completed.", fg="green"))
+
+
+@click.command("export-app-messages", help="Export messages for an app to JSONL.GZ.")
+@click.option("--app-id", required=True, help="Application ID to export messages for.")
+@click.option(
+    "--start-from",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    default=None,
+    help="Optional lower bound (inclusive) for created_at.",
+)
+@click.option(
+    "--end-before",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    required=True,
+    help="Upper bound (exclusive) for created_at.",
+)
+@click.option(
+    "--filename",
+    required=True,
+    help="Base filename (relative path). Do not include suffix like .jsonl.gz.",
+)
+@click.option("--use-cloud-storage", is_flag=True, default=False, help="Upload to cloud storage instead of local file.")
+@click.option("--batch-size", default=1000, show_default=True, help="Batch size for cursor pagination.")
+@click.option("--dry-run", is_flag=True, default=False, help="Scan only, print stats without writing any file.")
+def export_app_messages(
+    app_id: str,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime,
+    filename: str,
+    use_cloud_storage: bool,
+    batch_size: int,
+    dry_run: bool,
+):
+    if start_from and start_from >= end_before:
+        raise click.UsageError("--start-from must be before --end-before.")
+
+    from services.retention.conversation.message_export_service import AppMessageExportService
+
+    try:
+        validated_filename = AppMessageExportService.validate_export_filename(filename)
+    except ValueError as e:
+        raise click.BadParameter(str(e), param_hint="--filename") from e
+
+    click.echo(click.style(f"export_app_messages: starting export for app {app_id}.", fg="green"))
+    start_at = time.perf_counter()
+
+    try:
+        service = AppMessageExportService(
+            app_id=app_id,
+            end_before=end_before,
+            filename=validated_filename,
+            start_from=start_from,
+            batch_size=batch_size,
+            use_cloud_storage=use_cloud_storage,
+            dry_run=dry_run,
+        )
+        stats = service.run()
+
+        elapsed = time.perf_counter() - start_at
+        click.echo(
+            click.style(
+                f"export_app_messages: completed in {elapsed:.2f}s\n"
+                f"  - Batches: {stats.batches}\n"
+                f"  - Total messages: {stats.total_messages}\n"
+                f"  - Messages with feedback: {stats.messages_with_feedback}\n"
+                f"  - Total feedbacks: {stats.total_feedbacks}",
+                fg="green",
+            )
+        )
+    except Exception as e:
+        elapsed = time.perf_counter() - start_at
+        logger.exception("export_app_messages failed")
+        click.echo(click.style(f"export_app_messages: failed after {elapsed:.2f}s - {e}", fg="red"))
+        raise

api/configs/enterprise/__init__.py

@@ -18,3 +18,7 @@ class EnterpriseFeatureConfig(BaseSettings):
         description="Allow customization of the enterprise logo.",
         default=False,
     )
+
+    ENTERPRISE_REQUEST_TIMEOUT: int = Field(
+        ge=1, description="Maximum timeout in seconds for enterprise requests", default=5
+    )

api/configs/middleware/cache/redis_config.py

@@ -111,3 +111,8 @@ class RedisConfig(BaseSettings):
         description="Enable client side cache in redis",
         default=False,
     )
+
+    REDIS_MAX_CONNECTIONS: PositiveInt | None = Field(
+        description="Maximum connections in the Redis connection pool (unset for library default)",
+        default=None,
+    )

api/configs/middleware/cache/redis_pubsub_config.py

@@ -1,7 +1,7 @@
 from typing import Literal, Protocol
 from urllib.parse import quote_plus, urlunparse
 
-from pydantic import Field
+from pydantic import AliasChoices, Field
 from pydantic_settings import BaseSettings
 
 
@@ -23,41 +23,56 @@ class RedisConfigDefaultsMixin:
 
 class RedisPubSubConfig(BaseSettings, RedisConfigDefaultsMixin):
     """
-    Configuration settings for Redis pub/sub streaming.
+    Configuration settings for event transport between API and workers.
+
+    Supported transports:
+    - pubsub: Redis PUBLISH/SUBSCRIBE (at-most-once)
+    - sharded: Redis 7+ Sharded Pub/Sub (at-most-once, better scaling)
+    - streams: Redis Streams (at-least-once, supports late subscribers)
     """
 
     PUBSUB_REDIS_URL: str | None = Field(
-        alias="PUBSUB_REDIS_URL",
+        validation_alias=AliasChoices("EVENT_BUS_REDIS_URL", "PUBSUB_REDIS_URL"),
         description=(
-            "Redis connection URL for pub/sub streaming events between API "
-            "and celery worker, defaults to url constructed from "
-            "`REDIS_*` configurations"
+            "Redis connection URL for streaming events between API and celery worker; "
+            "defaults to URL constructed from `REDIS_*` configurations. Also accepts ENV: EVENT_BUS_REDIS_URL."
         ),
         default=None,
     )
 
     PUBSUB_REDIS_USE_CLUSTERS: bool = Field(
+        validation_alias=AliasChoices("EVENT_BUS_REDIS_CLUSTERS", "PUBSUB_REDIS_USE_CLUSTERS"),
         description=(
-            "Enable Redis Cluster mode for pub/sub streaming. It's highly "
-            "recommended to enable this for large deployments."
+            "Enable Redis Cluster mode for pub/sub or streams transport. Recommended for large deployments. "
+            "Also accepts ENV: EVENT_BUS_REDIS_CLUSTERS."
         ),
         default=False,
     )
 
-    PUBSUB_REDIS_CHANNEL_TYPE: Literal["pubsub", "sharded"] = Field(
+    PUBSUB_REDIS_CHANNEL_TYPE: Literal["pubsub", "sharded", "streams"] = Field(
+        validation_alias=AliasChoices("EVENT_BUS_REDIS_CHANNEL_TYPE", "PUBSUB_REDIS_CHANNEL_TYPE"),
         description=(
-            "Pub/sub channel type for streaming events. "
-            "Valid options are:\n"
-            "\n"
-            " - pubsub: for normal Pub/Sub\n"
-            " - sharded: for sharded Pub/Sub\n"
-            "\n"
-            "It's highly recommended to use sharded Pub/Sub AND redis cluster "
-            "for large deployments."
+            "Event transport type. Options are:\n\n"
+            " - pubsub: normal Pub/Sub (at-most-once)\n"
+            " - sharded: sharded Pub/Sub (at-most-once)\n"
+            " - streams: Redis Streams (at-least-once, recommended to avoid subscriber races)\n\n"
+            "Note: Before enabling 'streams' in production, estimate your expected event volume and retention needs.\n"
+            "Configure Redis memory limits and stream trimming appropriately (e.g., MAXLEN and key expiry) to reduce\n"
+            "the risk of data loss from Redis auto-eviction under memory pressure.\n"
+            "Also accepts ENV: EVENT_BUS_REDIS_CHANNEL_TYPE."
         ),
         default="pubsub",
     )
 
+    PUBSUB_STREAMS_RETENTION_SECONDS: int = Field(
+        validation_alias=AliasChoices("EVENT_BUS_STREAMS_RETENTION_SECONDS", "PUBSUB_STREAMS_RETENTION_SECONDS"),
+        description=(
+            "When using 'streams', expire each stream key this many seconds after the last event is published. "
+            "Also accepts ENV: EVENT_BUS_STREAMS_RETENTION_SECONDS."
+        ),
+        default=600,
+    )
+
     def _build_default_pubsub_url(self) -> str:
         defaults = self._redis_defaults()
         if not defaults.REDIS_HOST or not defaults.REDIS_PORT:

api/controllers/console/datasets/datasets.py

@@ -807,7 +807,7 @@ class DatasetApiKeyApi(Resource):
             console_ns.abort(
                 400,
                 message=f"Cannot create more than {self.max_keys} API keys for this resource type.",
-                code="max_keys_exceeded",
+                custom="max_keys_exceeded",
             )
 
         key = ApiToken.generate_api_key(self.token_prefix, 24)

api/controllers/console/explore/parameter.py

@@ -1,3 +1,5 @@
+from typing import Any, cast
+
 from controllers.common import fields
 from controllers.console import console_ns
 from controllers.console.app.error import AppUnavailableError
@@ -23,14 +25,14 @@ class AppParameterApi(InstalledAppResource):
             if workflow is None:
                 raise AppUnavailableError()
 
-            features_dict = workflow.features_dict
+            features_dict: dict[str, Any] = workflow.features_dict
             user_input_form = workflow.user_input_form(to_old_structure=True)
         else:
             app_model_config = app_model.app_model_config
             if app_model_config is None:
                 raise AppUnavailableError()
 
-            features_dict = app_model_config.to_dict()
+            features_dict = cast(dict[str, Any], app_model_config.to_dict())
 
             user_input_form = features_dict.get("user_input_form", [])
 

api/controllers/files/tool_files.py

@@ -10,7 +10,6 @@ from controllers.common.file_response import enforce_download_for_html
 from controllers.files import files_ns
 from core.tools.signature import verify_tool_file_signature
 from core.tools.tool_file_manager import ToolFileManager
-from extensions.ext_database import db as global_db
 
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
@@ -57,7 +56,7 @@ class ToolFileApi(Resource):
             raise Forbidden("Invalid request.")
 
         try:
-            tool_file_manager = ToolFileManager(engine=global_db.engine)
+            tool_file_manager = ToolFileManager()
             stream, tool_file = tool_file_manager.get_file_generator_by_tool_file_id(
                 file_id,
             )

api/controllers/service_api/app/annotation.py

@@ -185,4 +185,4 @@ class AnnotationUpdateDeleteApi(Resource):
     def delete(self, app_model: App, annotation_id: str):
         """Delete an annotation."""
         AppAnnotationService.delete_app_annotation(app_model.id, annotation_id)
-        return {"result": "success"}, 204
+        return "", 204

api/controllers/service_api/app/app.py

@@ -1,3 +1,5 @@
+from typing import Any, cast
+
 from flask_restx import Resource
 
 from controllers.common.fields import Parameters
@@ -33,14 +35,14 @@ class AppParameterApi(Resource):
             if workflow is None:
                 raise AppUnavailableError()
 
-            features_dict = workflow.features_dict
+            features_dict: dict[str, Any] = workflow.features_dict
             user_input_form = workflow.user_input_form(to_old_structure=True)
         else:
             app_model_config = app_model.app_model_config
             if app_model_config is None:
                 raise AppUnavailableError()
 
-            features_dict = app_model_config.to_dict()
+            features_dict = cast(dict[str, Any], app_model_config.to_dict())
 
             user_input_form = features_dict.get("user_input_form", [])
 

api/controllers/service_api/app/conversation.py

@@ -14,7 +14,6 @@ from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate
 from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
 from fields.conversation_fields import (
-    ConversationDelete,
     ConversationInfiniteScrollPagination,
     SimpleConversation,
 )
@@ -163,7 +162,7 @@ class ConversationDetailApi(Resource):
             ConversationService.delete(app_model, conversation_id, end_user)
         except services.errors.conversation.ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
-        return ConversationDelete(result="success").model_dump(mode="json"), 204
+        return "", 204
 
 
 @service_api_ns.route("/conversations/<uuid:c_id>/name")

api/controllers/service_api/app/workflow.py

@@ -132,6 +132,8 @@ class WorkflowRunDetailApi(Resource):
             app_id=app_model.id,
             run_id=workflow_run_id,
         )
+        if not workflow_run:
+            raise NotFound("Workflow run not found.")
         return workflow_run
 
 

api/controllers/service_api/dataset/document.py

@@ -1,8 +1,9 @@
 import json
+from contextlib import ExitStack
 from typing import Self
 from uuid import UUID
 
-from flask import request
+from flask import request, send_file
 from flask_restx import marshal
 from pydantic import BaseModel, Field, field_validator, model_validator
 from sqlalchemy import desc, select
@@ -100,6 +101,15 @@ class DocumentListQuery(BaseModel):
     status: str | None = Field(default=None, description="Document status filter")
 
 
+DOCUMENT_BATCH_DOWNLOAD_ZIP_MAX_DOCS = 100
+
+
+class DocumentBatchDownloadZipPayload(BaseModel):
+    """Request payload for bulk downloading uploaded documents as a ZIP archive."""
+
+    document_ids: list[UUID] = Field(..., min_length=1, max_length=DOCUMENT_BATCH_DOWNLOAD_ZIP_MAX_DOCS)
+
+
 register_enum_models(service_api_ns, RetrievalMethod)
 
 register_schema_models(
@@ -109,6 +119,7 @@ register_schema_models(
     DocumentTextCreatePayload,
     DocumentTextUpdate,
     DocumentListQuery,
+    DocumentBatchDownloadZipPayload,
     Rule,
     PreProcessingRule,
     Segmentation,
@@ -540,6 +551,46 @@ class DocumentListApi(DatasetApiResource):
         return response
 
 
+@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/download-zip")
+class DocumentBatchDownloadZipApi(DatasetApiResource):
+    """Download multiple uploaded-file documents as a single ZIP archive."""
+
+    @service_api_ns.expect(service_api_ns.models[DocumentBatchDownloadZipPayload.__name__])
+    @service_api_ns.doc("download_documents_as_zip")
+    @service_api_ns.doc(description="Download selected uploaded documents as a single ZIP archive")
+    @service_api_ns.doc(params={"dataset_id": "Dataset ID"})
+    @service_api_ns.doc(
+        responses={
+            200: "ZIP archive generated successfully",
+            401: "Unauthorized - invalid API token",
+            403: "Forbidden - insufficient permissions",
+            404: "Document or dataset not found",
+        }
+    )
+    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
+    def post(self, tenant_id, dataset_id):
+        payload = DocumentBatchDownloadZipPayload.model_validate(service_api_ns.payload or {})
+
+        upload_files, download_name = DocumentService.prepare_document_batch_download_zip(
+            dataset_id=str(dataset_id),
+            document_ids=[str(document_id) for document_id in payload.document_ids],
+            tenant_id=str(tenant_id),
+            current_user=current_user,
+        )
+
+        with ExitStack() as stack:
+            zip_path = stack.enter_context(FileService.build_upload_files_zip_tempfile(upload_files=upload_files))
+            response = send_file(
+                zip_path,
+                mimetype="application/zip",
+                as_attachment=True,
+                download_name=download_name,
+            )
+            cleanup = stack.pop_all()
+            response.call_on_close(cleanup.close)
+        return response
+
+
 @service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<string:batch>/indexing-status")
 class DocumentIndexingStatusApi(DatasetApiResource):
     @service_api_ns.doc("get_document_indexing_status")
@@ -600,6 +651,35 @@ class DocumentIndexingStatusApi(DatasetApiResource):
         return data
 
 
+@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/download")
+class DocumentDownloadApi(DatasetApiResource):
+    """Return a signed download URL for a document's original uploaded file."""
+
+    @service_api_ns.doc("get_document_download_url")
+    @service_api_ns.doc(description="Get a signed download URL for a document's original uploaded file")
+    @service_api_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
+    @service_api_ns.doc(
+        responses={
+            200: "Download URL generated successfully",
+            401: "Unauthorized - invalid API token",
+            403: "Forbidden - insufficient permissions",
+            404: "Document or upload file not found",
+        }
+    )
+    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
+    def get(self, tenant_id, dataset_id, document_id):
+        dataset = self.get_dataset(str(dataset_id), str(tenant_id))
+        document = DocumentService.get_document(dataset.id, str(document_id))
+
+        if not document:
+            raise NotFound("Document not found.")
+
+        if document.tenant_id != str(tenant_id):
+            raise Forbidden("No permission.")
+
+        return {"url": DocumentService.get_document_download_url(document)}
+
+
 @service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
 class DocumentApi(DatasetApiResource):
     METADATA_CHOICES = {"all", "only", "without"}

api/controllers/web/app.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Any, cast
 
 from flask import request
 from flask_restx import Resource
@@ -57,14 +58,14 @@ class AppParameterApi(WebApiResource):
             if workflow is None:
                 raise AppUnavailableError()
 
-            features_dict = workflow.features_dict
+            features_dict: dict[str, Any] = workflow.features_dict
             user_input_form = workflow.user_input_form(to_old_structure=True)
         else:
             app_model_config = app_model.app_model_config
             if app_model_config is None:
                 raise AppUnavailableError()
 
-            features_dict = app_model_config.to_dict()
+            features_dict = cast(dict[str, Any], app_model_config.to_dict())
 
             user_input_form = features_dict.get("user_input_form", [])
 

api/controllers/web/message.py

@@ -239,7 +239,7 @@ class MessageSuggestedQuestionApi(WebApiResource):
     def get(self, app_model, end_user, message_id):
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
-            raise NotCompletionAppError()
+            raise NotChatAppError()
 
         message_id = str(message_id)
 

api/core/app/app_config/common/sensitive_word_avoidance/manager.py

@@ -1,10 +1,13 @@
+from collections.abc import Mapping
+from typing import Any
+
 from core.app.app_config.entities import SensitiveWordAvoidanceEntity
 from core.moderation.factory import ModerationFactory
 
 
 class SensitiveWordAvoidanceConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> SensitiveWordAvoidanceEntity | None:
+    def convert(cls, config: Mapping[str, Any]) -> SensitiveWordAvoidanceEntity | None:
         sensitive_word_avoidance_dict = config.get("sensitive_word_avoidance")
         if not sensitive_word_avoidance_dict:
             return None
@@ -12,7 +15,7 @@ class SensitiveWordAvoidanceConfigManager:
         if sensitive_word_avoidance_dict.get("enabled"):
             return SensitiveWordAvoidanceEntity(
                 type=sensitive_word_avoidance_dict.get("type"),
-                config=sensitive_word_avoidance_dict.get("config"),
+                config=sensitive_word_avoidance_dict.get("config", {}),
             )
         else:
             return None

api/core/app/app_config/easy_ui_based_app/agent/manager.py

@@ -1,10 +1,13 @@
+from typing import Any, cast
+
 from core.agent.entities import AgentEntity, AgentPromptEntity, AgentToolEntity
 from core.agent.prompt.template import REACT_PROMPT_TEMPLATES
+from models.model import AppModelConfigDict
 
 
 class AgentConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> AgentEntity | None:
+    def convert(cls, config: AppModelConfigDict) -> AgentEntity | None:
         """
         Convert model config to model config
 
@@ -28,17 +31,17 @@ class AgentConfigManager:
 
             agent_tools = []
             for tool in agent_dict.get("tools", []):
-                keys = tool.keys()
-                if len(keys) >= 4:
-                    if "enabled" not in tool or not tool["enabled"]:
+                tool_dict = cast(dict[str, Any], tool)
+                if len(tool_dict) >= 4:
+                    if "enabled" not in tool_dict or not tool_dict["enabled"]:
                         continue
 
                     agent_tool_properties = {
-                        "provider_type": tool["provider_type"],
-                        "provider_id": tool["provider_id"],
-                        "tool_name": tool["tool_name"],
-                        "tool_parameters": tool.get("tool_parameters", {}),
-                        "credential_id": tool.get("credential_id", None),
+                        "provider_type": tool_dict["provider_type"],
+                        "provider_id": tool_dict["provider_id"],
+                        "tool_name": tool_dict["tool_name"],
+                        "tool_parameters": tool_dict.get("tool_parameters", {}),
+                        "credential_id": tool_dict.get("credential_id", None),
                     }
 
                     agent_tools.append(AgentToolEntity.model_validate(agent_tool_properties))
@@ -47,7 +50,8 @@ class AgentConfigManager:
                 "react_router",
                 "router",
             }:
-                agent_prompt = agent_dict.get("prompt", None) or {}
+                agent_prompt_raw = agent_dict.get("prompt", None)
+                agent_prompt: dict[str, Any] = agent_prompt_raw if isinstance(agent_prompt_raw, dict) else {}
                 # check model mode
                 model_mode = config.get("model", {}).get("mode", "completion")
                 if model_mode == "completion":
@@ -75,7 +79,7 @@ class AgentConfigManager:
                     strategy=strategy,
                     prompt=agent_prompt_entity,
                     tools=agent_tools,
-                    max_iteration=agent_dict.get("max_iteration", 10),
+                    max_iteration=cast(int, agent_dict.get("max_iteration", 10)),
                 )
 
         return None

api/core/app/app_config/easy_ui_based_app/dataset/manager.py

@@ -1,5 +1,5 @@
 import uuid
-from typing import Literal, cast
+from typing import Any, Literal, cast
 
 from core.app.app_config.entities import (
     DatasetEntity,
@@ -8,13 +8,13 @@ from core.app.app_config.entities import (
     ModelConfig,
 )
 from core.entities.agent_entities import PlanningStrategy
-from models.model import AppMode
+from models.model import AppMode, AppModelConfigDict
 from services.dataset_service import DatasetService
 
 
 class DatasetConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> DatasetEntity | None:
+    def convert(cls, config: AppModelConfigDict) -> DatasetEntity | None:
         """
         Convert model config to model config
 
@@ -25,11 +25,15 @@ class DatasetConfigManager:
             datasets = config.get("dataset_configs", {}).get("datasets", {"strategy": "router", "datasets": []})
 
             for dataset in datasets.get("datasets", []):
+                if not isinstance(dataset, dict):
+                    continue
                 keys = list(dataset.keys())
                 if len(keys) == 0 or keys[0] != "dataset":
                     continue
 
                 dataset = dataset["dataset"]
+                if not isinstance(dataset, dict):
+                    continue
 
                 if "enabled" not in dataset or not dataset["enabled"]:
                     continue
@@ -47,15 +51,14 @@ class DatasetConfigManager:
             agent_dict = config.get("agent_mode", {})
 
             for tool in agent_dict.get("tools", []):
-                keys = tool.keys()
-                if len(keys) == 1:
+                if len(tool) == 1:
                     # old standard
                     key = list(tool.keys())[0]
 
                     if key != "dataset":
                         continue
 
-                    tool_item = tool[key]
+                    tool_item = cast(dict[str, Any], tool)[key]
 
                     if "enabled" not in tool_item or not tool_item["enabled"]:
                         continue

api/core/app/app_config/easy_ui_based_app/model_config/manager.py

@@ -5,12 +5,13 @@ from core.app.app_config.entities import ModelConfigEntity
 from core.provider_manager import ProviderManager
 from dify_graph.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
 from dify_graph.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
+from models.model import AppModelConfigDict
 from models.provider_ids import ModelProviderID
 
 
 class ModelConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> ModelConfigEntity:
+    def convert(cls, config: AppModelConfigDict) -> ModelConfigEntity:
         """
         Convert model config to model config
 
@@ -22,7 +23,7 @@ class ModelConfigManager:
         if not model_config:
             raise ValueError("model is required")
 
-        completion_params = model_config.get("completion_params")
+        completion_params = model_config.get("completion_params") or {}
         stop = []
         if "stop" in completion_params:
             stop = completion_params["stop"]

api/core/app/app_config/easy_ui_based_app/prompt_template/manager.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from core.app.app_config.entities import (
     AdvancedChatMessageEntity,
     AdvancedChatPromptTemplateEntity,
@@ -6,12 +8,12 @@ from core.app.app_config.entities import (
 )
 from core.prompt.simple_prompt_transform import ModelMode
 from dify_graph.model_runtime.entities.message_entities import PromptMessageRole
-from models.model import AppMode
+from models.model import AppMode, AppModelConfigDict
 
 
 class PromptTemplateConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> PromptTemplateEntity:
+    def convert(cls, config: AppModelConfigDict) -> PromptTemplateEntity:
         if not config.get("prompt_type"):
             raise ValueError("prompt_type is required")
 
@@ -40,14 +42,15 @@ class PromptTemplateConfigManager:
             advanced_completion_prompt_template = None
             completion_prompt_config = config.get("completion_prompt_config", {})
             if completion_prompt_config:
-                completion_prompt_template_params = {
+                completion_prompt_template_params: dict[str, Any] = {
                     "prompt": completion_prompt_config["prompt"]["text"],
                 }
 
-                if "conversation_histories_role" in completion_prompt_config:
+                conv_role = completion_prompt_config.get("conversation_histories_role")
+                if conv_role:
                     completion_prompt_template_params["role_prefix"] = {
-                        "user": completion_prompt_config["conversation_histories_role"]["user_prefix"],
-                        "assistant": completion_prompt_config["conversation_histories_role"]["assistant_prefix"],
+                        "user": conv_role["user_prefix"],
+                        "assistant": conv_role["assistant_prefix"],
                     }
 
                 advanced_completion_prompt_template = AdvancedCompletionPromptTemplateEntity(

api/core/app/app_config/easy_ui_based_app/variables/manager.py

@@ -1,8 +1,10 @@
 import re
+from typing import cast
 
 from core.app.app_config.entities import ExternalDataVariableEntity
 from core.external_data_tool.factory import ExternalDataToolFactory
 from dify_graph.variables.input_entities import VariableEntity, VariableEntityType
+from models.model import AppModelConfigDict
 
 _ALLOWED_VARIABLE_ENTITY_TYPE = frozenset(
     [
@@ -18,7 +20,7 @@ _ALLOWED_VARIABLE_ENTITY_TYPE = frozenset(
 
 class BasicVariablesConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> tuple[list[VariableEntity], list[ExternalDataVariableEntity]]:
+    def convert(cls, config: AppModelConfigDict) -> tuple[list[VariableEntity], list[ExternalDataVariableEntity]]:
         """
         Convert model config to model config
 
@@ -51,7 +53,9 @@ class BasicVariablesConfigManager:
 
                 external_data_variables.append(
                     ExternalDataVariableEntity(
-                        variable=variable["variable"], type=variable["type"], config=variable["config"]
+                        variable=variable["variable"],
+                        type=variable.get("type", ""),
+                        config=variable.get("config", {}),
                     )
                 )
             elif variable_type in {
@@ -64,10 +68,10 @@ class BasicVariablesConfigManager:
                 variable = variables[variable_type]
                 variable_entities.append(
                     VariableEntity(
-                        type=variable_type,
-                        variable=variable.get("variable"),
+                        type=cast(VariableEntityType, variable_type),
+                        variable=variable["variable"],
                         description=variable.get("description") or "",
-                        label=variable.get("label"),
+                        label=variable["label"],
                         required=variable.get("required", False),
                         max_length=variable.get("max_length"),
                         options=variable.get("options") or [],

api/core/app/app_config/entities.py

@@ -281,7 +281,7 @@ class EasyUIBasedAppConfig(AppConfig):
 
     app_model_config_from: EasyUIBasedAppModelConfigFrom
     app_model_config_id: str
-    app_model_config_dict: dict
+    app_model_config_dict: dict[str, Any]
     model: ModelConfigEntity
     prompt_template: PromptTemplateEntity
     dataset: DatasetEntity | None = None

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

@@ -516,8 +516,10 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             graph_runtime_state=validated_state,
         )
 
+        yield from self._handle_advanced_chat_message_end_event(
+            QueueAdvancedChatMessageEndEvent(), graph_runtime_state=validated_state
+        )
         yield workflow_finish_resp
-        self._base_task_pipeline.queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
 
     def _handle_workflow_partial_success_event(
         self,
@@ -538,6 +540,9 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             exceptions_count=event.exceptions_count,
         )
 
+        yield from self._handle_advanced_chat_message_end_event(
+            QueueAdvancedChatMessageEndEvent(), graph_runtime_state=validated_state
+        )
         yield workflow_finish_resp
 
     def _handle_workflow_paused_event(
@@ -735,7 +740,6 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
 
     def _load_human_input_form_id(self, *, node_id: str) -> str | None:
         form_repository = HumanInputFormRepositoryImpl(
-            session_factory=db.engine,
             tenant_id=self._workflow_tenant_id,
         )
         form = form_repository.get_form(self._workflow_run_id, node_id)
@@ -855,6 +859,14 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                     yield from self._handle_workflow_paused_event(event)
                     break
 
+                case QueueWorkflowSucceededEvent():
+                    yield from self._handle_workflow_succeeded_event(event, trace_manager=trace_manager)
+                    break
+
+                case QueueWorkflowPartialSuccessEvent():
+                    yield from self._handle_workflow_partial_success_event(event, trace_manager=trace_manager)
+                    break
+
                 case QueueStopEvent():
                     yield from self._handle_stop_event(event, graph_runtime_state=None, trace_manager=trace_manager)
                     break

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

@@ -20,7 +20,7 @@ from core.app.app_config.features.suggested_questions_after_answer.manager impor
 )
 from core.app.app_config.features.text_to_speech.manager import TextToSpeechConfigManager
 from core.entities.agent_entities import PlanningStrategy
-from models.model import App, AppMode, AppModelConfig, Conversation
+from models.model import App, AppMode, AppModelConfig, AppModelConfigDict, Conversation
 
 OLD_TOOLS = ["dataset", "google_search", "web_reader", "wikipedia", "current_datetime"]
 
@@ -40,7 +40,7 @@ class AgentChatAppConfigManager(BaseAppConfigManager):
         app_model: App,
         app_model_config: AppModelConfig,
         conversation: Conversation | None = None,
-        override_config_dict: dict | None = None,
+        override_config_dict: AppModelConfigDict | None = None,
     ) -> AgentChatAppConfig:
         """
         Convert app model config to agent chat app config
@@ -61,7 +61,9 @@ class AgentChatAppConfigManager(BaseAppConfigManager):
             app_model_config_dict = app_model_config.to_dict()
             config_dict = app_model_config_dict.copy()
         else:
-            config_dict = override_config_dict or {}
+            if not override_config_dict:
+                raise Exception("override_config_dict is required when config_from is ARGS")
+            config_dict = override_config_dict
 
         app_mode = AppMode.value_of(app_model.mode)
         app_config = AgentChatAppConfig(
@@ -70,7 +72,7 @@ class AgentChatAppConfigManager(BaseAppConfigManager):
             app_mode=app_mode,
             app_model_config_from=config_from,
             app_model_config_id=app_model_config.id,
-            app_model_config_dict=config_dict,
+            app_model_config_dict=cast(dict[str, Any], config_dict),
             model=ModelConfigManager.convert(config=config_dict),
             prompt_template=PromptTemplateConfigManager.convert(config=config_dict),
             sensitive_word_avoidance=SensitiveWordAvoidanceConfigManager.convert(config=config_dict),
@@ -86,7 +88,7 @@ class AgentChatAppConfigManager(BaseAppConfigManager):
         return app_config
 
     @classmethod
-    def config_validate(cls, tenant_id: str, config: Mapping[str, Any]):
+    def config_validate(cls, tenant_id: str, config: Mapping[str, Any]) -> AppModelConfigDict:
         """
         Validate for agent chat app model config
 
@@ -157,7 +159,7 @@ class AgentChatAppConfigManager(BaseAppConfigManager):
         # Filter out extra parameters
         filtered_config = {key: config.get(key) for key in related_config_keys}
 
-        return filtered_config
+        return cast(AppModelConfigDict, filtered_config)
 
     @classmethod
     def validate_agent_mode_and_set_defaults(

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

@@ -1,3 +1,5 @@
+from typing import Any, cast
+
 from core.app.app_config.base_app_config_manager import BaseAppConfigManager
 from core.app.app_config.common.sensitive_word_avoidance.manager import SensitiveWordAvoidanceConfigManager
 from core.app.app_config.easy_ui_based_app.dataset.manager import DatasetConfigManager
@@ -13,7 +15,7 @@ from core.app.app_config.features.suggested_questions_after_answer.manager impor
     SuggestedQuestionsAfterAnswerConfigManager,
 )
 from core.app.app_config.features.text_to_speech.manager import TextToSpeechConfigManager
-from models.model import App, AppMode, AppModelConfig, Conversation
+from models.model import App, AppMode, AppModelConfig, AppModelConfigDict, Conversation
 
 
 class ChatAppConfig(EasyUIBasedAppConfig):
@@ -31,7 +33,7 @@ class ChatAppConfigManager(BaseAppConfigManager):
         app_model: App,
         app_model_config: AppModelConfig,
         conversation: Conversation | None = None,
-        override_config_dict: dict | None = None,
+        override_config_dict: AppModelConfigDict | None = None,
     ) -> ChatAppConfig:
         """
         Convert app model config to chat app config
@@ -64,7 +66,7 @@ class ChatAppConfigManager(BaseAppConfigManager):
             app_mode=app_mode,
             app_model_config_from=config_from,
             app_model_config_id=app_model_config.id,
-            app_model_config_dict=config_dict,
+            app_model_config_dict=cast(dict[str, Any], config_dict),
             model=ModelConfigManager.convert(config=config_dict),
             prompt_template=PromptTemplateConfigManager.convert(config=config_dict),
             sensitive_word_avoidance=SensitiveWordAvoidanceConfigManager.convert(config=config_dict),
@@ -79,7 +81,7 @@ class ChatAppConfigManager(BaseAppConfigManager):
         return app_config
 
     @classmethod
-    def config_validate(cls, tenant_id: str, config: dict):
+    def config_validate(cls, tenant_id: str, config: dict) -> AppModelConfigDict:
         """
         Validate for chat app model config
 
@@ -145,4 +147,4 @@ class ChatAppConfigManager(BaseAppConfigManager):
         # Filter out extra parameters
         filtered_config = {key: config.get(key) for key in related_config_keys}
 
-        return filtered_config
+        return cast(AppModelConfigDict, filtered_config)

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

@@ -173,8 +173,10 @@ class ChatAppRunner(AppRunner):
                 memory=memory,
                 message_id=message.id,
                 inputs=inputs,
-                vision_enabled=application_generate_entity.app_config.app_model_config_dict.get("file_upload", {}).get(
-                    "enabled", False
+                vision_enabled=bool(
+                    application_generate_entity.app_config.app_model_config_dict.get("file_upload", {})
+                    .get("image", {})
+                    .get("enabled", False)
                 ),
             )
             context_files = retrieved_files or []

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

@@ -1,3 +1,5 @@
+from typing import Any, cast
+
 from core.app.app_config.base_app_config_manager import BaseAppConfigManager
 from core.app.app_config.common.sensitive_word_avoidance.manager import SensitiveWordAvoidanceConfigManager
 from core.app.app_config.easy_ui_based_app.dataset.manager import DatasetConfigManager
@@ -8,7 +10,7 @@ from core.app.app_config.entities import EasyUIBasedAppConfig, EasyUIBasedAppMod
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.app_config.features.more_like_this.manager import MoreLikeThisConfigManager
 from core.app.app_config.features.text_to_speech.manager import TextToSpeechConfigManager
-from models.model import App, AppMode, AppModelConfig
+from models.model import App, AppMode, AppModelConfig, AppModelConfigDict
 
 
 class CompletionAppConfig(EasyUIBasedAppConfig):
@@ -22,7 +24,7 @@ class CompletionAppConfig(EasyUIBasedAppConfig):
 class CompletionAppConfigManager(BaseAppConfigManager):
     @classmethod
     def get_app_config(
-        cls, app_model: App, app_model_config: AppModelConfig, override_config_dict: dict | None = None
+        cls, app_model: App, app_model_config: AppModelConfig, override_config_dict: AppModelConfigDict | None = None
     ) -> CompletionAppConfig:
         """
         Convert app model config to completion app config
@@ -40,7 +42,9 @@ class CompletionAppConfigManager(BaseAppConfigManager):
             app_model_config_dict = app_model_config.to_dict()
             config_dict = app_model_config_dict.copy()
         else:
-            config_dict = override_config_dict or {}
+            if not override_config_dict:
+                raise Exception("override_config_dict is required when config_from is ARGS")
+            config_dict = override_config_dict
 
         app_mode = AppMode.value_of(app_model.mode)
         app_config = CompletionAppConfig(
@@ -49,7 +53,7 @@ class CompletionAppConfigManager(BaseAppConfigManager):
             app_mode=app_mode,
             app_model_config_from=config_from,
             app_model_config_id=app_model_config.id,
-            app_model_config_dict=config_dict,
+            app_model_config_dict=cast(dict[str, Any], config_dict),
             model=ModelConfigManager.convert(config=config_dict),
             prompt_template=PromptTemplateConfigManager.convert(config=config_dict),
             sensitive_word_avoidance=SensitiveWordAvoidanceConfigManager.convert(config=config_dict),
@@ -64,7 +68,7 @@ class CompletionAppConfigManager(BaseAppConfigManager):
         return app_config
 
     @classmethod
-    def config_validate(cls, tenant_id: str, config: dict):
+    def config_validate(cls, tenant_id: str, config: dict) -> AppModelConfigDict:
         """
         Validate for completion app model config
 
@@ -116,4 +120,4 @@ class CompletionAppConfigManager(BaseAppConfigManager):
         # Filter out extra parameters
         filtered_config = {key: config.get(key) for key in related_config_keys}
 
-        return filtered_config
+        return cast(AppModelConfigDict, filtered_config)

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

@@ -275,7 +275,7 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
             raise ValueError("Message app_model_config is None")
         override_model_config_dict = app_model_config.to_dict()
         model_dict = override_model_config_dict["model"]
-        completion_params = model_dict.get("completion_params")
+        completion_params = model_dict.get("completion_params", {})
         completion_params["temperature"] = 0.9
         model_dict["completion_params"] = completion_params
         override_model_config_dict["model"] = model_dict

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

@@ -132,8 +132,10 @@ class CompletionAppRunner(AppRunner):
                 hit_callback=hit_callback,
                 message_id=message.id,
                 inputs=inputs,
-                vision_enabled=application_generate_entity.app_config.app_model_config_dict.get("file_upload", {}).get(
-                    "enabled", False
+                vision_enabled=bool(
+                    application_generate_entity.app_config.app_model_config_dict.get("file_upload", {})
+                    .get("image", {})
+                    .get("enabled", False)
                 ),
             )
             context_files = retrieved_files or []

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

@@ -8,12 +8,14 @@ from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
 from core.app.entities.app_invoke_entities import (
     InvokeFrom,
     RagPipelineGenerateEntity,
+    UserFrom,
+    build_dify_run_context,
 )
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.workflow.node_factory import DifyNodeFactory
 from core.workflow.workflow_entry import WorkflowEntry
 from dify_graph.entities.graph_init_params import GraphInitParams
-from dify_graph.enums import UserFrom, WorkflowType
+from dify_graph.enums import WorkflowType
 from dify_graph.graph import Graph
 from dify_graph.graph_events import GraphEngineEvent, GraphRunFailedEvent
 from dify_graph.repositories.workflow_execution_repository import WorkflowExecutionRepository
@@ -256,13 +258,15 @@ class PipelineRunner(WorkflowBasedAppRunner):
         # init graph
         # Create required parameters for Graph.init
         graph_init_params = GraphInitParams(
-            tenant_id=workflow.tenant_id,
-            app_id=self._app_id,
             workflow_id=workflow.id,
             graph_config=graph_config,
-            user_id=self.application_generate_entity.user_id,
-            user_from=user_from,
-            invoke_from=invoke_from,
+            run_context=build_dify_run_context(
+                tenant_id=workflow.tenant_id,
+                app_id=self._app_id,
+                user_id=self.application_generate_entity.user_id,
+                user_from=user_from,
+                invoke_from=invoke_from,
+            ),
             call_depth=0,
         )
 

api/core/app/apps/workflow_app_runner.py

@@ -4,7 +4,7 @@ from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, build_dify_run_context
 from core.app.entities.queue_entities import (
     AppQueueEvent,
     QueueAgentLogEvent,
@@ -33,7 +33,6 @@ from core.workflow.node_factory import DifyNodeFactory
 from core.workflow.workflow_entry import WorkflowEntry
 from dify_graph.entities import GraphInitParams
 from dify_graph.entities.pause_reason import HumanInputRequired
-from dify_graph.enums import UserFrom
 from dify_graph.graph import Graph
 from dify_graph.graph_engine.layers.base import GraphEngineLayer
 from dify_graph.graph_events import (
@@ -119,13 +118,15 @@ class WorkflowBasedAppRunner:
 
         # Create required parameters for Graph.init
         graph_init_params = GraphInitParams(
-            tenant_id=tenant_id or "",
-            app_id=self._app_id,
             workflow_id=workflow_id,
             graph_config=graph_config,
-            user_id=user_id,
-            user_from=user_from,
-            invoke_from=invoke_from,
+            run_context=build_dify_run_context(
+                tenant_id=tenant_id or "",
+                app_id=self._app_id,
+                user_id=user_id,
+                user_from=user_from,
+                invoke_from=invoke_from,
+            ),
             call_depth=0,
         )
 
@@ -267,13 +268,15 @@ class WorkflowBasedAppRunner:
 
         # Create required parameters for Graph.init
         graph_init_params = GraphInitParams(
-            tenant_id=workflow.tenant_id,
-            app_id=self._app_id,
             workflow_id=workflow.id,
             graph_config=graph_config,
-            user_id="",
-            user_from=UserFrom.ACCOUNT,
-            invoke_from=InvokeFrom.DEBUGGER,
+            run_context=build_dify_run_context(
+                tenant_id=workflow.tenant_id,
+                app_id=self._app_id,
+                user_id="",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+            ),
             call_depth=0,
         )
 

api/core/app/entities/app_invoke_entities.py

@@ -1,4 +1,5 @@
 from collections.abc import Mapping, Sequence
+from enum import StrEnum
 from typing import TYPE_CHECKING, Any, Optional
 
 from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator
@@ -6,7 +7,7 @@ from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validat
 from constants import UUID_NIL
 from core.app.app_config.entities import EasyUIBasedAppConfig, WorkflowUIBasedAppConfig
 from core.entities.provider_configuration import ProviderModelBundle
-from dify_graph.enums import InvokeFrom
+from dify_graph.entities.graph_init_params import DIFY_RUN_CONTEXT_KEY
 from dify_graph.file import File, FileUploadConfig
 from dify_graph.model_runtime.entities.model_entities import AIModelEntity
 
@@ -14,6 +15,69 @@ if TYPE_CHECKING:
     from core.ops.ops_trace_manager import TraceQueueManager
 
 
+class UserFrom(StrEnum):
+    ACCOUNT = "account"
+    END_USER = "end-user"
+
+
+class InvokeFrom(StrEnum):
+    SERVICE_API = "service-api"
+    WEB_APP = "web-app"
+    TRIGGER = "trigger"
+    EXPLORE = "explore"
+    DEBUGGER = "debugger"
+    PUBLISHED_PIPELINE = "published"
+    VALIDATION = "validation"
+
+    @classmethod
+    def value_of(cls, value: str) -> "InvokeFrom":
+        return cls(value)
+
+    def to_source(self) -> str:
+        source_mapping = {
+            InvokeFrom.WEB_APP: "web_app",
+            InvokeFrom.DEBUGGER: "dev",
+            InvokeFrom.EXPLORE: "explore_app",
+            InvokeFrom.TRIGGER: "trigger",
+            InvokeFrom.SERVICE_API: "api",
+        }
+        return source_mapping.get(self, "dev")
+
+
+class DifyRunContext(BaseModel):
+    tenant_id: str
+    app_id: str
+    user_id: str
+    user_from: UserFrom
+    invoke_from: InvokeFrom
+
+
+def build_dify_run_context(
+    *,
+    tenant_id: str,
+    app_id: str,
+    user_id: str,
+    user_from: UserFrom,
+    invoke_from: InvokeFrom,
+    extra_context: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """
+    Build graph run_context with the reserved Dify runtime payload.
+
+    `extra_context` can carry user-defined context keys. The reserved `_dify`
+    payload is always overwritten by this function to keep one canonical source.
+    """
+    run_context = dict(extra_context) if extra_context else {}
+    run_context[DIFY_RUN_CONTEXT_KEY] = DifyRunContext(
+        tenant_id=tenant_id,
+        app_id=app_id,
+        user_id=user_id,
+        user_from=user_from,
+        invoke_from=invoke_from,
+    )
+    return run_context
+
+
 class ModelConfigWithCredentialsEntity(BaseModel):
     """
     Model Config With Credentials Entity.

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -2,7 +2,7 @@ import logging
 import time
 from collections.abc import Generator
 from threading import Thread
-from typing import Union, cast
+from typing import Any, Union, cast
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -44,14 +44,13 @@ from core.app.entities.task_entities import (
 )
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
+from core.app.task_pipeline.message_file_utils import prepare_file_dict
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.model_manager import ModelInstance
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
-from core.tools.signature import sign_tool_file
-from dify_graph.file import helpers as file_helpers
 from dify_graph.file.enums import FileTransferMethod
 from dify_graph.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
 from dify_graph.model_runtime.entities.message_entities import (
@@ -219,14 +218,14 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         tenant_id = self._application_generate_entity.app_config.tenant_id
         task_id = self._application_generate_entity.task_id
         publisher = None
-        text_to_speech_dict = self._app_config.app_model_config_dict.get("text_to_speech")
+        text_to_speech_dict = cast(dict[str, Any], self._app_config.app_model_config_dict.get("text_to_speech"))
         if (
             text_to_speech_dict
             and text_to_speech_dict.get("autoPlay") == "enabled"
             and text_to_speech_dict.get("enabled")
         ):
             publisher = AppGeneratorTTSPublisher(
-                tenant_id, text_to_speech_dict.get("voice", None), text_to_speech_dict.get("language", None)
+                tenant_id, text_to_speech_dict.get("voice", ""), text_to_speech_dict.get("language", None)
             )
         for response in self._process_stream_response(publisher=publisher, trace_manager=trace_manager):
             while True:
@@ -460,91 +459,40 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         """
         self._task_state.metadata.usage = self._task_state.llm_result.usage
         metadata_dict = self._task_state.metadata.model_dump()
+
+        # Fetch files associated with this message
+        files = None
+        with Session(db.engine, expire_on_commit=False) as session:
+            message_files = session.scalars(select(MessageFile).where(MessageFile.message_id == self._message_id)).all()
+
+            if message_files:
+                # Fetch all required UploadFile objects in a single query to avoid N+1 problem
+                upload_file_ids = list(
+                    dict.fromkeys(
+                        mf.upload_file_id
+                        for mf in message_files
+                        if mf.transfer_method == FileTransferMethod.LOCAL_FILE and mf.upload_file_id
+                    )
+                )
+                upload_files_map = {}
+                if upload_file_ids:
+                    upload_files = session.scalars(select(UploadFile).where(UploadFile.id.in_(upload_file_ids))).all()
+                    upload_files_map = {uf.id: uf for uf in upload_files}
+
+                files_list = []
+                for message_file in message_files:
+                    file_dict = prepare_file_dict(message_file, upload_files_map)
+                    files_list.append(file_dict)
+
+                files = files_list or None
+
         return MessageEndStreamResponse(
             task_id=self._application_generate_entity.task_id,
             id=self._message_id,
             metadata=metadata_dict,
+            files=files,
         )
 
-    def _record_files(self):
-        with Session(db.engine, expire_on_commit=False) as session:
-            message_files = session.scalars(select(MessageFile).where(MessageFile.message_id == self._message_id)).all()
-            if not message_files:
-                return None
-
-            files_list = []
-            upload_file_ids = [
-                mf.upload_file_id
-                for mf in message_files
-                if mf.transfer_method == FileTransferMethod.LOCAL_FILE and mf.upload_file_id
-            ]
-            upload_files_map = {}
-            if upload_file_ids:
-                upload_files = session.scalars(select(UploadFile).where(UploadFile.id.in_(upload_file_ids))).all()
-                upload_files_map = {uf.id: uf for uf in upload_files}
-
-            for message_file in message_files:
-                upload_file = None
-                if message_file.transfer_method == FileTransferMethod.LOCAL_FILE and message_file.upload_file_id:
-                    upload_file = upload_files_map.get(message_file.upload_file_id)
-
-                url = None
-                filename = "file"
-                mime_type = "application/octet-stream"
-                size = 0
-                extension = ""
-
-                if message_file.transfer_method == FileTransferMethod.REMOTE_URL:
-                    url = message_file.url
-                    if message_file.url:
-                        filename = message_file.url.split("/")[-1].split("?")[0]  # Remove query params
-                elif message_file.transfer_method == FileTransferMethod.LOCAL_FILE:
-                    if upload_file:
-                        url = file_helpers.get_signed_file_url(upload_file_id=str(upload_file.id))
-                        filename = upload_file.name
-                        mime_type = upload_file.mime_type or "application/octet-stream"
-                        size = upload_file.size or 0
-                        extension = f".{upload_file.extension}" if upload_file.extension else ""
-                    elif message_file.upload_file_id:
-                        # Fallback: generate URL even if upload_file not found
-                        url = file_helpers.get_signed_file_url(upload_file_id=str(message_file.upload_file_id))
-                elif message_file.transfer_method == FileTransferMethod.TOOL_FILE and message_file.url:
-                    # For tool files, use URL directly if it's HTTP, otherwise sign it
-                    if message_file.url.startswith("http"):
-                        url = message_file.url
-                        filename = message_file.url.split("/")[-1].split("?")[0]
-                    else:
-                        # Extract tool file id and extension from URL
-                        url_parts = message_file.url.split("/")
-                        if url_parts:
-                            file_part = url_parts[-1].split("?")[0]  # Remove query params first
-                            # Use rsplit to correctly handle filenames with multiple dots
-                            if "." in file_part:
-                                tool_file_id, ext = file_part.rsplit(".", 1)
-                                extension = f".{ext}"
-                            else:
-                                tool_file_id = file_part
-                                extension = ".bin"
-                            url = sign_tool_file(tool_file_id=tool_file_id, extension=extension)
-                            filename = file_part
-
-                transfer_method_value = message_file.transfer_method
-                remote_url = message_file.url if message_file.transfer_method == FileTransferMethod.REMOTE_URL else ""
-                file_dict = {
-                    "related_id": message_file.id,
-                    "extension": extension,
-                    "filename": filename,
-                    "size": size,
-                    "mime_type": mime_type,
-                    "transfer_method": transfer_method_value,
-                    "type": message_file.type,
-                    "url": url or "",
-                    "upload_file_id": message_file.upload_file_id or message_file.id,
-                    "remote_url": remote_url,
-                }
-                files_list.append(file_dict)
-            return files_list or None
-
     def _agent_message_to_stream_response(self, answer: str, message_id: str) -> AgentMessageStreamResponse:
         """
         Agent message to stream response.

api/core/app/task_pipeline/message_cycle_manager.py

@@ -1,7 +1,6 @@
 import hashlib
 import logging
-import time
-from threading import Thread
+from threading import Thread, Timer
 from typing import Union
 
 from flask import Flask, current_app
@@ -96,9 +95,9 @@ class MessageCycleManager:
         if auto_generate_conversation_name and is_first_message:
             # start generate thread
             # time.sleep not block other logic
-            time.sleep(1)
-            thread = Thread(
-                target=self._generate_conversation_name_worker,
+            thread = Timer(
+                1,
+                self._generate_conversation_name_worker,
                 kwargs={
                     "flask_app": current_app._get_current_object(),  # type: ignore
                     "conversation_id": conversation_id,

api/core/app/task_pipeline/message_file_utils.py

@@ -0,0 +1,76 @@
+from core.tools.signature import sign_tool_file
+from dify_graph.file import helpers as file_helpers
+from dify_graph.file.enums import FileTransferMethod
+from models.model import MessageFile, UploadFile
+
+MAX_TOOL_FILE_EXTENSION_LENGTH = 10
+
+
+def prepare_file_dict(message_file: MessageFile, upload_files_map: dict[str, UploadFile]) -> dict:
+    """
+    Prepare file dictionary for message end stream response.
+
+    :param message_file: MessageFile instance
+    :param upload_files_map: Dictionary mapping upload_file_id to UploadFile
+    :return: Dictionary containing file information
+    """
+    upload_file = None
+    if message_file.transfer_method == FileTransferMethod.LOCAL_FILE and message_file.upload_file_id:
+        upload_file = upload_files_map.get(message_file.upload_file_id)
+
+    url = None
+    filename = "file"
+    mime_type = "application/octet-stream"
+    size = 0
+    extension = ""
+
+    if message_file.transfer_method == FileTransferMethod.REMOTE_URL:
+        url = message_file.url
+        if message_file.url:
+            filename = message_file.url.split("/")[-1].split("?")[0]
+            if "." in filename:
+                extension = "." + filename.rsplit(".", 1)[1]
+    elif message_file.transfer_method == FileTransferMethod.LOCAL_FILE:
+        if upload_file:
+            url = file_helpers.get_signed_file_url(upload_file_id=str(upload_file.id))
+            filename = upload_file.name
+            mime_type = upload_file.mime_type or "application/octet-stream"
+            size = upload_file.size or 0
+            extension = f".{upload_file.extension}" if upload_file.extension else ""
+        elif message_file.upload_file_id:
+            url = file_helpers.get_signed_file_url(upload_file_id=str(message_file.upload_file_id))
+    elif message_file.transfer_method == FileTransferMethod.TOOL_FILE and message_file.url:
+        if message_file.url.startswith(("http://", "https://")):
+            url = message_file.url
+            filename = message_file.url.split("/")[-1].split("?")[0]
+            if "." in filename:
+                extension = "." + filename.rsplit(".", 1)[1]
+        else:
+            url_parts = message_file.url.split("/")
+            if url_parts:
+                file_part = url_parts[-1].split("?")[0]
+                if "." in file_part:
+                    tool_file_id, ext = file_part.rsplit(".", 1)
+                    extension = f".{ext}"
+                    if len(extension) > MAX_TOOL_FILE_EXTENSION_LENGTH:
+                        extension = ".bin"
+                else:
+                    tool_file_id = file_part
+                    extension = ".bin"
+                url = sign_tool_file(tool_file_id=tool_file_id, extension=extension)
+                filename = file_part
+
+    transfer_method_value = message_file.transfer_method.value
+    remote_url = message_file.url if message_file.transfer_method == FileTransferMethod.REMOTE_URL else ""
+    return {
+        "related_id": message_file.id,
+        "extension": extension,
+        "filename": filename,
+        "size": size,
+        "mime_type": mime_type,
+        "transfer_method": transfer_method_value,
+        "type": message_file.type,
+        "url": url or "",
+        "upload_file_id": message_file.upload_file_id or message_file.id,
+        "remote_url": remote_url,
+    }

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

@@ -75,8 +75,9 @@ class LLMQuotaLayer(GraphEngineLayer):
             return
 
         try:
+            dify_ctx = node.require_dify_context()
             deduct_llm_quota(
-                tenant_id=node.tenant_id,
+                tenant_id=dify_ctx.tenant_id,
                 model_instance=model_instance,
                 usage=result_event.node_run_result.llm_usage,
             )

api/core/ops/aliyun_trace/data_exporter/traceclient.py

@@ -7,7 +7,7 @@ import uuid
 from collections import deque
 from collections.abc import Sequence
 from datetime import datetime
-from typing import Final, cast
+from typing import Final
 from urllib.parse import urljoin
 
 import httpx
@@ -201,7 +201,7 @@ def convert_to_trace_id(uuid_v4: str | None) -> int:
         raise ValueError("UUID cannot be None")
     try:
         uuid_obj = uuid.UUID(uuid_v4)
-        return cast(int, uuid_obj.int)
+        return uuid_obj.int
     except ValueError as e:
         raise ValueError(f"Invalid UUID input: {uuid_v4}") from e
 

api/core/ops/opik_trace/opik_trace.py

@@ -1,3 +1,4 @@
+import hashlib
 import logging
 import os
 import uuid
@@ -46,6 +47,22 @@ def wrap_metadata(metadata, **kwargs):
     return metadata
 
 
+def _seed_to_uuid4(seed: str) -> str:
+    """Derive a deterministic UUID4-formatted string from an arbitrary seed.
+
+    uuid4_to_uuid7 requires a valid UUID v4 string, but some Dify identifiers
+    are not UUIDs (e.g. a workflow_run_id with a "-root" suffix appended to
+    distinguish the root span from the trace).  This helper hashes the seed
+    with MD5 and patches the version/variant bits so the result satisfies the
+    UUID v4 contract.
+    """
+    raw = hashlib.md5(seed.encode()).digest()
+    ba = bytearray(raw)
+    ba[6] = (ba[6] & 0x0F) | 0x40  # version 4
+    ba[8] = (ba[8] & 0x3F) | 0x80  # variant 1
+    return str(uuid.UUID(bytes=bytes(ba)))
+
+
 def prepare_opik_uuid(user_datetime: datetime | None, user_uuid: str | None):
     """Opik needs UUIDv7 while Dify uses UUIDv4 for identifier of most
     messages and objects. The type-hints of BaseTraceInfo indicates that
@@ -95,60 +112,52 @@ class OpikDataTrace(BaseTraceInstance):
             self.generate_name_trace(trace_info)
 
     def workflow_trace(self, trace_info: WorkflowTraceInfo):
-        dify_trace_id = trace_info.trace_id or trace_info.workflow_run_id
-        opik_trace_id = prepare_opik_uuid(trace_info.start_time, dify_trace_id)
         workflow_metadata = wrap_metadata(
             trace_info.metadata, message_id=trace_info.message_id, workflow_app_log_id=trace_info.workflow_app_log_id
         )
-        root_span_id = None
 
         if trace_info.message_id:
             dify_trace_id = trace_info.trace_id or trace_info.message_id
-            opik_trace_id = prepare_opik_uuid(trace_info.start_time, dify_trace_id)
-
-            trace_data = {
-                "id": opik_trace_id,
-                "name": TraceTaskName.MESSAGE_TRACE,
-                "start_time": trace_info.start_time,
-                "end_time": trace_info.end_time,
-                "metadata": workflow_metadata,
-                "input": wrap_dict("input", trace_info.workflow_run_inputs),
-                "output": wrap_dict("output", trace_info.workflow_run_outputs),
-                "thread_id": trace_info.conversation_id,
-                "tags": ["message", "workflow"],
-                "project_name": self.project,
-            }
-            self.add_trace(trace_data)
-
-            root_span_id = prepare_opik_uuid(trace_info.start_time, trace_info.workflow_run_id)
-            span_data = {
-                "id": root_span_id,
-                "parent_span_id": None,
-                "trace_id": opik_trace_id,
-                "name": TraceTaskName.WORKFLOW_TRACE,
-                "input": wrap_dict("input", trace_info.workflow_run_inputs),
-                "output": wrap_dict("output", trace_info.workflow_run_outputs),
-                "start_time": trace_info.start_time,
-                "end_time": trace_info.end_time,
-                "metadata": workflow_metadata,
-                "tags": ["workflow"],
-                "project_name": self.project,
-            }
-            self.add_span(span_data)
+            trace_name = TraceTaskName.MESSAGE_TRACE
+            trace_tags = ["message", "workflow"]
+            root_span_seed = trace_info.workflow_run_id
         else:
-            trace_data = {
-                "id": opik_trace_id,
-                "name": TraceTaskName.MESSAGE_TRACE,
-                "start_time": trace_info.start_time,
-                "end_time": trace_info.end_time,
-                "metadata": workflow_metadata,
-                "input": wrap_dict("input", trace_info.workflow_run_inputs),
-                "output": wrap_dict("output", trace_info.workflow_run_outputs),
-                "thread_id": trace_info.conversation_id,
-                "tags": ["workflow"],
-                "project_name": self.project,
-            }
-            self.add_trace(trace_data)
+            dify_trace_id = trace_info.trace_id or trace_info.workflow_run_id
+            trace_name = TraceTaskName.WORKFLOW_TRACE
+            trace_tags = ["workflow"]
+            root_span_seed = _seed_to_uuid4(trace_info.workflow_run_id + "-root")
+
+        opik_trace_id = prepare_opik_uuid(trace_info.start_time, dify_trace_id)
+
+        trace_data = {
+            "id": opik_trace_id,
+            "name": trace_name,
+            "start_time": trace_info.start_time,
+            "end_time": trace_info.end_time,
+            "metadata": workflow_metadata,
+            "input": wrap_dict("input", trace_info.workflow_run_inputs),
+            "output": wrap_dict("output", trace_info.workflow_run_outputs),
+            "thread_id": trace_info.conversation_id,
+            "tags": trace_tags,
+            "project_name": self.project,
+        }
+        self.add_trace(trace_data)
+
+        root_span_id = prepare_opik_uuid(trace_info.start_time, root_span_seed)
+        span_data = {
+            "id": root_span_id,
+            "parent_span_id": None,
+            "trace_id": opik_trace_id,
+            "name": TraceTaskName.WORKFLOW_TRACE,
+            "input": wrap_dict("input", trace_info.workflow_run_inputs),
+            "output": wrap_dict("output", trace_info.workflow_run_outputs),
+            "start_time": trace_info.start_time,
+            "end_time": trace_info.end_time,
+            "metadata": workflow_metadata,
+            "tags": ["workflow"],
+            "project_name": self.project,
+        }
+        self.add_span(span_data)
 
         # through workflow_run_id get all_nodes_execution using repository
         session_factory = sessionmaker(bind=db.engine)
@@ -231,15 +240,13 @@ class OpikDataTrace(BaseTraceInstance):
             else:
                 run_type = "tool"
 
-            parent_span_id = trace_info.workflow_app_log_id or trace_info.workflow_run_id
-
             if not total_tokens:
                 total_tokens = execution_metadata.get(WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS) or 0
 
             span_data = {
                 "trace_id": opik_trace_id,
                 "id": prepare_opik_uuid(created_at, node_execution_id),
-                "parent_span_id": prepare_opik_uuid(trace_info.start_time, parent_span_id),
+                "parent_span_id": root_span_id,
                 "name": node_name,
                 "type": run_type,
                 "start_time": created_at,

api/core/ops/tencent_trace/client.py

@@ -120,7 +120,8 @@ class TencentTraceClient:
 
         # Metrics exporter and instruments
         try:
-            from opentelemetry.sdk.metrics import Histogram, MeterProvider
+            from opentelemetry.sdk.metrics import Histogram as SdkHistogram
+            from opentelemetry.sdk.metrics import MeterProvider
             from opentelemetry.sdk.metrics.export import AggregationTemporality, PeriodicExportingMetricReader
 
             protocol = os.getenv("OTEL_EXPORTER_OTLP_PROTOCOL", "").strip().lower()
@@ -128,7 +129,7 @@ class TencentTraceClient:
             use_http_json = protocol in {"http/json", "http-json"}
 
             # Tencent APM works best with delta aggregation temporality
-            preferred_temporality: dict[type, AggregationTemporality] = {Histogram: AggregationTemporality.DELTA}
+            preferred_temporality: dict[type, AggregationTemporality] = {SdkHistogram: AggregationTemporality.DELTA}
 
             def _create_metric_exporter(exporter_cls, **kwargs):
                 """Create metric exporter with preferred_temporality support"""

api/core/ops/tencent_trace/utils.py

@@ -6,7 +6,6 @@ import hashlib
 import random
 import uuid
 from datetime import datetime
-from typing import cast
 
 from opentelemetry.trace import Link, SpanContext, TraceFlags
 
@@ -23,7 +22,7 @@ class TencentTraceUtils:
             uuid_obj = uuid.UUID(uuid_v4) if uuid_v4 else uuid.uuid4()
         except Exception as e:
             raise ValueError(f"Invalid UUID input: {e}")
-        return cast(int, uuid_obj.int)
+        return uuid_obj.int
 
     @staticmethod
     def convert_to_span_id(uuid_v4: str | None, span_type: str) -> int:
@@ -52,9 +51,9 @@ class TencentTraceUtils:
     @staticmethod
     def create_link(trace_id_str: str) -> Link:
         try:
-            trace_id = int(trace_id_str, 16) if len(trace_id_str) == 32 else cast(int, uuid.UUID(trace_id_str).int)
+            trace_id = int(trace_id_str, 16) if len(trace_id_str) == 32 else uuid.UUID(trace_id_str).int
         except (ValueError, TypeError):
-            trace_id = cast(int, uuid.uuid4().int)
+            trace_id = uuid.uuid4().int
 
         span_context = SpanContext(
             trace_id=trace_id,

api/core/plugin/backwards_invocation/app.py

@@ -1,6 +1,6 @@
 import uuid
 from collections.abc import Generator, Mapping
-from typing import Union
+from typing import Any, Union, cast
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -34,14 +34,14 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
             if workflow is None:
                 raise ValueError("unexpected app type")
 
-            features_dict = workflow.features_dict
+            features_dict: dict[str, Any] = workflow.features_dict
             user_input_form = workflow.user_input_form(to_old_structure=True)
         else:
             app_model_config = app.app_model_config
             if app_model_config is None:
                 raise ValueError("unexpected app type")
 
-            features_dict = app_model_config.to_dict()
+            features_dict = cast(dict[str, Any], app_model_config.to_dict())
 
             user_input_form = features_dict.get("user_input_form", [])
 

api/core/rag/datasource/vdb/chroma/chroma_vector.py

@@ -65,7 +65,7 @@ class ChromaVector(BaseVector):
             self._client.get_or_create_collection(collection_name)
             redis_client.set(collection_exist_cache_key, 1, ex=3600)
 
-    def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs):
+    def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs) -> list[str]:
         uuids = self._get_uuids(documents)
         texts = [d.page_content for d in documents]
         metadatas = [d.metadata for d in documents]
@@ -73,6 +73,7 @@ class ChromaVector(BaseVector):
         collection = self._client.get_or_create_collection(self._collection_name)
         # FIXME: chromadb using numpy array, fix the type error later
         collection.upsert(ids=uuids, documents=texts, embeddings=embeddings, metadatas=metadatas)  # type: ignore
+        return uuids
 
     def delete_by_metadata_field(self, key: str, value: str):
         collection = self._client.get_or_create_collection(self._collection_name)

api/core/rag/datasource/vdb/clickzetta/clickzetta_vector.py

@@ -605,25 +605,36 @@ class ClickzettaVector(BaseVector):
                 logger.warning("Failed to create inverted index: %s", e)
                 # Continue without inverted index - full-text search will fall back to LIKE
 
-    def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs):
+    def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs) -> list[str]:
         """Add documents with embeddings to the collection."""
         if not documents:
-            return
+            return []
 
         batch_size = self._config.batch_size
         total_batches = (len(documents) + batch_size - 1) // batch_size
+        added_ids = []
 
         for i in range(0, len(documents), batch_size):
             batch_docs = documents[i : i + batch_size]
             batch_embeddings = embeddings[i : i + batch_size]
+            batch_doc_ids = []
+            for doc in batch_docs:
+                metadata = doc.metadata if isinstance(doc.metadata, dict) else {}
+                batch_doc_ids.append(self._safe_doc_id(metadata.get("doc_id", str(uuid.uuid4()))))
+            added_ids.extend(batch_doc_ids)
 
             # Execute batch insert through write queue
-            self._execute_write(self._insert_batch, batch_docs, batch_embeddings, i, batch_size, total_batches)
+            self._execute_write(
+                self._insert_batch, batch_docs, batch_embeddings, batch_doc_ids, i, batch_size, total_batches
+            )
+
+        return added_ids
 
     def _insert_batch(
         self,
         batch_docs: list[Document],
         batch_embeddings: list[list[float]],
+        batch_doc_ids: list[str],
         batch_index: int,
         batch_size: int,
         total_batches: int,
@@ -641,14 +652,9 @@ class ClickzettaVector(BaseVector):
         data_rows = []
         vector_dimension = len(batch_embeddings[0]) if batch_embeddings and batch_embeddings[0] else 768
 
-        for doc, embedding in zip(batch_docs, batch_embeddings):
+        for doc, embedding, doc_id in zip(batch_docs, batch_embeddings, batch_doc_ids):
             # Optimized: minimal checks for common case, fallback for edge cases
-            metadata = doc.metadata or {}
-
-            if not isinstance(metadata, dict):
-                metadata = {}
-
-            doc_id = self._safe_doc_id(metadata.get("doc_id", str(uuid.uuid4())))
+            metadata = doc.metadata if isinstance(doc.metadata, dict) else {}
 
             # Fast path for JSON serialization
             try:

api/core/rag/extractor/pdf_extractor.py

@@ -114,7 +114,6 @@ class PdfExtractor(BaseExtractor):
         """
         image_content = []
         upload_files = []
-        base_url = dify_config.INTERNAL_FILES_URL or dify_config.FILES_URL
 
         try:
             image_objects = page.get_objects(filter=(pdfium_c.FPDF_PAGEOBJ_IMAGE,))
@@ -164,7 +163,7 @@ class PdfExtractor(BaseExtractor):
                         used_at=naive_utc_now(),
                     )
                     upload_files.append(upload_file)
-                    image_content.append(f"![image]({base_url}/files/{upload_file.id}/file-preview)")
+                    image_content.append(f"![image](/files/{upload_file.id}/file-preview)")
                 except Exception as e:
                     logger.warning("Failed to extract image from PDF: %s", e)
                     continue

api/core/rag/extractor/word_extractor.py

@@ -87,7 +87,6 @@ class WordExtractor(BaseExtractor):
     def _extract_images_from_docx(self, doc):
         image_count = 0
         image_map = {}
-        base_url = dify_config.INTERNAL_FILES_URL or dify_config.FILES_URL
 
         for r_id, rel in doc.part.rels.items():
             if "image" in rel.target_ref:
@@ -126,7 +125,7 @@ class WordExtractor(BaseExtractor):
                             used_at=naive_utc_now(),
                         )
                         db.session.add(upload_file)
-                        image_map[r_id] = f"![image]({base_url}/files/{upload_file.id}/file-preview)"
+                        image_map[r_id] = f"![image](/files/{upload_file.id}/file-preview)"
                 else:
                     image_ext = rel.target_ref.split(".")[-1]
                     if image_ext is None:
@@ -154,7 +153,7 @@ class WordExtractor(BaseExtractor):
                         used_at=naive_utc_now(),
                     )
                     db.session.add(upload_file)
-                    image_map[rel.target_part] = f"![image]({base_url}/files/{upload_file.id}/file-preview)"
+                    image_map[rel.target_part] = f"![image](/files/{upload_file.id}/file-preview)"
         db.session.commit()
         return image_map
 

api/core/rag/retrieval/dataset_retrieval.py

@@ -57,7 +57,7 @@ from core.rag.retrieval.template_prompts import (
 from core.tools.signature import sign_upload_file
 from core.tools.utils.dataset_retriever.dataset_retriever_base_tool import DatasetRetrieverBaseTool
 from dify_graph.file import File, FileTransferMethod, FileType
-from dify_graph.model_runtime.entities.llm_entities import LLMResult, LLMUsage
+from dify_graph.model_runtime.entities.llm_entities import LLMMode, LLMResult, LLMUsage
 from dify_graph.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
 from dify_graph.model_runtime.entities.model_entities import ModelFeature, ModelType
 from dify_graph.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
@@ -127,11 +127,12 @@ class DatasetRetrieval:
         metadata_filter_document_ids, metadata_condition = None, None
 
         if request.metadata_filtering_mode != "disabled":
-            # Convert workflow layer types to app_config layer types
-            if not request.metadata_model_config:
-                raise ValueError("metadata_model_config is required for this method")
+            app_metadata_model_config = ModelConfig(provider="", name="", mode=LLMMode.CHAT, completion_params={})
+            if request.metadata_filtering_mode == "automatic":
+                if not request.metadata_model_config:
+                    raise ValueError("metadata_model_config is required for this method")
 
-            app_metadata_model_config = ModelConfig.model_validate(request.metadata_model_config.model_dump())
+                app_metadata_model_config = ModelConfig.model_validate(request.metadata_model_config.model_dump())
 
             app_metadata_filtering_conditions = None
             if request.metadata_filtering_conditions is not None:

api/core/repositories/human_input_repository.py

@@ -4,9 +4,10 @@ from collections.abc import Mapping, Sequence
 from datetime import datetime
 from typing import Any
 
-from sqlalchemy import Engine, select
-from sqlalchemy.orm import Session, selectinload, sessionmaker
+from sqlalchemy import select
+from sqlalchemy.orm import Session, selectinload
 
+from core.db.session_factory import session_factory
 from dify_graph.nodes.human_input.entities import (
     DeliveryChannelConfig,
     EmailDeliveryMethod,
@@ -198,12 +199,9 @@ class _InvalidTimeoutStatusError(ValueError):
 class HumanInputFormRepositoryImpl:
     def __init__(
         self,
-        session_factory: sessionmaker | Engine,
+        *,
         tenant_id: str,
     ):
-        if isinstance(session_factory, Engine):
-            session_factory = sessionmaker(bind=session_factory)
-        self._session_factory = session_factory
         self._tenant_id = tenant_id
 
     def _delivery_method_to_model(
@@ -217,7 +215,7 @@ class HumanInputFormRepositoryImpl:
             id=delivery_id,
             form_id=form_id,
             delivery_method_type=delivery_method.type,
-            delivery_config_id=delivery_method.id,
+            delivery_config_id=str(delivery_method.id),
             channel_payload=delivery_method.model_dump_json(),
         )
         recipients: list[HumanInputFormRecipient] = []
@@ -343,7 +341,7 @@ class HumanInputFormRepositoryImpl:
     def create_form(self, params: FormCreateParams) -> HumanInputFormEntity:
         form_config: HumanInputNodeData = params.form_config
 
-        with self._session_factory(expire_on_commit=False) as session, session.begin():
+        with session_factory.create_session() as session, session.begin():
             # Generate unique form ID
             form_id = str(uuidv7())
             start_time = naive_utc_now()
@@ -435,7 +433,7 @@ class HumanInputFormRepositoryImpl:
             HumanInputForm.node_id == node_id,
             HumanInputForm.tenant_id == self._tenant_id,
         )
-        with self._session_factory(expire_on_commit=False) as session:
+        with session_factory.create_session() as session:
             form_model: HumanInputForm | None = session.scalars(form_query).first()
             if form_model is None:
                 return None
@@ -448,18 +446,13 @@ class HumanInputFormRepositoryImpl:
 class HumanInputFormSubmissionRepository:
     """Repository for fetching and submitting human input forms."""
 
-    def __init__(self, session_factory: sessionmaker | Engine):
-        if isinstance(session_factory, Engine):
-            session_factory = sessionmaker(bind=session_factory)
-        self._session_factory = session_factory
-
     def get_by_token(self, form_token: str) -> HumanInputFormRecord | None:
         query = (
             select(HumanInputFormRecipient)
             .options(selectinload(HumanInputFormRecipient.form))
             .where(HumanInputFormRecipient.access_token == form_token)
         )
-        with self._session_factory(expire_on_commit=False) as session:
+        with session_factory.create_session() as session:
             recipient_model = session.scalars(query).first()
             if recipient_model is None or recipient_model.form is None:
                 return None
@@ -478,7 +471,7 @@ class HumanInputFormSubmissionRepository:
                 HumanInputFormRecipient.recipient_type == recipient_type,
             )
         )
-        with self._session_factory(expire_on_commit=False) as session:
+        with session_factory.create_session() as session:
             recipient_model = session.scalars(query).first()
             if recipient_model is None or recipient_model.form is None:
                 return None
@@ -494,7 +487,7 @@ class HumanInputFormSubmissionRepository:
         submission_user_id: str | None,
         submission_end_user_id: str | None,
     ) -> HumanInputFormRecord:
-        with self._session_factory(expire_on_commit=False) as session, session.begin():
+        with session_factory.create_session() as session, session.begin():
             form_model = session.get(HumanInputForm, form_id)
             if form_model is None:
                 raise FormNotFoundError(f"form not found, id={form_id}")
@@ -524,7 +517,7 @@ class HumanInputFormSubmissionRepository:
         timeout_status: HumanInputFormStatus,
         reason: str | None = None,
     ) -> HumanInputFormRecord:
-        with self._session_factory(expire_on_commit=False) as session, session.begin():
+        with session_factory.create_session() as session, session.begin():
             form_model = session.get(HumanInputForm, form_id)
             if form_model is None:
                 raise FormNotFoundError(f"form not found, id={form_id}")

api/core/repositories/sqlalchemy_workflow_execution_repository.py

@@ -194,6 +194,13 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
 
         # Create a new database session
         with self._session_factory() as session:
+            existing_model = session.get(WorkflowRun, db_model.id)
+            if existing_model:
+                if existing_model.tenant_id != self._tenant_id:
+                    raise ValueError("Unauthorized access to workflow run")
+                # Preserve the original start time for pause/resume flows.
+                db_model.created_at = existing_model.created_at
+
             # SQLAlchemy merge intelligently handles both insert and update operations
             # based on the presence of the primary key
             session.merge(db_model)

api/core/tools/tool_file_manager.py

@@ -10,28 +10,19 @@ from typing import Union
 from uuid import uuid4
 
 import httpx
-from sqlalchemy.orm import Session
 
 from configs import dify_config
+from core.db.session_factory import session_factory
 from core.helper import ssrf_proxy
-from extensions.ext_database import db as global_db
+from dify_graph.file.models import ToolFile as ToolFilePydanticModel
 from extensions.ext_storage import storage
 from models.model import MessageFile
 from models.tools import ToolFile
 
 logger = logging.getLogger(__name__)
 
-from sqlalchemy.engine import Engine
-
 
 class ToolFileManager:
-    _engine: Engine
-
-    def __init__(self, engine: Engine | None = None):
-        if engine is None:
-            engine = global_db.engine
-        self._engine = engine
-
     @staticmethod
     def sign_file(tool_file_id: str, extension: str) -> str:
         """
@@ -89,7 +80,7 @@ class ToolFileManager:
         filepath = f"tools/{tenant_id}/{unique_filename}"
         storage.save(filepath, file_binary)
 
-        with Session(self._engine, expire_on_commit=False) as session:
+        with session_factory.create_session() as session:
             tool_file = ToolFile(
                 user_id=user_id,
                 tenant_id=tenant_id,
@@ -132,7 +123,7 @@ class ToolFileManager:
         filename = f"{unique_name}{extension}"
         filepath = f"tools/{tenant_id}/{filename}"
         storage.save(filepath, blob)
-        with Session(self._engine, expire_on_commit=False) as session:
+        with session_factory.create_session() as session:
             tool_file = ToolFile(
                 user_id=user_id,
                 tenant_id=tenant_id,
@@ -157,7 +148,7 @@ class ToolFileManager:
 
         :return: the binary of the file, mime type
         """
-        with Session(self._engine, expire_on_commit=False) as session:
+        with session_factory.create_session() as session:
             tool_file: ToolFile | None = (
                 session.query(ToolFile)
                 .where(
@@ -181,7 +172,7 @@ class ToolFileManager:
 
         :return: the binary of the file, mime type
         """
-        with Session(self._engine, expire_on_commit=False) as session:
+        with session_factory.create_session() as session:
             message_file: MessageFile | None = (
                 session.query(MessageFile)
                 .where(
@@ -217,7 +208,9 @@ class ToolFileManager:
 
         return blob, tool_file.mimetype
 
-    def get_file_generator_by_tool_file_id(self, tool_file_id: str) -> tuple[Generator | None, ToolFile | None]:
+    def get_file_generator_by_tool_file_id(
+        self, tool_file_id: str
+    ) -> tuple[Generator | None, ToolFilePydanticModel | None]:
         """
         get file binary
 
@@ -225,7 +218,7 @@ class ToolFileManager:
 
         :return: the binary of the file, mime type
         """
-        with Session(self._engine, expire_on_commit=False) as session:
+        with session_factory.create_session() as session:
             tool_file: ToolFile | None = (
                 session.query(ToolFile)
                 .where(
@@ -239,7 +232,7 @@ class ToolFileManager:
 
         stream = storage.load_stream(tool_file.file_key)
 
-        return stream, tool_file
+        return stream, ToolFilePydanticModel.model_validate(tool_file)
 
 
 # init tool_file_parser

api/core/tools/workflow_as_tool/provider.py

@@ -37,6 +37,7 @@ VARIABLE_TO_PARAMETER_TYPE_MAPPING = {
     VariableEntityType.CHECKBOX: ToolParameter.ToolParameterType.BOOLEAN,
     VariableEntityType.FILE: ToolParameter.ToolParameterType.FILE,
     VariableEntityType.FILE_LIST: ToolParameter.ToolParameterType.FILES,
+    VariableEntityType.JSON_OBJECT: ToolParameter.ToolParameterType.OBJECT,
 }
 
 

api/core/workflow/node_factory.py

@@ -6,6 +6,7 @@ from sqlalchemy.orm import Session
 from typing_extensions import override
 
 from configs import dify_config
+from core.app.entities.app_invoke_entities import DifyRunContext
 from core.app.llm.model_access import build_dify_model_access
 from core.datasource.datasource_manager import DatasourceManager
 from core.helper.code_executor.code_executor import (
@@ -19,8 +20,10 @@ from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.rag.index_processor.index_processor import IndexProcessor
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from core.rag.summary_index.summary_index import SummaryIndex
+from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
 from core.tools.tool_file_manager import ToolFileManager
 from dify_graph.entities.graph_config import NodeConfigDict
+from dify_graph.entities.graph_init_params import DIFY_RUN_CONTEXT_KEY
 from dify_graph.enums import NodeType, SystemVariableKey
 from dify_graph.file.file_manager import file_manager
 from dify_graph.graph.graph import NodeFactory
@@ -34,6 +37,7 @@ from dify_graph.nodes.code.limits import CodeNodeLimits
 from dify_graph.nodes.datasource import DatasourceNode
 from dify_graph.nodes.document_extractor import DocumentExtractorNode, UnstructuredApiConfig
 from dify_graph.nodes.http_request import HttpRequestNode, build_http_request_config
+from dify_graph.nodes.human_input.human_input_node import HumanInputNode
 from dify_graph.nodes.knowledge_index.knowledge_index_node import KnowledgeIndexNode
 from dify_graph.nodes.knowledge_retrieval.knowledge_retrieval_node import KnowledgeRetrievalNode
 from dify_graph.nodes.llm.entities import ModelConfig
@@ -46,6 +50,7 @@ from dify_graph.nodes.template_transform.template_renderer import (
     CodeExecutorJinja2TemplateRenderer,
 )
 from dify_graph.nodes.template_transform.template_transform_node import TemplateTransformNode
+from dify_graph.nodes.tool.tool_node import ToolNode
 from dify_graph.variables.segments import StringSegment
 from extensions.ext_database import db
 from models.model import Conversation
@@ -108,6 +113,7 @@ class DifyNodeFactory(NodeFactory):
     ) -> None:
         self.graph_init_params = graph_init_params
         self.graph_runtime_state = graph_runtime_state
+        self._dify_context = self._resolve_dify_context(graph_init_params.run_context)
         self._code_executor: WorkflowCodeExecutor = DefaultWorkflowCodeExecutor()
         self._code_limits = CodeNodeLimits(
             max_string_length=dify_config.CODE_MAX_STRING_LENGTH,
@@ -139,7 +145,16 @@ class DifyNodeFactory(NodeFactory):
             ssrf_default_max_retries=dify_config.SSRF_DEFAULT_MAX_RETRIES,
         )
 
-        self._llm_credentials_provider, self._llm_model_factory = build_dify_model_access(graph_init_params.tenant_id)
+        self._llm_credentials_provider, self._llm_model_factory = build_dify_model_access(self._dify_context.tenant_id)
+
+    @staticmethod
+    def _resolve_dify_context(run_context: Mapping[str, Any]) -> DifyRunContext:
+        raw_ctx = run_context.get(DIFY_RUN_CONTEXT_KEY)
+        if raw_ctx is None:
+            raise ValueError(f"run_context missing required key: {DIFY_RUN_CONTEXT_KEY}")
+        if isinstance(raw_ctx, DifyRunContext):
+            return raw_ctx
+        return DifyRunContext.model_validate(raw_ctx)
 
     @override
     def create_node(self, node_config: NodeConfigDict) -> Node:
@@ -205,6 +220,15 @@ class DifyNodeFactory(NodeFactory):
                 file_manager=self._http_request_file_manager,
             )
 
+        if node_type == NodeType.HUMAN_INPUT:
+            return HumanInputNode(
+                id=node_id,
+                config=node_config,
+                graph_init_params=self.graph_init_params,
+                graph_runtime_state=self.graph_runtime_state,
+                form_repository=HumanInputFormRepositoryImpl(tenant_id=self._dify_context.tenant_id),
+            )
+
         if node_type == NodeType.KNOWLEDGE_INDEX:
             return KnowledgeIndexNode(
                 id=node_id,
@@ -227,6 +251,7 @@ class DifyNodeFactory(NodeFactory):
                 model_factory=self._llm_model_factory,
                 model_instance=model_instance,
                 memory=memory,
+                http_client=self._http_request_http_client,
             )
 
         if node_type == NodeType.DATASOURCE:
@@ -254,6 +279,7 @@ class DifyNodeFactory(NodeFactory):
                 graph_init_params=self.graph_init_params,
                 graph_runtime_state=self.graph_runtime_state,
                 unstructured_api_config=self._document_extractor_unstructured_api_config,
+                http_client=self._http_request_http_client,
             )
 
         if node_type == NodeType.QUESTION_CLASSIFIER:
@@ -268,6 +294,7 @@ class DifyNodeFactory(NodeFactory):
                 model_factory=self._llm_model_factory,
                 model_instance=model_instance,
                 memory=memory,
+                http_client=self._http_request_http_client,
             )
 
         if node_type == NodeType.PARAMETER_EXTRACTOR:
@@ -284,6 +311,15 @@ class DifyNodeFactory(NodeFactory):
                 memory=memory,
             )
 
+        if node_type == NodeType.TOOL:
+            return ToolNode(
+                id=node_id,
+                config=node_config,
+                graph_init_params=self.graph_init_params,
+                graph_runtime_state=self.graph_runtime_state,
+                tool_file_manager_factory=self._http_request_tool_file_manager_factory(),
+            )
+
         return node_class(
             id=node_id,
             config=node_config,
@@ -344,7 +380,7 @@ class DifyNodeFactory(NodeFactory):
         )
         return fetch_memory(
             conversation_id=conversation_id,
-            app_id=self.graph_init_params.app_id,
+            app_id=self._dify_context.app_id,
             node_data_memory=node_memory,
             model_instance=model_instance,
         )

api/core/workflow/workflow_entry.py

@@ -5,26 +5,26 @@ from typing import Any, cast
 
 from configs import dify_config
 from core.app.apps.exc import GenerateTaskStoppedError
-from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, build_dify_run_context
 from core.app.workflow.layers.llm_quota import LLMQuotaLayer
 from core.app.workflow.layers.observability import ObservabilityLayer
 from core.workflow.node_factory import DifyNodeFactory
 from dify_graph.constants import ENVIRONMENT_VARIABLE_NODE_ID
 from dify_graph.entities import GraphInitParams
 from dify_graph.entities.graph_config import NodeConfigData, NodeConfigDict
-from dify_graph.enums import UserFrom
 from dify_graph.errors import WorkflowNodeRunFailedError
 from dify_graph.file.models import File
 from dify_graph.graph import Graph
 from dify_graph.graph_engine import GraphEngine, GraphEngineConfig
 from dify_graph.graph_engine.command_channels import InMemoryChannel
 from dify_graph.graph_engine.layers import DebugLoggingLayer, ExecutionLimitsLayer
+from dify_graph.graph_engine.layers.base import GraphEngineLayer
 from dify_graph.graph_engine.protocols.command_channel import CommandChannel
 from dify_graph.graph_events import GraphEngineEvent, GraphNodeEventBase, GraphRunFailedEvent
 from dify_graph.nodes import NodeType
 from dify_graph.nodes.base.node import Node
 from dify_graph.nodes.node_mapping import NODE_TYPE_CLASSES_MAPPING
-from dify_graph.runtime import GraphRuntimeState, VariablePool
+from dify_graph.runtime import ChildGraphNotFoundError, GraphRuntimeState, VariablePool
 from dify_graph.system_variable import SystemVariable
 from dify_graph.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
 from extensions.otel.runtime import is_instrument_flag_enabled
@@ -34,6 +34,66 @@ from models.workflow import Workflow
 logger = logging.getLogger(__name__)
 
 
+class _WorkflowChildEngineBuilder:
+    @staticmethod
+    def _has_node_id(graph_config: Mapping[str, Any], node_id: str) -> bool | None:
+        """
+        Return whether `graph_config["nodes"]` contains the given node id.
+
+        Returns `None` when the nodes payload shape is unexpected, so graph-level
+        validation can surface the original configuration error.
+        """
+        nodes = graph_config.get("nodes")
+        if not isinstance(nodes, list):
+            return None
+
+        for node in nodes:
+            if not isinstance(node, Mapping):
+                return None
+            current_id = node.get("id")
+            if isinstance(current_id, str) and current_id == node_id:
+                return True
+        return False
+
+    def build_child_engine(
+        self,
+        *,
+        workflow_id: str,
+        graph_init_params: GraphInitParams,
+        graph_runtime_state: GraphRuntimeState,
+        graph_config: Mapping[str, Any],
+        root_node_id: str,
+        layers: Sequence[object] = (),
+    ) -> GraphEngine:
+        node_factory = DifyNodeFactory(
+            graph_init_params=graph_init_params,
+            graph_runtime_state=graph_runtime_state,
+        )
+
+        has_root_node = self._has_node_id(graph_config=graph_config, node_id=root_node_id)
+        if has_root_node is False:
+            raise ChildGraphNotFoundError(f"child graph root node '{root_node_id}' not found")
+
+        child_graph = Graph.init(
+            graph_config=graph_config,
+            node_factory=node_factory,
+            root_node_id=root_node_id,
+        )
+
+        child_engine = GraphEngine(
+            workflow_id=workflow_id,
+            graph=child_graph,
+            graph_runtime_state=graph_runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+            child_engine_builder=self,
+        )
+        child_engine.layer(LLMQuotaLayer())
+        for layer in layers:
+            child_engine.layer(cast(GraphEngineLayer, layer))
+        return child_engine
+
+
 class WorkflowEntry:
     def __init__(
         self,
@@ -77,6 +137,7 @@ class WorkflowEntry:
             command_channel = InMemoryChannel()
 
         self.command_channel = command_channel
+        self._child_engine_builder = _WorkflowChildEngineBuilder()
         self.graph_engine = GraphEngine(
             workflow_id=workflow_id,
             graph=graph,
@@ -88,6 +149,7 @@ class WorkflowEntry:
                 scale_up_threshold=dify_config.GRAPH_ENGINE_SCALE_UP_THRESHOLD,
                 scale_down_idle_time=dify_config.GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME,
             ),
+            child_engine_builder=self._child_engine_builder,
         )
 
         # Add debug logging layer when in debug mode
@@ -154,13 +216,15 @@ class WorkflowEntry:
 
         # init graph init params and runtime state
         graph_init_params = GraphInitParams(
-            tenant_id=workflow.tenant_id,
-            app_id=workflow.app_id,
             workflow_id=workflow.id,
             graph_config=workflow.graph_dict,
-            user_id=user_id,
-            user_from=UserFrom.ACCOUNT,
-            invoke_from=InvokeFrom.DEBUGGER,
+            run_context=build_dify_run_context(
+                tenant_id=workflow.tenant_id,
+                app_id=workflow.app_id,
+                user_id=user_id,
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+            ),
             call_depth=0,
         )
         graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
@@ -293,13 +357,15 @@ class WorkflowEntry:
 
         # init graph init params and runtime state
         graph_init_params = GraphInitParams(
-            tenant_id=tenant_id,
-            app_id="",
             workflow_id="",
             graph_config=graph_dict,
-            user_id=user_id,
-            user_from=UserFrom.ACCOUNT,
-            invoke_from=InvokeFrom.DEBUGGER,
+            run_context=build_dify_run_context(
+                tenant_id=tenant_id,
+                app_id="",
+                user_id=user_id,
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+            ),
             call_depth=0,
         )
         graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())

api/dify_graph/entities/graph_init_params.py

@@ -3,7 +3,7 @@ from typing import Any
 
 from pydantic import BaseModel, Field
 
-from dify_graph.enums import InvokeFrom, UserFrom
+DIFY_RUN_CONTEXT_KEY = "_dify"
 
 
 class GraphInitParams(BaseModel):
@@ -18,11 +18,7 @@ class GraphInitParams(BaseModel):
     """
 
     # init params
-    tenant_id: str = Field(..., description="tenant / workspace id")
-    app_id: str = Field(..., description="app id")
     workflow_id: str = Field(..., description="workflow id")
     graph_config: Mapping[str, Any] = Field(..., description="graph config")
-    user_id: str = Field(..., description="user id")
-    user_from: UserFrom = Field(..., description="user from, account or end-user")
-    invoke_from: InvokeFrom = Field(..., description="invoke from, service-api, web-app, explore or debugger")
+    run_context: Mapping[str, Any] = Field(..., description="runtime context")
     call_depth: int = Field(..., description="call depth")

api/dify_graph/enums.py

@@ -33,39 +33,6 @@ class SystemVariableKey(StrEnum):
     INVOKE_FROM = "invoke_from"
 
 
-class UserFrom(StrEnum):
-    ACCOUNT = "account"
-    END_USER = "end-user"
-
-
-class InvokeFrom(StrEnum):
-    SERVICE_API = "service-api"
-    WEB_APP = "web-app"
-    TRIGGER = "trigger"
-    EXPLORE = "explore"
-    DEBUGGER = "debugger"
-    PUBLISHED_PIPELINE = "published"
-    VALIDATION = "validation"
-
-    @classmethod
-    def value_of(cls, value: str) -> "InvokeFrom":
-        return cls(value)
-
-    def to_source(self) -> str:
-        """Get source of invoke from.
-
-        :return: source
-        """
-        source_mapping = {
-            InvokeFrom.WEB_APP: "web_app",
-            InvokeFrom.DEBUGGER: "dev",
-            InvokeFrom.EXPLORE: "explore_app",
-            InvokeFrom.TRIGGER: "trigger",
-            InvokeFrom.SERVICE_API: "api",
-        }
-        return source_mapping.get(self, "dev")
-
-
 class NodeType(StrEnum):
     START = "start"
     END = "end"

api/dify_graph/file/models.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Mapping, Sequence
 from typing import Any
+from uuid import UUID, uuid4
 
 from pydantic import BaseModel, Field, model_validator
 
@@ -43,6 +44,24 @@ class FileUploadConfig(BaseModel):
     number_limits: int = 0
 
 
+class ToolFile(BaseModel):
+    id: UUID = Field(default_factory=uuid4, description="Unique identifier for the file")
+    user_id: UUID = Field(..., description="ID of the user who owns this file")
+    tenant_id: UUID = Field(..., description="ID of the tenant/organization")
+    conversation_id: UUID | None = Field(None, description="ID of the associated conversation")
+    file_key: str = Field(..., max_length=255, description="Storage key for the file")
+    mimetype: str = Field(..., max_length=255, description="MIME type of the file")
+    original_url: str | None = Field(
+        None, max_length=2048, description="Original URL if file was fetched from external source"
+    )
+    name: str = Field(default="", max_length=255, description="Display name of the file")
+    size: int = Field(default=-1, ge=-1, description="File size in bytes (-1 if unknown)")
+
+    class Config:
+        from_attributes = True  # Enable ORM mode for SQLAlchemy compatibility
+        populate_by_name = True
+
+
 class File(BaseModel):
     # NOTE: dify_model_identity is a special identifier used to distinguish between
     # new and old data formats during serialization and deserialization.

api/dify_graph/graph_engine/graph_engine.py

@@ -9,7 +9,7 @@ from __future__ import annotations
 
 import logging
 import queue
-from collections.abc import Generator
+from collections.abc import Generator, Mapping
 from typing import TYPE_CHECKING, cast, final
 
 from dify_graph.context import capture_current_context
@@ -27,6 +27,7 @@ from dify_graph.graph_events import (
     GraphRunSucceededEvent,
 )
 from dify_graph.runtime import GraphRuntimeState, ReadOnlyGraphRuntimeStateWrapper
+from dify_graph.runtime.graph_runtime_state import ChildGraphEngineBuilderProtocol
 
 if TYPE_CHECKING:  # pragma: no cover - used only for static analysis
     from dify_graph.runtime.graph_runtime_state import GraphProtocol
@@ -49,6 +50,7 @@ from .protocols.command_channel import CommandChannel
 from .worker_management import WorkerPool
 
 if TYPE_CHECKING:
+    from dify_graph.entities import GraphInitParams
     from dify_graph.graph_engine.domain.graph_execution import GraphExecution
     from dify_graph.graph_engine.response_coordinator import ResponseStreamCoordinator
 
@@ -74,6 +76,7 @@ class GraphEngine:
         graph_runtime_state: GraphRuntimeState,
         command_channel: CommandChannel,
         config: GraphEngineConfig = _DEFAULT_CONFIG,
+        child_engine_builder: ChildGraphEngineBuilderProtocol | None = None,
     ) -> None:
         """Initialize the graph engine with all subsystems and dependencies."""
 
@@ -83,6 +86,9 @@ class GraphEngine:
         self._graph_runtime_state.configure(graph=cast("GraphProtocol", graph))
         self._command_channel = command_channel
         self._config = config
+        self._child_engine_builder = child_engine_builder
+        if child_engine_builder is not None:
+            self._graph_runtime_state.bind_child_engine_builder(child_engine_builder)
 
         # Graph execution tracks the overall execution state
         self._graph_execution = cast("GraphExecution", self._graph_runtime_state.graph_execution)
@@ -214,6 +220,25 @@ class GraphEngine:
         self._bind_layer_context(layer)
         return self
 
+    def create_child_engine(
+        self,
+        *,
+        workflow_id: str,
+        graph_init_params: GraphInitParams,
+        graph_runtime_state: GraphRuntimeState,
+        graph_config: dict[str, object] | Mapping[str, object],
+        root_node_id: str,
+        layers: list[GraphEngineLayer] | tuple[GraphEngineLayer, ...] = (),
+    ) -> GraphEngine:
+        return self._graph_runtime_state.create_child_engine(
+            workflow_id=workflow_id,
+            graph_init_params=graph_init_params,
+            graph_runtime_state=graph_runtime_state,
+            graph_config=graph_config,
+            root_node_id=root_node_id,
+            layers=layers,
+        )
+
     def run(self) -> Generator[GraphEngineEvent, None, None]:
         """
         Execute the graph using the modular architecture.

api/dify_graph/nodes/agent/agent_node.py

@@ -80,9 +80,11 @@ class AgentNode(Node[AgentNodeData]):
     def _run(self) -> Generator[NodeEventBase, None, None]:
         from core.plugin.impl.exc import PluginDaemonClientSideError
 
+        dify_ctx = self.require_dify_context()
+
         try:
             strategy = get_plugin_agent_strategy(
-                tenant_id=self.tenant_id,
+                tenant_id=dify_ctx.tenant_id,
                 agent_strategy_provider_name=self.node_data.agent_strategy_provider_name,
                 agent_strategy_name=self.node_data.agent_strategy_name,
             )
@@ -120,8 +122,8 @@ class AgentNode(Node[AgentNodeData]):
         try:
             message_stream = strategy.invoke(
                 params=parameters,
-                user_id=self.user_id,
-                app_id=self.app_id,
+                user_id=dify_ctx.user_id,
+                app_id=dify_ctx.app_id,
                 conversation_id=conversation_id.text if conversation_id else None,
                 credentials=credentials,
             )
@@ -144,8 +146,8 @@ class AgentNode(Node[AgentNodeData]):
                     "agent_strategy": self.node_data.agent_strategy_name,
                 },
                 parameters_for_log=parameters_for_log,
-                user_id=self.user_id,
-                tenant_id=self.tenant_id,
+                user_id=dify_ctx.user_id,
+                tenant_id=dify_ctx.tenant_id,
                 node_type=self.node_type,
                 node_id=self._node_id,
                 node_execution_id=self.id,
@@ -283,8 +285,13 @@ class AgentNode(Node[AgentNodeData]):
                         runtime_variable_pool: VariablePool | None = None
                         if node_data.version != "1" or node_data.tool_node_version is not None:
                             runtime_variable_pool = variable_pool
+                        dify_ctx = self.require_dify_context()
                         tool_runtime = ToolManager.get_agent_tool_runtime(
-                            self.tenant_id, self.app_id, entity, self.invoke_from, runtime_variable_pool
+                            dify_ctx.tenant_id,
+                            dify_ctx.app_id,
+                            entity,
+                            dify_ctx.invoke_from,
+                            runtime_variable_pool,
                         )
                         if tool_runtime.entity.description:
                             tool_runtime.entity.description.llm = (
@@ -396,7 +403,8 @@ class AgentNode(Node[AgentNodeData]):
         from core.plugin.impl.plugin import PluginInstaller
 
         manager = PluginInstaller()
-        plugins = manager.list_plugins(self.tenant_id)
+        dify_ctx = self.require_dify_context()
+        plugins = manager.list_plugins(dify_ctx.tenant_id)
         try:
             current_plugin = next(
                 plugin
@@ -417,8 +425,11 @@ class AgentNode(Node[AgentNodeData]):
             return None
         conversation_id = conversation_id_variable.value
 
+        dify_ctx = self.require_dify_context()
         with Session(db.engine, expire_on_commit=False) as session:
-            stmt = select(Conversation).where(Conversation.app_id == self.app_id, Conversation.id == conversation_id)
+            stmt = select(Conversation).where(
+                Conversation.app_id == dify_ctx.app_id, Conversation.id == conversation_id
+            )
             conversation = session.scalar(stmt)
 
             if not conversation:
@@ -429,9 +440,10 @@ class AgentNode(Node[AgentNodeData]):
         return memory
 
     def _fetch_model(self, value: dict[str, Any]) -> tuple[ModelInstance, AIModelEntity | None]:
+        dify_ctx = self.require_dify_context()
         provider_manager = ProviderManager()
         provider_model_bundle = provider_manager.get_provider_model_bundle(
-            tenant_id=self.tenant_id, provider=value.get("provider", ""), model_type=ModelType.LLM
+            tenant_id=dify_ctx.tenant_id, provider=value.get("provider", ""), model_type=ModelType.LLM
         )
         model_name = value.get("model", "")
         model_credentials = provider_model_bundle.configuration.get_current_credentials(
@@ -440,7 +452,7 @@ class AgentNode(Node[AgentNodeData]):
         provider_name = provider_model_bundle.configuration.provider.provider
         model_type_instance = provider_model_bundle.model_type_instance
         model_instance = ModelManager().get_model_instance(
-            tenant_id=self.tenant_id,
+            tenant_id=dify_ctx.tenant_id,
             provider=provider_name,
             model_type=ModelType(value.get("model_type", "")),
             model=model_name,

api/dify_graph/nodes/base/node.py

@@ -8,10 +8,11 @@ from abc import abstractmethod
 from collections.abc import Generator, Mapping, Sequence
 from functools import singledispatchmethod
 from types import MappingProxyType
-from typing import Any, ClassVar, Generic, TypeVar, cast, get_args, get_origin
+from typing import Any, ClassVar, Generic, Protocol, TypeVar, cast, get_args, get_origin
 from uuid import uuid4
 
 from dify_graph.entities import AgentNodeStrategyInit, GraphInitParams
+from dify_graph.entities.graph_init_params import DIFY_RUN_CONTEXT_KEY
 from dify_graph.enums import (
     ErrorStrategy,
     NodeExecutionType,
@@ -64,10 +65,28 @@ from libs.datetime_utils import naive_utc_now
 from .entities import BaseNodeData, RetryConfig
 
 NodeDataT = TypeVar("NodeDataT", bound=BaseNodeData)
+_MISSING_RUN_CONTEXT_VALUE = object()
 
 logger = logging.getLogger(__name__)
 
 
+class DifyRunContextProtocol(Protocol):
+    tenant_id: str
+    app_id: str
+    user_id: str
+    user_from: Any
+    invoke_from: Any
+
+
+class _MappingDifyRunContext:
+    def __init__(self, mapping: Mapping[str, Any]) -> None:
+        self.tenant_id = str(mapping["tenant_id"])
+        self.app_id = str(mapping["app_id"])
+        self.user_id = str(mapping["user_id"])
+        self.user_from = mapping["user_from"]
+        self.invoke_from = mapping["invoke_from"]
+
+
 class Node(Generic[NodeDataT]):
     """BaseNode serves as the foundational class for all node implementations.
 
@@ -227,14 +246,10 @@ class Node(Generic[NodeDataT]):
         graph_runtime_state: GraphRuntimeState,
     ) -> None:
         self._graph_init_params = graph_init_params
+        self._run_context = MappingProxyType(dict(graph_init_params.run_context))
         self.id = id
-        self.tenant_id = graph_init_params.tenant_id
-        self.app_id = graph_init_params.app_id
         self.workflow_id = graph_init_params.workflow_id
         self.graph_config = graph_init_params.graph_config
-        self.user_id = graph_init_params.user_id
-        self.user_from = graph_init_params.user_from
-        self.invoke_from = graph_init_params.invoke_from
         self.workflow_call_depth = graph_init_params.call_depth
         self.graph_runtime_state = graph_runtime_state
         self.state: NodeState = NodeState.UNKNOWN  # node execution state
@@ -263,6 +278,38 @@ class Node(Generic[NodeDataT]):
     def graph_init_params(self) -> GraphInitParams:
         return self._graph_init_params
 
+    @property
+    def run_context(self) -> Mapping[str, Any]:
+        return self._run_context
+
+    def get_run_context_value(self, key: str, default: Any = None) -> Any:
+        return self._run_context.get(key, default)
+
+    def require_run_context_value(self, key: str) -> Any:
+        value = self.get_run_context_value(key, _MISSING_RUN_CONTEXT_VALUE)
+        if value is _MISSING_RUN_CONTEXT_VALUE:
+            raise ValueError(f"run_context missing required key: {key}")
+        return value
+
+    def require_dify_context(self) -> DifyRunContextProtocol:
+        raw_ctx = self.require_run_context_value(DIFY_RUN_CONTEXT_KEY)
+        if raw_ctx is None:
+            raise ValueError(f"run_context missing required key: {DIFY_RUN_CONTEXT_KEY}")
+
+        if isinstance(raw_ctx, Mapping):
+            missing_keys = [
+                key for key in ("tenant_id", "app_id", "user_id", "user_from", "invoke_from") if key not in raw_ctx
+            ]
+            if missing_keys:
+                raise ValueError(f"dify context missing required keys: {', '.join(missing_keys)}")
+            return _MappingDifyRunContext(raw_ctx)
+
+        for attr in ("tenant_id", "app_id", "user_id", "user_from", "invoke_from"):
+            if not hasattr(raw_ctx, attr):
+                raise TypeError(f"invalid dify context object, missing attribute: {attr}")
+
+        return cast(DifyRunContextProtocol, raw_ctx)
+
     @property
     def execution_id(self) -> str:
         return self._node_execution_id

api/dify_graph/nodes/datasource/datasource_node.py

@@ -52,6 +52,7 @@ class DatasourceNode(Node[DatasourceNodeData]):
         Run the datasource node
         """
 
+        dify_ctx = self.require_dify_context()
         node_data = self.node_data
         variable_pool = self.graph_runtime_state.variable_pool
         datasource_type_segment = variable_pool.get(["sys", SystemVariableKey.DATASOURCE_TYPE])
@@ -75,7 +76,7 @@ class DatasourceNode(Node[DatasourceNodeData]):
         datasource_info["icon"] = self.datasource_manager.get_icon_url(
             provider_id=provider_id,
             datasource_name=node_data.datasource_name or "",
-            tenant_id=self.tenant_id,
+            tenant_id=dify_ctx.tenant_id,
             datasource_type=datasource_type.value,
         )
 
@@ -104,11 +105,11 @@ class DatasourceNode(Node[DatasourceNodeData]):
 
                     yield from self.datasource_manager.stream_node_events(
                         node_id=self._node_id,
-                        user_id=self.user_id,
+                        user_id=dify_ctx.user_id,
                         datasource_name=node_data.datasource_name or "",
                         datasource_type=datasource_type.value,
                         provider_id=provider_id,
-                        tenant_id=self.tenant_id,
+                        tenant_id=dify_ctx.tenant_id,
                         provider=node_data.provider_name,
                         plugin_id=node_data.plugin_id,
                         credential_id=credential_id,
@@ -136,7 +137,7 @@ class DatasourceNode(Node[DatasourceNodeData]):
                         raise DatasourceNodeError("File is not exist")
 
                     file_info = self.datasource_manager.get_upload_file_by_id(
-                        file_id=related_id, tenant_id=self.tenant_id
+                        file_id=related_id, tenant_id=dify_ctx.tenant_id
                     )
                     variable_pool.add([self._node_id, "file"], file_info)
                     # variable_pool.add([self.node_id, "file"], file_info.to_dict())

api/dify_graph/nodes/document_extractor/node.py

@@ -4,6 +4,7 @@ import json
 import logging
 import os
 import tempfile
+import zipfile
 from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
@@ -20,11 +21,11 @@ from docx.oxml.text.paragraph import CT_P
 from docx.table import Table
 from docx.text.paragraph import Paragraph
 
-from core.helper import ssrf_proxy
 from dify_graph.enums import NodeType, WorkflowNodeExecutionStatus
 from dify_graph.file import File, FileTransferMethod, file_manager
 from dify_graph.node_events import NodeRunResult
 from dify_graph.nodes.base.node import Node
+from dify_graph.nodes.protocols import HttpClientProtocol
 from dify_graph.variables import ArrayFileSegment
 from dify_graph.variables.segments import ArrayStringSegment, FileSegment
 
@@ -58,6 +59,7 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
         graph_runtime_state: "GraphRuntimeState",
         *,
         unstructured_api_config: UnstructuredApiConfig | None = None,
+        http_client: HttpClientProtocol,
     ) -> None:
         super().__init__(
             id=id,
@@ -66,6 +68,7 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
             graph_runtime_state=graph_runtime_state,
         )
         self._unstructured_api_config = unstructured_api_config or UnstructuredApiConfig()
+        self._http_client = http_client
 
     def _run(self):
         variable_selector = self.node_data.variable_selector
@@ -80,12 +83,24 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
 
         value = variable.value
         inputs = {"variable_selector": variable_selector}
+        if isinstance(value, list):
+            value = list(filter(lambda x: x, value))
         process_data = {"documents": value if isinstance(value, list) else [value]}
 
+        if not value:
+            return NodeRunResult(
+                status=WorkflowNodeExecutionStatus.SUCCEEDED,
+                inputs=inputs,
+                process_data=process_data,
+                outputs={"text": ArrayStringSegment(value=[])},
+            )
+
         try:
             if isinstance(value, list):
                 extracted_text_list = [
-                    _extract_text_from_file(file, unstructured_api_config=self._unstructured_api_config)
+                    _extract_text_from_file(
+                        self._http_client, file, unstructured_api_config=self._unstructured_api_config
+                    )
                     for file in value
                 ]
                 return NodeRunResult(
@@ -95,7 +110,9 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
                     outputs={"text": ArrayStringSegment(value=extracted_text_list)},
                 )
             elif isinstance(value, File):
-                extracted_text = _extract_text_from_file(value, unstructured_api_config=self._unstructured_api_config)
+                extracted_text = _extract_text_from_file(
+                    self._http_client, value, unstructured_api_config=self._unstructured_api_config
+                )
                 return NodeRunResult(
                     status=WorkflowNodeExecutionStatus.SUCCEEDED,
                     inputs=inputs,
@@ -105,6 +122,7 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
             else:
                 raise DocumentExtractorError(f"Unsupported variable type: {type(value)}")
         except DocumentExtractorError as e:
+            logger.warning(e, exc_info=True)
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.FAILED,
                 error=str(e),
@@ -379,6 +397,32 @@ def parser_docx_part(block, doc: Document, content_items, i):
         content_items.append((i, "table", Table(block, doc)))
 
 
+def _normalize_docx_zip(file_content: bytes) -> bytes:
+    """
+    Some DOCX files (e.g. exported by Evernote on Windows) are malformed:
+    ZIP entry names use backslash (\\) as path separator instead of the forward
+    slash (/) required by both the ZIP spec and OOXML.  On Linux/Mac the entry
+    "word\\document.xml" is never found when python-docx looks for
+    "word/document.xml", which triggers a KeyError about a missing relationship.
+
+    This function rewrites the ZIP in-memory, normalizing all entry names to
+    use forward slashes without touching any actual document content.
+    """
+    try:
+        with zipfile.ZipFile(io.BytesIO(file_content), "r") as zin:
+            out_buf = io.BytesIO()
+            with zipfile.ZipFile(out_buf, "w", compression=zipfile.ZIP_DEFLATED) as zout:
+                for item in zin.infolist():
+                    data = zin.read(item.filename)
+                    # Normalize backslash path separators to forward slash
+                    item.filename = item.filename.replace("\\", "/")
+                    zout.writestr(item, data)
+            return out_buf.getvalue()
+    except zipfile.BadZipFile:
+        # Not a valid zip — return as-is and let python-docx report the real error
+        return file_content
+
+
 def _extract_text_from_docx(file_content: bytes) -> str:
     """
     Extract text from a DOCX file.
@@ -386,7 +430,15 @@ def _extract_text_from_docx(file_content: bytes) -> str:
     """
     try:
         doc_file = io.BytesIO(file_content)
-        doc = docx.Document(doc_file)
+        try:
+            doc = docx.Document(doc_file)
+        except Exception as e:
+            logger.warning("Failed to parse DOCX, attempting to normalize ZIP entry paths: %s", e)
+            # Some DOCX files exported by tools like Evernote on Windows use
+            # backslash path separators in ZIP entries and/or single-quoted XML
+            # attributes, both of which break python-docx on Linux. Normalize and retry.
+            file_content = _normalize_docx_zip(file_content)
+            doc = docx.Document(io.BytesIO(file_content))
         text = []
 
         # Keep track of paragraph and table positions
@@ -439,13 +491,13 @@ def _extract_text_from_docx(file_content: bytes) -> str:
         raise TextExtractionError(f"Failed to extract text from DOCX: {str(e)}") from e
 
 
-def _download_file_content(file: File) -> bytes:
+def _download_file_content(http_client: HttpClientProtocol, file: File) -> bytes:
     """Download the content of a file based on its transfer method."""
     try:
         if file.transfer_method == FileTransferMethod.REMOTE_URL:
             if file.remote_url is None:
                 raise FileDownloadError("Missing URL for remote file")
-            response = ssrf_proxy.get(file.remote_url)
+            response = http_client.get(file.remote_url)
             response.raise_for_status()
             return response.content
         else:
@@ -454,8 +506,10 @@ def _download_file_content(file: File) -> bytes:
         raise FileDownloadError(f"Error downloading file: {str(e)}") from e
 
 
-def _extract_text_from_file(file: File, *, unstructured_api_config: UnstructuredApiConfig) -> str:
-    file_content = _download_file_content(file)
+def _extract_text_from_file(
+    http_client: HttpClientProtocol, file: File, *, unstructured_api_config: UnstructuredApiConfig
+) -> str:
+    file_content = _download_file_content(http_client, file)
     if file.extension:
         extracted_text = _extract_text_by_file_extension(
             file_content=file_content,

api/dify_graph/nodes/http_request/node.py

@@ -212,6 +212,7 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
         """
         Extract files from response by checking both Content-Type header and URL
         """
+        dify_ctx = self.require_dify_context()
         files: list[File] = []
         is_file = response.is_file
         content_type = response.content_type
@@ -236,8 +237,8 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
         tool_file_manager = self._tool_file_manager_factory()
 
         tool_file = tool_file_manager.create_file_by_raw(
-            user_id=self.user_id,
-            tenant_id=self.tenant_id,
+            user_id=dify_ctx.user_id,
+            tenant_id=dify_ctx.tenant_id,
             conversation_id=None,
             file_binary=content,
             mimetype=mime_type,
@@ -249,7 +250,7 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
         }
         file = file_factory.build_from_mapping(
             mapping=mapping,
-            tenant_id=self.tenant_id,
+            tenant_id=dify_ctx.tenant_id,
         )
         files.append(file)
 

api/dify_graph/nodes/human_input/human_input_node.py

@@ -3,9 +3,8 @@ import logging
 from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
-from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
 from dify_graph.entities.pause_reason import HumanInputRequired
-from dify_graph.enums import InvokeFrom, NodeExecutionType, NodeType, WorkflowNodeExecutionStatus
+from dify_graph.enums import NodeExecutionType, NodeType, WorkflowNodeExecutionStatus
 from dify_graph.node_events import (
     HumanInputFormFilledEvent,
     HumanInputFormTimeoutEvent,
@@ -21,7 +20,6 @@ from dify_graph.repositories.human_input_form_repository import (
     HumanInputFormRepository,
 )
 from dify_graph.workflow_type_encoder import WorkflowRuntimeTypeConverter
-from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 
 from .entities import DeliveryChannelConfig, HumanInputNodeData, apply_debug_email_recipient
@@ -33,6 +31,8 @@ if TYPE_CHECKING:
 
 
 _SELECTED_BRANCH_KEY = "selected_branch"
+_INVOKE_FROM_DEBUGGER = "debugger"
+_INVOKE_FROM_EXPLORE = "explore"
 
 
 logger = logging.getLogger(__name__)
@@ -66,7 +66,7 @@ class HumanInputNode(Node[HumanInputNodeData]):
         config: Mapping[str, Any],
         graph_init_params: "GraphInitParams",
         graph_runtime_state: "GraphRuntimeState",
-        form_repository: HumanInputFormRepository | None = None,
+        form_repository: HumanInputFormRepository,
     ) -> None:
         super().__init__(
             id=id,
@@ -74,11 +74,6 @@ class HumanInputNode(Node[HumanInputNodeData]):
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )
-        if form_repository is None:
-            form_repository = HumanInputFormRepositoryImpl(
-                session_factory=db.engine,
-                tenant_id=self.tenant_id,
-            )
         self._form_repository = form_repository
 
     @classmethod
@@ -162,30 +157,39 @@ class HumanInputNode(Node[HumanInputNodeData]):
         return resolved_defaults
 
     def _should_require_console_recipient(self) -> bool:
-        if self.invoke_from == InvokeFrom.DEBUGGER:
+        invoke_from = self._invoke_from_value()
+        if invoke_from == _INVOKE_FROM_DEBUGGER:
             return True
-        if self.invoke_from == InvokeFrom.EXPLORE:
+        if invoke_from == _INVOKE_FROM_EXPLORE:
             return self._node_data.is_webapp_enabled()
         return False
 
     def _display_in_ui(self) -> bool:
-        if self.invoke_from == InvokeFrom.DEBUGGER:
+        if self._invoke_from_value() == _INVOKE_FROM_DEBUGGER:
             return True
         return self._node_data.is_webapp_enabled()
 
     def _effective_delivery_methods(self) -> Sequence[DeliveryChannelConfig]:
+        dify_ctx = self.require_dify_context()
+        invoke_from = self._invoke_from_value()
         enabled_methods = [method for method in self._node_data.delivery_methods if method.enabled]
-        if self.invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE}:
+        if invoke_from in {_INVOKE_FROM_DEBUGGER, _INVOKE_FROM_EXPLORE}:
             enabled_methods = [method for method in enabled_methods if method.type != DeliveryMethodType.WEBAPP]
         return [
             apply_debug_email_recipient(
                 method,
-                enabled=self.invoke_from == InvokeFrom.DEBUGGER,
-                user_id=self.user_id or "",
+                enabled=invoke_from == _INVOKE_FROM_DEBUGGER,
+                user_id=dify_ctx.user_id,
             )
             for method in enabled_methods
         ]
 
+    def _invoke_from_value(self) -> str:
+        invoke_from = self.require_dify_context().invoke_from
+        if isinstance(invoke_from, str):
+            return invoke_from
+        return str(getattr(invoke_from, "value", invoke_from))
+
     def _human_input_required_event(self, form_entity: HumanInputFormEntity) -> HumanInputRequired:
         node_data = self._node_data
         resolved_default_values = self.resolve_default_values()
@@ -219,10 +223,11 @@ class HumanInputNode(Node[HumanInputNodeData]):
         """
         repo = self._form_repository
         form = repo.get_form(self._workflow_execution_id, self.id)
+        dify_ctx = self.require_dify_context()
         if form is None:
             display_in_ui = self._display_in_ui()
             params = FormCreateParams(
-                app_id=self.app_id,
+                app_id=dify_ctx.app_id,
                 workflow_execution_id=self._workflow_execution_id,
                 node_id=self.id,
                 form_config=self._node_data,
@@ -232,7 +237,9 @@ class HumanInputNode(Node[HumanInputNodeData]):
                 resolved_default_values=self.resolve_default_values(),
                 console_recipient_required=self._should_require_console_recipient(),
                 console_creator_account_id=(
-                    self.user_id if self.invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE} else None
+                    dify_ctx.user_id
+                    if self._invoke_from_value() in {_INVOKE_FROM_DEBUGGER, _INVOKE_FROM_EXPLORE}
+                    else None
                 ),
                 backstage_recipient_required=True,
             )

api/dify_graph/nodes/iteration/iteration_node.py

@@ -587,24 +587,14 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
                         return
 
     def _create_graph_engine(self, index: int, item: object):
-        # Import dependencies
-        from core.app.workflow.layers.llm_quota import LLMQuotaLayer
-        from core.workflow.node_factory import DifyNodeFactory
         from dify_graph.entities import GraphInitParams
-        from dify_graph.graph import Graph
-        from dify_graph.graph_engine import GraphEngine, GraphEngineConfig
-        from dify_graph.graph_engine.command_channels import InMemoryChannel
-        from dify_graph.runtime import GraphRuntimeState
+        from dify_graph.runtime import ChildGraphNotFoundError, GraphRuntimeState
 
-        # Create GraphInitParams from node attributes
+        # Create GraphInitParams for child graph execution.
         graph_init_params = GraphInitParams(
-            tenant_id=self.tenant_id,
-            app_id=self.app_id,
             workflow_id=self.workflow_id,
             graph_config=self.graph_config,
-            user_id=self.user_id,
-            user_from=self.user_from,
-            invoke_from=self.invoke_from,
+            run_context=self.run_context,
             call_depth=self.workflow_call_depth,
         )
         # Create a deep copy of the variable pool for each iteration
@@ -621,28 +611,17 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
             total_tokens=0,
             node_run_steps=0,
         )
+        root_node_id = self.node_data.start_node_id
+        if root_node_id is None:
+            raise StartNodeIdNotFoundError(f"field start_node_id in iteration {self._node_id} not found")
 
-        # Create a new node factory with the new GraphRuntimeState
-        node_factory = DifyNodeFactory(
-            graph_init_params=graph_init_params, graph_runtime_state=graph_runtime_state_copy
-        )
-
-        # Initialize the iteration graph with the new node factory
-        iteration_graph = Graph.init(
-            graph_config=self.graph_config, node_factory=node_factory, root_node_id=self.node_data.start_node_id
-        )
-
-        if not iteration_graph:
-            raise IterationGraphNotFoundError("iteration graph not found")
-
-        # Create a new GraphEngine for this iteration
-        graph_engine = GraphEngine(
-            workflow_id=self.workflow_id,
-            graph=iteration_graph,
-            graph_runtime_state=graph_runtime_state_copy,
-            command_channel=InMemoryChannel(),  # Use InMemoryChannel for sub-graphs
-            config=GraphEngineConfig(),
-        )
-        graph_engine.layer(LLMQuotaLayer())
-
-        return graph_engine
+        try:
+            return self.graph_runtime_state.create_child_engine(
+                workflow_id=self.workflow_id,
+                graph_init_params=graph_init_params,
+                graph_runtime_state=graph_runtime_state_copy,
+                graph_config=self.graph_config,
+                root_node_id=root_node_id,
+            )
+        except ChildGraphNotFoundError as exc:
+            raise IterationGraphNotFoundError("iteration graph not found") from exc

api/dify_graph/nodes/knowledge_index/knowledge_index_node.py

@@ -3,7 +3,7 @@ from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any
 
 from dify_graph.entities.workflow_node_execution import WorkflowNodeExecutionStatus
-from dify_graph.enums import InvokeFrom, NodeExecutionType, NodeType, SystemVariableKey
+from dify_graph.enums import NodeExecutionType, NodeType, SystemVariableKey
 from dify_graph.node_events import NodeRunResult
 from dify_graph.nodes.base.node import Node
 from dify_graph.nodes.base.template import Template
@@ -20,6 +20,7 @@ if TYPE_CHECKING:
     from dify_graph.runtime import GraphRuntimeState
 
 logger = logging.getLogger(__name__)
+_INVOKE_FROM_DEBUGGER = "debugger"
 
 
 class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
@@ -58,7 +59,8 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
         if not variable:
             raise KnowledgeIndexNodeError("Index chunk variable is required.")
         invoke_from = variable_pool.get(["sys", SystemVariableKey.INVOKE_FROM])
-        is_preview = invoke_from.value == InvokeFrom.DEBUGGER if invoke_from else False
+        invoke_from_value = str(invoke_from.value) if invoke_from else None
+        is_preview = invoke_from_value == _INVOKE_FROM_DEBUGGER
 
         chunks = variable.value
         variables = {"chunks": chunks}

api/dify_graph/nodes/knowledge_retrieval/knowledge_retrieval_node.py

@@ -14,7 +14,6 @@ from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from dify_graph.node_events import NodeRunResult
 from dify_graph.nodes.base import LLMUsageTrackingMixin
 from dify_graph.nodes.base.node import Node
-from dify_graph.nodes.llm.file_saver import FileSaverImpl, LLMFileSaver
 from dify_graph.repositories.rag_retrieval_protocol import KnowledgeRetrievalRequest, RAGRetrievalProtocol, Source
 from dify_graph.variables import (
     ArrayFileSegment,
@@ -23,7 +22,11 @@ from dify_graph.variables import (
 )
 from dify_graph.variables.segments import ArrayObjectSegment
 
-from .entities import KnowledgeRetrievalNodeData
+from .entities import (
+    Condition,
+    KnowledgeRetrievalNodeData,
+    MetadataFilteringCondition,
+)
 from .exc import (
     KnowledgeRetrievalNodeError,
     RateLimitExceededError,
@@ -43,8 +46,6 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
     # Output variable for file
     _file_outputs: list["File"]
 
-    _llm_file_saver: LLMFileSaver
-
     def __init__(
         self,
         id: str,
@@ -52,8 +53,6 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
         graph_init_params: "GraphInitParams",
         graph_runtime_state: "GraphRuntimeState",
         rag_retrieval: RAGRetrievalProtocol,
-        *,
-        llm_file_saver: LLMFileSaver | None = None,
     ):
         super().__init__(
             id=id,
@@ -65,13 +64,6 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
         self._file_outputs = []
         self._rag_retrieval = rag_retrieval
 
-        if llm_file_saver is None:
-            llm_file_saver = FileSaverImpl(
-                user_id=graph_init_params.user_id,
-                tenant_id=graph_init_params.tenant_id,
-            )
-        self._llm_file_saver = llm_file_saver
-
     @classmethod
     def version(cls):
         return "1"
@@ -115,7 +107,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
 
         try:
             results, usage = self._fetch_dataset_retriever(node_data=self._node_data, variables=variables)
-            outputs = {"result": ArrayObjectSegment(value=[item.model_dump() for item in results])}
+            outputs = {"result": ArrayObjectSegment(value=[item.model_dump(by_alias=True) for item in results])}
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.SUCCEEDED,
                 inputs=variables,
@@ -160,6 +152,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
     def _fetch_dataset_retriever(
         self, node_data: KnowledgeRetrievalNodeData, variables: dict[str, Any]
     ) -> tuple[list[Source], LLMUsage]:
+        dify_ctx = self.require_dify_context()
         dataset_ids = node_data.dataset_ids
         query = variables.get("query")
         attachments = variables.get("attachments")
@@ -169,6 +162,12 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
         if node_data.metadata_filtering_mode is not None:
             metadata_filtering_mode = node_data.metadata_filtering_mode
 
+        resolved_metadata_conditions = (
+            self._resolve_metadata_filtering_conditions(node_data.metadata_filtering_conditions)
+            if node_data.metadata_filtering_conditions
+            else None
+        )
+
         if str(node_data.retrieval_mode) == DatasetRetrieveConfigEntity.RetrieveStrategy.SINGLE and query:
             # fetch model config
             if node_data.single_retrieval_config is None:
@@ -176,10 +175,10 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
             model = node_data.single_retrieval_config.model
             retrieval_resource_list = self._rag_retrieval.knowledge_retrieval(
                 request=KnowledgeRetrievalRequest(
-                    tenant_id=self.tenant_id,
-                    user_id=self.user_id,
-                    app_id=self.app_id,
-                    user_from=self.user_from.value,
+                    tenant_id=dify_ctx.tenant_id,
+                    user_id=dify_ctx.user_id,
+                    app_id=dify_ctx.app_id,
+                    user_from=dify_ctx.user_from.value,
                     dataset_ids=dataset_ids,
                     retrieval_mode=DatasetRetrieveConfigEntity.RetrieveStrategy.SINGLE.value,
                     completion_params=model.completion_params,
@@ -187,7 +186,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
                     model_mode=model.mode,
                     model_name=model.name,
                     metadata_model_config=node_data.metadata_model_config,
-                    metadata_filtering_conditions=node_data.metadata_filtering_conditions,
+                    metadata_filtering_conditions=resolved_metadata_conditions,
                     metadata_filtering_mode=metadata_filtering_mode,
                     query=query,
                 )
@@ -229,10 +228,10 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
 
             retrieval_resource_list = self._rag_retrieval.knowledge_retrieval(
                 request=KnowledgeRetrievalRequest(
-                    app_id=self.app_id,
-                    tenant_id=self.tenant_id,
-                    user_id=self.user_id,
-                    user_from=self.user_from.value,
+                    app_id=dify_ctx.app_id,
+                    tenant_id=dify_ctx.tenant_id,
+                    user_id=dify_ctx.user_id,
+                    user_from=dify_ctx.user_from.value,
                     dataset_ids=dataset_ids,
                     query=query,
                     retrieval_mode=DatasetRetrieveConfigEntity.RetrieveStrategy.MULTIPLE.value,
@@ -245,7 +244,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
                     weights=weights,
                     reranking_enable=node_data.multiple_retrieval_config.reranking_enable,
                     metadata_model_config=node_data.metadata_model_config,
-                    metadata_filtering_conditions=node_data.metadata_filtering_conditions,
+                    metadata_filtering_conditions=resolved_metadata_conditions,
                     metadata_filtering_mode=metadata_filtering_mode,
                     attachment_ids=[attachment.related_id for attachment in attachments] if attachments else None,
                 )
@@ -254,6 +253,48 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
         usage = self._rag_retrieval.llm_usage
         return retrieval_resource_list, usage
 
+    def _resolve_metadata_filtering_conditions(
+        self, conditions: MetadataFilteringCondition
+    ) -> MetadataFilteringCondition:
+        if conditions.conditions is None:
+            return MetadataFilteringCondition(
+                logical_operator=conditions.logical_operator,
+                conditions=None,
+            )
+
+        variable_pool = self.graph_runtime_state.variable_pool
+        resolved_conditions: list[Condition] = []
+        for cond in conditions.conditions or []:
+            value = cond.value
+            if isinstance(value, str):
+                segment_group = variable_pool.convert_template(value)
+                if len(segment_group.value) == 1:
+                    resolved_value = segment_group.value[0].to_object()
+                else:
+                    resolved_value = segment_group.text
+            elif isinstance(value, Sequence) and all(isinstance(v, str) for v in value):
+                resolved_values = []
+                for v in value:  # type: ignore
+                    segment_group = variable_pool.convert_template(v)
+                    if len(segment_group.value) == 1:
+                        resolved_values.append(segment_group.value[0].to_object())
+                    else:
+                        resolved_values.append(segment_group.text)
+                resolved_value = resolved_values
+            else:
+                resolved_value = value
+            resolved_conditions.append(
+                Condition(
+                    name=cond.name,
+                    comparison_operator=cond.comparison_operator,
+                    value=resolved_value,
+                )
+            )
+        return MetadataFilteringCondition(
+            logical_operator=conditions.logical_operator or "and",
+            conditions=resolved_conditions,
+        )
+
     @classmethod
     def _extract_variable_selector_to_variable_mapping(
         cls,