Merge remote-tracking branch 'origin/main' into feat/model-provider-refactor

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: crazywoola <100913391+crazywoola@users.noreply.github.com>

.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/dependabot.yml

@@ -1,25 +1,37 @@
 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
+    groups:
+      storybook:
+        patterns:
+          - "storybook"
+          - "@storybook/*"
+      npm-dependencies:
+        patterns:
+          - "*"
+        exclude-patterns:
+          - "storybook"
+          - "@storybook/*"

.github/workflows/style.yml

@@ -89,7 +89,7 @@ jobs:
         uses: actions/setup-node@v6
         if: steps.changed-files.outputs.any_changed == 'true'
         with:
-          node-version: 24
+          node-version: 22
           cache: pnpm
           cache-dependency-path: ./web/pnpm-lock.yaml
 

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

@@ -28,7 +28,7 @@ jobs:
       - name: Use Node.js
         uses: actions/setup-node@v6
         with:
-          node-version: 24
+          node-version: 22
           cache: ''
           cache-dependency-path: 'pnpm-lock.yaml'
 

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

@@ -57,7 +57,7 @@ jobs:
       - name: Set up Node.js
         uses: actions/setup-node@v6
         with:
-          node-version: 24
+          node-version: 22
           cache: pnpm
           cache-dependency-path: ./web/pnpm-lock.yaml
 

.github/workflows/web-tests.yml

@@ -39,7 +39,7 @@ jobs:
       - name: Setup Node.js
         uses: actions/setup-node@v6
         with:
-          node-version: 24
+          node-version: 22
           cache: pnpm
           cache-dependency-path: ./web/pnpm-lock.yaml
 
@@ -83,7 +83,7 @@ jobs:
       - name: Setup Node.js
         uses: actions/setup-node@v6
         with:
-          node-version: 24
+          node-version: 22
           cache: pnpm
           cache-dependency-path: ./web/pnpm-lock.yaml
 
@@ -457,7 +457,7 @@ jobs:
         uses: actions/setup-node@v6
         if: steps.changed-files.outputs.any_changed == 'true'
         with:
-          node-version: 24
+          node-version: 22
           cache: pnpm
           cache-dependency-path: ./web/pnpm-lock.yaml
 

.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:"

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
 
@@ -58,8 +49,6 @@ ignore_imports =
     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 +92,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
@@ -153,10 +138,7 @@ ignore_imports =
     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/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/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/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/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/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
@@ -219,14 +219,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:

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/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/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/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
@@ -108,6 +112,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 +144,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 +219,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,
@@ -254,6 +277,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:
@@ -344,7 +368,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/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

@@ -20,11 +20,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 +58,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 +67,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
@@ -85,7 +87,9 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
         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 +99,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,
@@ -439,13 +445,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 +460,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

@@ -66,9 +66,10 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
         self._rag_retrieval = rag_retrieval
 
         if llm_file_saver is None:
+            dify_ctx = self.require_dify_context()
             llm_file_saver = FileSaverImpl(
-                user_id=graph_init_params.user_id,
-                tenant_id=graph_init_params.tenant_id,
+                user_id=dify_ctx.user_id,
+                tenant_id=dify_ctx.tenant_id,
             )
         self._llm_file_saver = llm_file_saver
 
@@ -115,7 +116,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 +161,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")
@@ -176,10 +178,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,
@@ -229,10 +231,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,

api/dify_graph/nodes/llm/node.py

@@ -145,9 +145,10 @@ class LLMNode(Node[LLMNodeData]):
         self._memory = memory
 
         if llm_file_saver is None:
+            dify_ctx = self.require_dify_context()
             llm_file_saver = FileSaverImpl(
-                user_id=graph_init_params.user_id,
-                tenant_id=graph_init_params.tenant_id,
+                user_id=dify_ctx.user_id,
+                tenant_id=dify_ctx.tenant_id,
             )
         self._llm_file_saver = llm_file_saver
 
@@ -242,7 +243,7 @@ class LLMNode(Node[LLMNodeData]):
                 model_instance=model_instance,
                 prompt_messages=prompt_messages,
                 stop=stop,
-                user_id=self.user_id,
+                user_id=self.require_dify_context().user_id,
                 structured_output_enabled=self.node_data.structured_output_enabled,
                 structured_output=self.node_data.structured_output,
                 file_saver=self._llm_file_saver,
@@ -702,7 +703,7 @@ class LLMNode(Node[LLMNodeData]):
                                         filename=upload_file.name,
                                         extension="." + upload_file.extension,
                                         mime_type=upload_file.mime_type,
-                                        tenant_id=self.tenant_id,
+                                        tenant_id=self.require_dify_context().tenant_id,
                                         type=FileType.IMAGE,
                                         transfer_method=FileTransferMethod.LOCAL_FILE,
                                         remote_url=upload_file.source_url,

api/dify_graph/nodes/loop/loop_node.py

@@ -412,24 +412,14 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
             return build_segment_with_type(var_type, value)
 
     def _create_graph_engine(self, start_at: datetime, root_node_id: str):
-        # 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
 
-        # 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,
         )
 
@@ -439,22 +429,10 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
             start_at=start_at.timestamp(),
         )
 
-        # 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 loop graph with the new node factory
-        loop_graph = Graph.init(graph_config=self.graph_config, node_factory=node_factory, root_node_id=root_node_id)
-
-        # Create a new GraphEngine for this iteration
-        graph_engine = GraphEngine(
+        return self.graph_runtime_state.create_child_engine(
             workflow_id=self.workflow_id,
-            graph=loop_graph,
+            graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state_copy,
-            command_channel=InMemoryChannel(),  # Use InMemoryChannel for sub-graphs
-            config=GraphEngineConfig(),
+            graph_config=self.graph_config,
+            root_node_id=root_node_id,
         )
-        graph_engine.layer(LLMQuotaLayer())
-
-        return graph_engine

api/dify_graph/nodes/parameter_extractor/parameter_extractor_node.py

@@ -297,7 +297,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
             tools=tools,
             stop=list(stop),
             stream=False,
-            user=self.user_id,
+            user=self.require_dify_context().user_id,
         )
 
         # handle invoke result

api/dify_graph/nodes/question_classifier/question_classifier_node.py

@@ -86,9 +86,10 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         self._memory = memory
 
         if llm_file_saver is None:
+            dify_ctx = self.require_dify_context()
             llm_file_saver = FileSaverImpl(
-                user_id=graph_init_params.user_id,
-                tenant_id=graph_init_params.tenant_id,
+                user_id=dify_ctx.user_id,
+                tenant_id=dify_ctx.tenant_id,
             )
         self._llm_file_saver = llm_file_saver
 
@@ -160,7 +161,7 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
                 model_instance=model_instance,
                 prompt_messages=prompt_messages,
                 stop=stop,
-                user_id=self.user_id,
+                user_id=self.require_dify_context().user_id,
                 structured_output_enabled=False,
                 structured_output=None,
                 file_saver=self._llm_file_saver,

api/dify_graph/nodes/tool/tool_node.py

@@ -56,6 +56,8 @@ class ToolNode(Node[ToolNodeData]):
         """
         from core.plugin.impl.exc import PluginDaemonClientSideError, PluginInvokeError
 
+        dify_ctx = self.require_dify_context()
+
         # fetch tool icon
         tool_info = {
             "provider_type": self.node_data.provider_type.value,
@@ -75,7 +77,12 @@ class ToolNode(Node[ToolNodeData]):
             if self.node_data.version != "1" or self.node_data.tool_node_version is not None:
                 variable_pool = self.graph_runtime_state.variable_pool
             tool_runtime = ToolManager.get_workflow_tool_runtime(
-                self.tenant_id, self.app_id, self._node_id, self.node_data, self.invoke_from, variable_pool
+                dify_ctx.tenant_id,
+                dify_ctx.app_id,
+                self._node_id,
+                self.node_data,
+                dify_ctx.invoke_from,
+                variable_pool,
             )
         except ToolNodeError as e:
             yield StreamCompletedEvent(
@@ -109,10 +116,10 @@ class ToolNode(Node[ToolNodeData]):
             message_stream = ToolEngine.generic_invoke(
                 tool=tool_runtime,
                 tool_parameters=parameters,
-                user_id=self.user_id,
+                user_id=dify_ctx.user_id,
                 workflow_tool_callback=DifyWorkflowCallbackHandler(),
                 workflow_call_depth=self.workflow_call_depth,
-                app_id=self.app_id,
+                app_id=dify_ctx.app_id,
                 conversation_id=conversation_id.text if conversation_id else None,
             )
         except ToolNodeError as e:
@@ -133,8 +140,8 @@ class ToolNode(Node[ToolNodeData]):
                 messages=message_stream,
                 tool_info=tool_info,
                 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_id=self._node_id,
                 tool_runtime=tool_runtime,
             )

api/dify_graph/nodes/trigger_webhook/node.py

@@ -69,6 +69,7 @@ class TriggerWebhookNode(Node[WebhookData]):
         )
 
     def generate_file_var(self, param_name: str, file: dict):
+        dify_ctx = self.require_dify_context()
         related_id = file.get("related_id")
         transfer_method_value = file.get("transfer_method")
         if transfer_method_value:
@@ -84,7 +85,7 @@ class TriggerWebhookNode(Node[WebhookData]):
             try:
                 file_obj = file_factory.build_from_mapping(
                     mapping=file,
-                    tenant_id=self.tenant_id,
+                    tenant_id=dify_ctx.tenant_id,
                 )
                 file_segment = build_segment_with_type(SegmentType.FILE, file_obj)
                 return FileVariable(name=param_name, value=file_segment.value, selector=[self.id, param_name])

api/dify_graph/runtime/__init__.py

@@ -1,9 +1,17 @@
-from .graph_runtime_state import GraphRuntimeState
+from .graph_runtime_state import (
+    ChildEngineBuilderNotConfiguredError,
+    ChildEngineError,
+    ChildGraphNotFoundError,
+    GraphRuntimeState,
+)
 from .graph_runtime_state_protocol import ReadOnlyGraphRuntimeState, ReadOnlyVariablePool
 from .read_only_wrappers import ReadOnlyGraphRuntimeStateWrapper, ReadOnlyVariablePoolWrapper
 from .variable_pool import VariablePool, VariableValue
 
 __all__ = [
+    "ChildEngineBuilderNotConfiguredError",
+    "ChildEngineError",
+    "ChildGraphNotFoundError",
     "GraphRuntimeState",
     "ReadOnlyGraphRuntimeState",
     "ReadOnlyGraphRuntimeStateWrapper",

api/dify_graph/runtime/graph_runtime_state.py

@@ -15,6 +15,7 @@ from dify_graph.model_runtime.entities.llm_entities import LLMUsage
 from dify_graph.runtime.variable_pool import VariablePool
 
 if TYPE_CHECKING:
+    from dify_graph.entities import GraphInitParams
     from dify_graph.entities.pause_reason import PauseReason
 
 
@@ -135,6 +136,31 @@ class GraphProtocol(Protocol):
     def get_outgoing_edges(self, node_id: str) -> Sequence[EdgeProtocol]: ...
 
 
+class ChildGraphEngineBuilderProtocol(Protocol):
+    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] = (),
+    ) -> Any: ...
+
+
+class ChildEngineError(ValueError):
+    """Base error type for child-engine creation failures."""
+
+
+class ChildEngineBuilderNotConfiguredError(ChildEngineError):
+    """Raised when child-engine creation is requested without a bound builder."""
+
+
+class ChildGraphNotFoundError(ChildEngineError):
+    """Raised when the requested child graph entry point cannot be resolved."""
+
+
 class _GraphStateSnapshot(BaseModel):
     """Serializable graph state snapshot for node/edge states."""
 
@@ -209,6 +235,7 @@ class GraphRuntimeState:
         self._pending_graph_execution_workflow_id: str | None = None
         self._paused_nodes: set[str] = set()
         self._deferred_nodes: set[str] = set()
+        self._child_engine_builder: ChildGraphEngineBuilderProtocol | None = None
 
         # Node and edges states needed to be restored into
         # graph object.
@@ -250,6 +277,31 @@ class GraphRuntimeState:
         if self._graph is not None:
             _ = self.response_coordinator
 
+    def bind_child_engine_builder(self, builder: ChildGraphEngineBuilderProtocol) -> None:
+        self._child_engine_builder = builder
+
+    def create_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] = (),
+    ) -> Any:
+        if self._child_engine_builder is None:
+            raise ChildEngineBuilderNotConfiguredError("Child engine builder is not configured.")
+
+        return self._child_engine_builder.build_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,
+        )
+
     # ------------------------------------------------------------------
     # Primary collaborators
     # ------------------------------------------------------------------

api/dify_graph/runtime/variable_pool.py

@@ -65,9 +65,15 @@ class VariablePool(BaseModel):
         # Add environment variables to the variable pool
         for var in self.environment_variables:
             self.add((ENVIRONMENT_VARIABLE_NODE_ID, var.name), var)
-        # Add conversation variables to the variable pool
+        # Add conversation variables to the variable pool. When restoring from a serialized
+        # snapshot, `variable_dictionary` already carries the latest runtime values.
+        # In that case, keep existing entries instead of overwriting them with the
+        # bootstrap list.
         for var in self.conversation_variables:
-            self.add((CONVERSATION_VARIABLE_NODE_ID, var.name), var)
+            selector = (CONVERSATION_VARIABLE_NODE_ID, var.name)
+            if self._has(selector):
+                continue
+            self.add(selector, var)
         # Add rag pipeline variables to the variable pool
         if self.rag_pipeline_variables:
             rag_pipeline_variables_map: defaultdict[Any, dict[Any, Any]] = defaultdict(dict)

api/docker/entrypoint.sh

@@ -35,10 +35,10 @@ if [[ "${MODE}" == "worker" ]]; then
   if [[ -z "${CELERY_QUEUES}" ]]; then
     if [[ "${EDITION}" == "CLOUD" ]]; then
       # Cloud edition: separate queues for dataset and trigger tasks
-      DEFAULT_QUEUES="api_token,dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow_professional,workflow_team,workflow_sandbox,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention,workflow_based_app_execution"
+      DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow_professional,workflow_team,workflow_sandbox,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention,workflow_based_app_execution"
     else
       # Community edition (SELF_HOSTED): dataset, pipeline and workflow have separate queues
-      DEFAULT_QUEUES="api_token,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"
+      DEFAULT_QUEUES="api_token,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"
     fi
   else
     DEFAULT_QUEUES="${CELERY_QUEUES}"

api/events/event_handlers/update_app_dataset_join_when_app_model_config_updated.py

@@ -1,3 +1,5 @@
+from typing import Any, cast
+
 from sqlalchemy import select
 
 from events.app_event import app_model_config_was_updated
@@ -54,9 +56,11 @@ def get_dataset_ids_from_model_config(app_model_config: AppModelConfig) -> set[s
             continue
 
         tool_type = list(tool.keys())[0]
-        tool_config = list(tool.values())[0]
+        tool_config = cast(dict[str, Any], list(tool.values())[0])
         if tool_type == "dataset":
-            dataset_ids.add(tool_config.get("id"))
+            dataset_id = tool_config.get("id")
+            if isinstance(dataset_id, str):
+                dataset_ids.add(dataset_id)
 
     # get dataset from dataset_configs
     dataset_configs = app_model_config.dataset_configs_dict

api/extensions/ext_redis.py

@@ -18,6 +18,7 @@ from dify_app import DifyApp
 from libs.broadcast_channel.channel import BroadcastChannel as BroadcastChannelProtocol
 from libs.broadcast_channel.redis.channel import BroadcastChannel as RedisBroadcastChannel
 from libs.broadcast_channel.redis.sharded_channel import ShardedRedisBroadcastChannel
+from libs.broadcast_channel.redis.streams_channel import StreamsBroadcastChannel
 
 if TYPE_CHECKING:
     from redis.lock import Lock
@@ -181,13 +182,18 @@ def _create_sentinel_client(redis_params: dict[str, Any]) -> Union[redis.Redis,
 
     sentinel_hosts = [(node.split(":")[0], int(node.split(":")[1])) for node in dify_config.REDIS_SENTINELS.split(",")]
 
+    sentinel_kwargs = {
+        "socket_timeout": dify_config.REDIS_SENTINEL_SOCKET_TIMEOUT,
+        "username": dify_config.REDIS_SENTINEL_USERNAME,
+        "password": dify_config.REDIS_SENTINEL_PASSWORD,
+    }
+
+    if dify_config.REDIS_MAX_CONNECTIONS:
+        sentinel_kwargs["max_connections"] = dify_config.REDIS_MAX_CONNECTIONS
+
     sentinel = Sentinel(
         sentinel_hosts,
-        sentinel_kwargs={
-            "socket_timeout": dify_config.REDIS_SENTINEL_SOCKET_TIMEOUT,
-            "username": dify_config.REDIS_SENTINEL_USERNAME,
-            "password": dify_config.REDIS_SENTINEL_PASSWORD,
-        },
+        sentinel_kwargs=sentinel_kwargs,
     )
 
     master: redis.Redis = sentinel.master_for(dify_config.REDIS_SENTINEL_SERVICE_NAME, **redis_params)
@@ -204,12 +210,15 @@ def _create_cluster_client() -> Union[redis.Redis, RedisCluster]:
         for node in dify_config.REDIS_CLUSTERS.split(",")
     ]
 
-    cluster: RedisCluster = RedisCluster(
-        startup_nodes=nodes,
-        password=dify_config.REDIS_CLUSTERS_PASSWORD,
-        protocol=dify_config.REDIS_SERIALIZATION_PROTOCOL,
-        cache_config=_get_cache_configuration(),
-    )
+    cluster_kwargs: dict[str, Any] = {
+        "startup_nodes": nodes,
+        "password": dify_config.REDIS_CLUSTERS_PASSWORD,
+        "protocol": dify_config.REDIS_SERIALIZATION_PROTOCOL,
+        "cache_config": _get_cache_configuration(),
+    }
+    if dify_config.REDIS_MAX_CONNECTIONS:
+        cluster_kwargs["max_connections"] = dify_config.REDIS_MAX_CONNECTIONS
+    cluster: RedisCluster = RedisCluster(**cluster_kwargs)
     return cluster
 
 
@@ -225,6 +234,9 @@ def _create_standalone_client(redis_params: dict[str, Any]) -> Union[redis.Redis
         }
     )
 
+    if dify_config.REDIS_MAX_CONNECTIONS:
+        redis_params["max_connections"] = dify_config.REDIS_MAX_CONNECTIONS
+
     if ssl_kwargs:
         redis_params.update(ssl_kwargs)
 
@@ -234,9 +246,17 @@ def _create_standalone_client(redis_params: dict[str, Any]) -> Union[redis.Redis
 
 
 def _create_pubsub_client(pubsub_url: str, use_clusters: bool) -> redis.Redis | RedisCluster:
+    max_conns = dify_config.REDIS_MAX_CONNECTIONS
     if use_clusters:
-        return RedisCluster.from_url(pubsub_url)
-    return redis.Redis.from_url(pubsub_url)
+        if max_conns:
+            return RedisCluster.from_url(pubsub_url, max_connections=max_conns)
+        else:
+            return RedisCluster.from_url(pubsub_url)
+
+    if max_conns:
+        return redis.Redis.from_url(pubsub_url, max_connections=max_conns)
+    else:
+        return redis.Redis.from_url(pubsub_url)
 
 
 def init_app(app: DifyApp):
@@ -269,6 +289,11 @@ def get_pubsub_broadcast_channel() -> BroadcastChannelProtocol:
     assert _pubsub_redis_client is not None, "PubSub redis Client should be initialized here."
     if dify_config.PUBSUB_REDIS_CHANNEL_TYPE == "sharded":
         return ShardedRedisBroadcastChannel(_pubsub_redis_client)
+    if dify_config.PUBSUB_REDIS_CHANNEL_TYPE == "streams":
+        return StreamsBroadcastChannel(
+            _pubsub_redis_client,
+            retention_seconds=dify_config.PUBSUB_STREAMS_RETENTION_SECONDS,
+        )
     return RedisBroadcastChannel(_pubsub_redis_client)
 
 

api/libs/broadcast_channel/redis/streams_channel.py

@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+import logging
+import queue
+import threading
+from collections.abc import Iterator
+from typing import Self
+
+from libs.broadcast_channel.channel import Producer, Subscriber, Subscription
+from libs.broadcast_channel.exc import SubscriptionClosedError
+from redis import Redis, RedisCluster
+
+logger = logging.getLogger(__name__)
+
+
+class StreamsBroadcastChannel:
+    """
+    Redis Streams based broadcast channel implementation.
+
+    Characteristics:
+    - At-least-once delivery for late subscribers within the stream retention window.
+    - Each topic is stored as a dedicated Redis Stream key.
+    - The stream key expires `retention_seconds` after the last event is published (to bound storage).
+    """
+
+    def __init__(self, redis_client: Redis | RedisCluster, *, retention_seconds: int = 600):
+        self._client = redis_client
+        self._retention_seconds = max(int(retention_seconds or 0), 0)
+
+    def topic(self, topic: str) -> StreamsTopic:
+        return StreamsTopic(self._client, topic, retention_seconds=self._retention_seconds)
+
+
+class StreamsTopic:
+    def __init__(self, redis_client: Redis | RedisCluster, topic: str, *, retention_seconds: int = 600):
+        self._client = redis_client
+        self._topic = topic
+        self._key = f"stream:{topic}"
+        self._retention_seconds = retention_seconds
+        self.max_length = 5000
+
+    def as_producer(self) -> Producer:
+        return self
+
+    def publish(self, payload: bytes) -> None:
+        self._client.xadd(self._key, {b"data": payload}, maxlen=self.max_length)
+        if self._retention_seconds > 0:
+            try:
+                self._client.expire(self._key, self._retention_seconds)
+            except Exception as e:
+                logger.warning("Failed to set expire for stream key %s: %s", self._key, e, exc_info=True)
+
+    def as_subscriber(self) -> Subscriber:
+        return self
+
+    def subscribe(self) -> Subscription:
+        return _StreamsSubscription(self._client, self._key)
+
+
+class _StreamsSubscription(Subscription):
+    _SENTINEL = object()
+
+    def __init__(self, client: Redis | RedisCluster, key: str):
+        self._client = client
+        self._key = key
+        self._closed = threading.Event()
+        self._last_id = "0-0"
+        self._queue: queue.Queue[object] = queue.Queue()
+        self._start_lock = threading.Lock()
+        self._listener: threading.Thread | None = None
+
+    def _listen(self) -> None:
+        try:
+            while not self._closed.is_set():
+                streams = self._client.xread({self._key: self._last_id}, block=1000, count=100)
+
+                if not streams:
+                    continue
+
+                for _key, entries in streams:
+                    for entry_id, fields in entries:
+                        data = None
+                        if isinstance(fields, dict):
+                            data = fields.get(b"data")
+                        data_bytes: bytes | None = None
+                        if isinstance(data, str):
+                            data_bytes = data.encode()
+                        elif isinstance(data, (bytes, bytearray)):
+                            data_bytes = bytes(data)
+                        if data_bytes is not None:
+                            self._queue.put_nowait(data_bytes)
+                        self._last_id = entry_id
+        finally:
+            self._queue.put_nowait(self._SENTINEL)
+            self._listener = None
+
+    def _start_if_needed(self) -> None:
+        if self._listener is not None:
+            return
+        # Ensure only one listener thread is created under concurrent calls
+        with self._start_lock:
+            if self._listener is not None or self._closed.is_set():
+                return
+            self._listener = threading.Thread(
+                target=self._listen,
+                name=f"redis-streams-sub-{self._key}",
+                daemon=True,
+            )
+            self._listener.start()
+
+    def __iter__(self) -> Iterator[bytes]:
+        # Iterator delegates to receive with timeout; stops on closure.
+        self._start_if_needed()
+        while not self._closed.is_set():
+            item = self.receive(timeout=1)
+            if item is not None:
+                yield item
+
+    def receive(self, timeout: float | None = 0.1) -> bytes | None:
+        if self._closed.is_set():
+            raise SubscriptionClosedError("The Redis streams subscription is closed")
+        self._start_if_needed()
+
+        try:
+            if timeout is None:
+                item = self._queue.get()
+            else:
+                item = self._queue.get(timeout=timeout)
+        except queue.Empty:
+            return None
+
+        if item is self._SENTINEL or self._closed.is_set():
+            raise SubscriptionClosedError("The Redis streams subscription is closed")
+        assert isinstance(item, (bytes, bytearray)), "Unexpected item type in stream queue"
+        return bytes(item)
+
+    def close(self) -> None:
+        if self._closed.is_set():
+            return
+        self._closed.set()
+        listener = self._listener
+        if listener is not None:
+            listener.join(timeout=2.0)
+            if listener.is_alive():
+                logger.warning(
+                    "Streams subscription listener for key %s did not stop within timeout; keeping reference.",
+                    self._key,
+                )
+            else:
+                self._listener = None
+
+    # Context manager helpers
+    def __enter__(self) -> Self:
+        self._start_if_needed()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback) -> bool | None:
+        self.close()
+        return None

api/migrations/versions/2026_02_26_1336-e288952f2994_add_partial_indexes_on_conversations_.py

@@ -0,0 +1,37 @@
+"""add partial indexes on conversations for app_id with created_at and updated_at
+
+Revision ID: e288952f2994
+Revises: fce013ca180e
+Create Date: 2026-02-26 13:36:45.928922
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = 'e288952f2994'
+down_revision = 'fce013ca180e'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with op.batch_alter_table('conversations', schema=None) as batch_op:
+        batch_op.create_index(
+            'conversation_app_created_at_idx',
+            ['app_id', sa.literal_column('created_at DESC')],
+            unique=False,
+            postgresql_where=sa.text('is_deleted IS false'),
+        )
+        batch_op.create_index(
+            'conversation_app_updated_at_idx',
+            ['app_id', sa.literal_column('updated_at DESC')],
+            unique=False,
+            postgresql_where=sa.text('is_deleted IS false'),
+        )
+
+
+def downgrade():
+    with op.batch_alter_table('conversations', schema=None) as batch_op:
+        batch_op.drop_index('conversation_app_updated_at_idx')
+        batch_op.drop_index('conversation_app_created_at_idx')

api/models/model.py

@@ -7,7 +7,7 @@ from collections.abc import Mapping, Sequence
 from datetime import datetime
 from decimal import Decimal
 from enum import StrEnum, auto
-from typing import TYPE_CHECKING, Any, Literal, cast
+from typing import TYPE_CHECKING, Any, Literal, NotRequired, cast
 from uuid import uuid4
 
 import sqlalchemy as sa
@@ -15,6 +15,7 @@ from flask import request
 from flask_login import UserMixin  # type: ignore[import-untyped]
 from sqlalchemy import BigInteger, Float, Index, PrimaryKeyConstraint, String, exists, func, select, text
 from sqlalchemy.orm import Mapped, Session, mapped_column
+from typing_extensions import TypedDict
 
 from configs import dify_config
 from constants import DEFAULT_FILE_NUMBER_LIMITS
@@ -36,6 +37,259 @@ if TYPE_CHECKING:
     from .workflow import Workflow
 
 
+# --- TypedDict definitions for structured dict return types ---
+
+
+class EnabledConfig(TypedDict):
+    enabled: bool
+
+
+class EmbeddingModelInfo(TypedDict):
+    embedding_provider_name: str
+    embedding_model_name: str
+
+
+class AnnotationReplyDisabledConfig(TypedDict):
+    enabled: Literal[False]
+
+
+class AnnotationReplyEnabledConfig(TypedDict):
+    id: str
+    enabled: Literal[True]
+    score_threshold: float
+    embedding_model: EmbeddingModelInfo
+
+
+AnnotationReplyConfig = AnnotationReplyEnabledConfig | AnnotationReplyDisabledConfig
+
+
+class SensitiveWordAvoidanceConfig(TypedDict):
+    enabled: bool
+    type: str
+    config: dict[str, Any]
+
+
+class AgentToolConfig(TypedDict):
+    provider_type: str
+    provider_id: str
+    tool_name: str
+    tool_parameters: dict[str, Any]
+    plugin_unique_identifier: NotRequired[str | None]
+    credential_id: NotRequired[str | None]
+
+
+class AgentModeConfig(TypedDict):
+    enabled: bool
+    strategy: str | None
+    tools: list[AgentToolConfig | dict[str, Any]]
+    prompt: str | None
+
+
+class ImageUploadConfig(TypedDict):
+    enabled: bool
+    number_limits: int
+    detail: str
+    transfer_methods: list[str]
+
+
+class FileUploadConfig(TypedDict):
+    image: ImageUploadConfig
+
+
+class DeletedToolInfo(TypedDict):
+    type: str
+    tool_name: str
+    provider_id: str
+
+
+class ExternalDataToolConfig(TypedDict):
+    enabled: bool
+    variable: str
+    type: str
+    config: dict[str, Any]
+
+
+class UserInputFormItemConfig(TypedDict):
+    variable: str
+    label: str
+    description: NotRequired[str]
+    required: NotRequired[bool]
+    max_length: NotRequired[int]
+    options: NotRequired[list[str]]
+    default: NotRequired[str]
+    type: NotRequired[str]
+    config: NotRequired[dict[str, Any]]
+
+
+# Each item is a single-key dict, e.g. {"text-input": UserInputFormItemConfig}
+UserInputFormItem = dict[str, UserInputFormItemConfig]
+
+
+class DatasetConfigs(TypedDict):
+    retrieval_model: str
+    datasets: NotRequired[dict[str, Any]]
+    top_k: NotRequired[int]
+    score_threshold: NotRequired[float]
+    score_threshold_enabled: NotRequired[bool]
+    reranking_model: NotRequired[dict[str, Any] | None]
+    weights: NotRequired[dict[str, Any] | None]
+    reranking_enabled: NotRequired[bool]
+    reranking_mode: NotRequired[str]
+    metadata_filtering_mode: NotRequired[str]
+    metadata_model_config: NotRequired[dict[str, Any] | None]
+    metadata_filtering_conditions: NotRequired[dict[str, Any] | None]
+
+
+class ChatPromptMessage(TypedDict):
+    text: str
+    role: str
+
+
+class ChatPromptConfig(TypedDict, total=False):
+    prompt: list[ChatPromptMessage]
+
+
+class CompletionPromptText(TypedDict):
+    text: str
+
+
+class ConversationHistoriesRole(TypedDict):
+    user_prefix: str
+    assistant_prefix: str
+
+
+class CompletionPromptConfig(TypedDict):
+    prompt: CompletionPromptText
+    conversation_histories_role: NotRequired[ConversationHistoriesRole]
+
+
+class ModelConfig(TypedDict):
+    provider: str
+    name: str
+    mode: str
+    completion_params: NotRequired[dict[str, Any]]
+
+
+class AppModelConfigDict(TypedDict):
+    opening_statement: str | None
+    suggested_questions: list[str]
+    suggested_questions_after_answer: EnabledConfig
+    speech_to_text: EnabledConfig
+    text_to_speech: EnabledConfig
+    retriever_resource: EnabledConfig
+    annotation_reply: AnnotationReplyConfig
+    more_like_this: EnabledConfig
+    sensitive_word_avoidance: SensitiveWordAvoidanceConfig
+    external_data_tools: list[ExternalDataToolConfig]
+    model: ModelConfig
+    user_input_form: list[UserInputFormItem]
+    dataset_query_variable: str | None
+    pre_prompt: str | None
+    agent_mode: AgentModeConfig
+    prompt_type: str
+    chat_prompt_config: ChatPromptConfig
+    completion_prompt_config: CompletionPromptConfig
+    dataset_configs: DatasetConfigs
+    file_upload: FileUploadConfig
+    # Added dynamically in Conversation.model_config
+    model_id: NotRequired[str | None]
+    provider: NotRequired[str | None]
+
+
+class ConversationDict(TypedDict):
+    id: str
+    app_id: str
+    app_model_config_id: str | None
+    model_provider: str | None
+    override_model_configs: str | None
+    model_id: str | None
+    mode: str
+    name: str
+    summary: str | None
+    inputs: dict[str, Any]
+    introduction: str | None
+    system_instruction: str | None
+    system_instruction_tokens: int
+    status: str
+    invoke_from: str | None
+    from_source: str
+    from_end_user_id: str | None
+    from_account_id: str | None
+    read_at: datetime | None
+    read_account_id: str | None
+    dialogue_count: int
+    created_at: datetime
+    updated_at: datetime
+
+
+class MessageDict(TypedDict):
+    id: str
+    app_id: str
+    conversation_id: str
+    model_id: str | None
+    inputs: dict[str, Any]
+    query: str
+    total_price: Decimal | None
+    message: dict[str, Any]
+    answer: str
+    status: str
+    error: str | None
+    message_metadata: dict[str, Any]
+    from_source: str
+    from_end_user_id: str | None
+    from_account_id: str | None
+    created_at: str
+    updated_at: str
+    agent_based: bool
+    workflow_run_id: str | None
+
+
+class MessageFeedbackDict(TypedDict):
+    id: str
+    app_id: str
+    conversation_id: str
+    message_id: str
+    rating: str
+    content: str | None
+    from_source: str
+    from_end_user_id: str | None
+    from_account_id: str | None
+    created_at: str
+    updated_at: str
+
+
+class MessageFileInfo(TypedDict, total=False):
+    belongs_to: str | None
+    upload_file_id: str | None
+    id: str
+    tenant_id: str
+    type: str
+    transfer_method: str
+    remote_url: str | None
+    related_id: str | None
+    filename: str | None
+    extension: str | None
+    mime_type: str | None
+    size: int
+    dify_model_identity: str
+    url: str | None
+
+
+class ExtraContentDict(TypedDict, total=False):
+    type: str
+    workflow_run_id: str
+
+
+class TraceAppConfigDict(TypedDict):
+    id: str
+    app_id: str
+    tracing_provider: str | None
+    tracing_config: dict[str, Any]
+    is_active: bool
+    created_at: str | None
+    updated_at: str | None
+
+
 class DifySetup(TypeBase):
     __tablename__ = "dify_setups"
     __table_args__ = (sa.PrimaryKeyConstraint("version", name="dify_setup_pkey"),)
@@ -176,7 +430,7 @@ class App(Base):
         return str(self.mode)
 
     @property
-    def deleted_tools(self) -> list[dict[str, str]]:
+    def deleted_tools(self) -> list[DeletedToolInfo]:
         from core.tools.tool_manager import ToolManager, ToolProviderType
         from services.plugin.plugin_service import PluginService
 
@@ -257,7 +511,7 @@ class App(Base):
             provider_id.provider_name: existence[i] for i, provider_id in enumerate(builtin_provider_ids)
         }
 
-        deleted_tools: list[dict[str, str]] = []
+        deleted_tools: list[DeletedToolInfo] = []
 
         for tool in tools:
             keys = list(tool.keys())
@@ -364,35 +618,38 @@ class AppModelConfig(TypeBase):
         return app
 
     @property
-    def model_dict(self) -> dict[str, Any]:
-        return json.loads(self.model) if self.model else {}
+    def model_dict(self) -> ModelConfig:
+        return cast(ModelConfig, json.loads(self.model) if self.model else {})
 
     @property
     def suggested_questions_list(self) -> list[str]:
         return json.loads(self.suggested_questions) if self.suggested_questions else []
 
     @property
-    def suggested_questions_after_answer_dict(self) -> dict[str, Any]:
-        return (
+    def suggested_questions_after_answer_dict(self) -> EnabledConfig:
+        return cast(
+            EnabledConfig,
             json.loads(self.suggested_questions_after_answer)
             if self.suggested_questions_after_answer
-            else {"enabled": False}
+            else {"enabled": False},
         )
 
     @property
-    def speech_to_text_dict(self) -> dict[str, Any]:
-        return json.loads(self.speech_to_text) if self.speech_to_text else {"enabled": False}
+    def speech_to_text_dict(self) -> EnabledConfig:
+        return cast(EnabledConfig, json.loads(self.speech_to_text) if self.speech_to_text else {"enabled": False})
 
     @property
-    def text_to_speech_dict(self) -> dict[str, Any]:
-        return json.loads(self.text_to_speech) if self.text_to_speech else {"enabled": False}
+    def text_to_speech_dict(self) -> EnabledConfig:
+        return cast(EnabledConfig, json.loads(self.text_to_speech) if self.text_to_speech else {"enabled": False})
 
     @property
-    def retriever_resource_dict(self) -> dict[str, Any]:
-        return json.loads(self.retriever_resource) if self.retriever_resource else {"enabled": True}
+    def retriever_resource_dict(self) -> EnabledConfig:
+        return cast(
+            EnabledConfig, json.loads(self.retriever_resource) if self.retriever_resource else {"enabled": True}
+        )
 
     @property
-    def annotation_reply_dict(self) -> dict[str, Any]:
+    def annotation_reply_dict(self) -> AnnotationReplyConfig:
         annotation_setting = (
             db.session.query(AppAnnotationSetting).where(AppAnnotationSetting.app_id == self.app_id).first()
         )
@@ -415,56 +672,62 @@ class AppModelConfig(TypeBase):
             return {"enabled": False}
 
     @property
-    def more_like_this_dict(self) -> dict[str, Any]:
-        return json.loads(self.more_like_this) if self.more_like_this else {"enabled": False}
+    def more_like_this_dict(self) -> EnabledConfig:
+        return cast(EnabledConfig, json.loads(self.more_like_this) if self.more_like_this else {"enabled": False})
 
     @property
-    def sensitive_word_avoidance_dict(self) -> dict[str, Any]:
-        return (
+    def sensitive_word_avoidance_dict(self) -> SensitiveWordAvoidanceConfig:
+        return cast(
+            SensitiveWordAvoidanceConfig,
             json.loads(self.sensitive_word_avoidance)
             if self.sensitive_word_avoidance
-            else {"enabled": False, "type": "", "configs": []}
+            else {"enabled": False, "type": "", "config": {}},
         )
 
     @property
-    def external_data_tools_list(self) -> list[dict[str, Any]]:
+    def external_data_tools_list(self) -> list[ExternalDataToolConfig]:
         return json.loads(self.external_data_tools) if self.external_data_tools else []
 
     @property
-    def user_input_form_list(self) -> list[dict[str, Any]]:
+    def user_input_form_list(self) -> list[UserInputFormItem]:
         return json.loads(self.user_input_form) if self.user_input_form else []
 
     @property
-    def agent_mode_dict(self) -> dict[str, Any]:
-        return (
+    def agent_mode_dict(self) -> AgentModeConfig:
+        return cast(
+            AgentModeConfig,
             json.loads(self.agent_mode)
             if self.agent_mode
-            else {"enabled": False, "strategy": None, "tools": [], "prompt": None}
+            else {"enabled": False, "strategy": None, "tools": [], "prompt": None},
         )
 
     @property
-    def chat_prompt_config_dict(self) -> dict[str, Any]:
-        return json.loads(self.chat_prompt_config) if self.chat_prompt_config else {}
+    def chat_prompt_config_dict(self) -> ChatPromptConfig:
+        return cast(ChatPromptConfig, json.loads(self.chat_prompt_config) if self.chat_prompt_config else {})
 
     @property
-    def completion_prompt_config_dict(self) -> dict[str, Any]:
-        return json.loads(self.completion_prompt_config) if self.completion_prompt_config else {}
+    def completion_prompt_config_dict(self) -> CompletionPromptConfig:
+        return cast(
+            CompletionPromptConfig,
+            json.loads(self.completion_prompt_config) if self.completion_prompt_config else {},
+        )
 
     @property
-    def dataset_configs_dict(self) -> dict[str, Any]:
+    def dataset_configs_dict(self) -> DatasetConfigs:
         if self.dataset_configs:
-            dataset_configs: dict[str, Any] = json.loads(self.dataset_configs)
+            dataset_configs = json.loads(self.dataset_configs)
             if "retrieval_model" not in dataset_configs:
                 return {"retrieval_model": "single"}
             else:
-                return dataset_configs
+                return cast(DatasetConfigs, dataset_configs)
         return {
             "retrieval_model": "multiple",
         }
 
     @property
-    def file_upload_dict(self) -> dict[str, Any]:
-        return (
+    def file_upload_dict(self) -> FileUploadConfig:
+        return cast(
+            FileUploadConfig,
             json.loads(self.file_upload)
             if self.file_upload
             else {
@@ -474,10 +737,10 @@ class AppModelConfig(TypeBase):
                     "detail": "high",
                     "transfer_methods": ["remote_url", "local_file"],
                 }
-            }
+            },
         )
 
-    def to_dict(self) -> dict[str, Any]:
+    def to_dict(self) -> AppModelConfigDict:
         return {
             "opening_statement": self.opening_statement,
             "suggested_questions": self.suggested_questions_list,
@@ -501,36 +764,42 @@ class AppModelConfig(TypeBase):
             "file_upload": self.file_upload_dict,
         }
 
-    def from_model_config_dict(self, model_config: Mapping[str, Any]):
+    def from_model_config_dict(self, model_config: AppModelConfigDict):
         self.opening_statement = model_config.get("opening_statement")
         self.suggested_questions = (
-            json.dumps(model_config["suggested_questions"]) if model_config.get("suggested_questions") else None
+            json.dumps(model_config.get("suggested_questions")) if model_config.get("suggested_questions") else None
         )
         self.suggested_questions_after_answer = (
-            json.dumps(model_config["suggested_questions_after_answer"])
+            json.dumps(model_config.get("suggested_questions_after_answer"))
             if model_config.get("suggested_questions_after_answer")
             else None
         )
-        self.speech_to_text = json.dumps(model_config["speech_to_text"]) if model_config.get("speech_to_text") else None
-        self.text_to_speech = json.dumps(model_config["text_to_speech"]) if model_config.get("text_to_speech") else None
-        self.more_like_this = json.dumps(model_config["more_like_this"]) if model_config.get("more_like_this") else None
+        self.speech_to_text = (
+            json.dumps(model_config.get("speech_to_text")) if model_config.get("speech_to_text") else None
+        )
+        self.text_to_speech = (
+            json.dumps(model_config.get("text_to_speech")) if model_config.get("text_to_speech") else None
+        )
+        self.more_like_this = (
+            json.dumps(model_config.get("more_like_this")) if model_config.get("more_like_this") else None
+        )
         self.sensitive_word_avoidance = (
-            json.dumps(model_config["sensitive_word_avoidance"])
+            json.dumps(model_config.get("sensitive_word_avoidance"))
             if model_config.get("sensitive_word_avoidance")
             else None
         )
         self.external_data_tools = (
-            json.dumps(model_config["external_data_tools"]) if model_config.get("external_data_tools") else None
+            json.dumps(model_config.get("external_data_tools")) if model_config.get("external_data_tools") else None
         )
-        self.model = json.dumps(model_config["model"]) if model_config.get("model") else None
+        self.model = json.dumps(model_config.get("model")) if model_config.get("model") else None
         self.user_input_form = (
-            json.dumps(model_config["user_input_form"]) if model_config.get("user_input_form") else None
+            json.dumps(model_config.get("user_input_form")) if model_config.get("user_input_form") else None
         )
         self.dataset_query_variable = model_config.get("dataset_query_variable")
-        self.pre_prompt = model_config["pre_prompt"]
-        self.agent_mode = json.dumps(model_config["agent_mode"]) if model_config.get("agent_mode") else None
+        self.pre_prompt = model_config.get("pre_prompt")
+        self.agent_mode = json.dumps(model_config.get("agent_mode")) if model_config.get("agent_mode") else None
         self.retriever_resource = (
-            json.dumps(model_config["retriever_resource"]) if model_config.get("retriever_resource") else None
+            json.dumps(model_config.get("retriever_resource")) if model_config.get("retriever_resource") else None
         )
         self.prompt_type = model_config.get("prompt_type", "simple")
         self.chat_prompt_config = (
@@ -711,6 +980,18 @@ class Conversation(Base):
     __table_args__ = (
         sa.PrimaryKeyConstraint("id", name="conversation_pkey"),
         sa.Index("conversation_app_from_user_idx", "app_id", "from_source", "from_end_user_id"),
+        sa.Index(
+            "conversation_app_created_at_idx",
+            "app_id",
+            sa.text("created_at DESC"),
+            postgresql_where=sa.text("is_deleted IS false"),
+        ),
+        sa.Index(
+            "conversation_app_updated_at_idx",
+            "app_id",
+            sa.text("updated_at DESC"),
+            postgresql_where=sa.text("is_deleted IS false"),
+        ),
     )
 
     id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()))
@@ -811,24 +1092,26 @@ class Conversation(Base):
         self._inputs = inputs
 
     @property
-    def model_config(self):
-        model_config = {}
+    def model_config(self) -> AppModelConfigDict:
+        model_config = cast(AppModelConfigDict, {})
         app_model_config: AppModelConfig | None = None
 
         if self.mode == AppMode.ADVANCED_CHAT:
             if self.override_model_configs:
                 override_model_configs = json.loads(self.override_model_configs)
-                model_config = override_model_configs
+                model_config = cast(AppModelConfigDict, override_model_configs)
         else:
             if self.override_model_configs:
                 override_model_configs = json.loads(self.override_model_configs)
 
                 if "model" in override_model_configs:
                     # where is app_id?
-                    app_model_config = AppModelConfig(app_id=self.app_id).from_model_config_dict(override_model_configs)
+                    app_model_config = AppModelConfig(app_id=self.app_id).from_model_config_dict(
+                        cast(AppModelConfigDict, override_model_configs)
+                    )
                     model_config = app_model_config.to_dict()
                 else:
-                    model_config["configs"] = override_model_configs
+                    model_config["configs"] = override_model_configs  # type: ignore[typeddict-unknown-key]
             else:
                 app_model_config = (
                     db.session.query(AppModelConfig).where(AppModelConfig.id == self.app_model_config_id).first()
@@ -1003,7 +1286,7 @@ class Conversation(Base):
     def in_debug_mode(self) -> bool:
         return self.override_model_configs is not None
 
-    def to_dict(self) -> dict[str, Any]:
+    def to_dict(self) -> ConversationDict:
         return {
             "id": self.id,
             "app_id": self.app_id,
@@ -1283,7 +1566,7 @@ class Message(Base):
         return self.message_metadata_dict.get("retriever_resources") if self.message_metadata else []
 
     @property
-    def message_files(self) -> list[dict[str, Any]]:
+    def message_files(self) -> list[MessageFileInfo]:
         from factories import file_factory
 
         message_files = db.session.scalars(select(MessageFile).where(MessageFile.message_id == self.id)).all()
@@ -1338,10 +1621,13 @@ class Message(Base):
                 )
             files.append(file)
 
-        result: list[dict[str, Any]] = [
-            {"belongs_to": message_file.belongs_to, "upload_file_id": message_file.upload_file_id, **file.to_dict()}
-            for (file, message_file) in zip(files, message_files)
-        ]
+        result = cast(
+            list[MessageFileInfo],
+            [
+                {"belongs_to": message_file.belongs_to, "upload_file_id": message_file.upload_file_id, **file.to_dict()}
+                for (file, message_file) in zip(files, message_files)
+            ],
+        )
 
         db.session.commit()
         return result
@@ -1351,7 +1637,7 @@ class Message(Base):
         self._extra_contents = list(contents)
 
     @property
-    def extra_contents(self) -> list[dict[str, Any]]:
+    def extra_contents(self) -> list[ExtraContentDict]:
         return getattr(self, "_extra_contents", [])
 
     @property
@@ -1367,7 +1653,7 @@ class Message(Base):
 
         return None
 
-    def to_dict(self) -> dict[str, Any]:
+    def to_dict(self) -> MessageDict:
         return {
             "id": self.id,
             "app_id": self.app_id,
@@ -1391,7 +1677,7 @@ class Message(Base):
         }
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> Message:
+    def from_dict(cls, data: MessageDict) -> Message:
         return cls(
             id=data["id"],
             app_id=data["app_id"],
@@ -1451,7 +1737,7 @@ class MessageFeedback(TypeBase):
         account = db.session.query(Account).where(Account.id == self.from_account_id).first()
         return account
 
-    def to_dict(self) -> dict[str, Any]:
+    def to_dict(self) -> MessageFeedbackDict:
         return {
             "id": str(self.id),
             "app_id": str(self.app_id),
@@ -1714,8 +2000,8 @@ class AppMCPServer(TypeBase):
             return result
 
     @property
-    def parameters_dict(self) -> dict[str, Any]:
-        return cast(dict[str, Any], json.loads(self.parameters))
+    def parameters_dict(self) -> dict[str, str]:
+        return cast(dict[str, str], json.loads(self.parameters))
 
 
 class Site(Base):
@@ -2155,7 +2441,7 @@ class TraceAppConfig(TypeBase):
     def tracing_config_str(self) -> str:
         return json.dumps(self.tracing_config_dict)
 
-    def to_dict(self) -> dict[str, Any]:
+    def to_dict(self) -> TraceAppConfigDict:
         return {
             "id": self.id,
             "app_id": self.app_id,

api/pyproject.toml

@@ -35,7 +35,7 @@ dependencies = [
     "jsonschema>=4.25.1",
     "langfuse~=2.51.3",
     "langsmith~=0.1.77",
-    "markdown~=3.5.1",
+    "markdown~=3.8.1",
     "mlflow-skinny>=3.0.0",
     "numpy~=1.26.4",
     "openpyxl~=3.1.5",
@@ -113,7 +113,7 @@ dev = [
     "dotenv-linter~=0.5.0",
     "faker~=38.2.0",
     "lxml-stubs~=0.5.1",
-    "basedpyright~=1.31.0",
+    "basedpyright~=1.38.2",
     "ruff~=0.14.0",
     "pytest~=8.3.2",
     "pytest-benchmark~=4.0.0",
@@ -167,12 +167,12 @@ dev = [
     "import-linter>=2.3",
     "types-redis>=4.6.0.20241004",
     "celery-types>=0.23.0",
-    "mypy~=1.17.1",
+    "mypy~=1.19.1",
     # "locust>=2.40.4",  # Temporarily removed due to compatibility issues. Uncomment when resolved.
     "sseclient-py>=1.8.0",
     "pytest-timeout>=2.4.0",
     "pytest-xdist>=3.8.0",
-    "pyrefly>=0.54.0",
+    "pyrefly>=0.55.0",
 ]
 
 ############################################################
@@ -247,3 +247,13 @@ module = [
     "extensions.logstore.repositories.logstore_api_workflow_run_repository",
 ]
 ignore_errors = true
+
+[tool.pyrefly]
+project-includes = ["."]
+project-excludes = [
+    ".venv",
+    "migrations/",
+]
+python-platform = "linux"
+python-version = "3.11.0"
+infer-with-first-use = false

api/pyrefly-local-excludes.txt

@@ -0,0 +1,200 @@
+configs/middleware/cache/redis_pubsub_config.py
+controllers/console/app/annotation.py
+controllers/console/app/app.py
+controllers/console/app/app_import.py
+controllers/console/app/mcp_server.py
+controllers/console/app/site.py
+controllers/console/auth/email_register.py
+controllers/console/human_input_form.py
+controllers/console/init_validate.py
+controllers/console/ping.py
+controllers/console/setup.py
+controllers/console/version.py
+controllers/console/workspace/trigger_providers.py
+controllers/service_api/app/annotation.py
+controllers/web/workflow_events.py
+core/agent/fc_agent_runner.py
+core/app/apps/advanced_chat/app_generator.py
+core/app/apps/advanced_chat/app_runner.py
+core/app/apps/advanced_chat/generate_task_pipeline.py
+core/app/apps/agent_chat/app_generator.py
+core/app/apps/base_app_generate_response_converter.py
+core/app/apps/base_app_generator.py
+core/app/apps/chat/app_generator.py
+core/app/apps/common/workflow_response_converter.py
+core/app/apps/completion/app_generator.py
+core/app/apps/pipeline/pipeline_generator.py
+core/app/apps/pipeline/pipeline_runner.py
+core/app/apps/workflow/app_generator.py
+core/app/apps/workflow/app_runner.py
+core/app/apps/workflow/generate_task_pipeline.py
+core/app/apps/workflow_app_runner.py
+core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py
+core/datasource/datasource_manager.py
+core/external_data_tool/api/api.py
+core/llm_generator/llm_generator.py
+core/llm_generator/output_parser/structured_output.py
+core/mcp/mcp_client.py
+core/ops/aliyun_trace/data_exporter/traceclient.py
+core/ops/arize_phoenix_trace/arize_phoenix_trace.py
+core/ops/mlflow_trace/mlflow_trace.py
+core/ops/ops_trace_manager.py
+core/ops/tencent_trace/client.py
+core/ops/tencent_trace/utils.py
+core/plugin/backwards_invocation/base.py
+core/plugin/backwards_invocation/model.py
+core/prompt/utils/extract_thread_messages.py
+core/rag/datasource/keyword/jieba/jieba.py
+core/rag/datasource/keyword/jieba/jieba_keyword_table_handler.py
+core/rag/datasource/vdb/analyticdb/analyticdb_vector.py
+core/rag/datasource/vdb/analyticdb/analyticdb_vector_openapi.py
+core/rag/datasource/vdb/baidu/baidu_vector.py
+core/rag/datasource/vdb/chroma/chroma_vector.py
+core/rag/datasource/vdb/clickzetta/clickzetta_vector.py
+core/rag/datasource/vdb/couchbase/couchbase_vector.py
+core/rag/datasource/vdb/elasticsearch/elasticsearch_vector.py
+core/rag/datasource/vdb/huawei/huawei_cloud_vector.py
+core/rag/datasource/vdb/lindorm/lindorm_vector.py
+core/rag/datasource/vdb/matrixone/matrixone_vector.py
+core/rag/datasource/vdb/milvus/milvus_vector.py
+core/rag/datasource/vdb/myscale/myscale_vector.py
+core/rag/datasource/vdb/oceanbase/oceanbase_vector.py
+core/rag/datasource/vdb/opensearch/opensearch_vector.py
+core/rag/datasource/vdb/oracle/oraclevector.py
+core/rag/datasource/vdb/pgvecto_rs/pgvecto_rs.py
+core/rag/datasource/vdb/relyt/relyt_vector.py
+core/rag/datasource/vdb/tablestore/tablestore_vector.py
+core/rag/datasource/vdb/tencent/tencent_vector.py
+core/rag/datasource/vdb/tidb_on_qdrant/tidb_on_qdrant_vector.py
+core/rag/datasource/vdb/tidb_on_qdrant/tidb_service.py
+core/rag/datasource/vdb/tidb_vector/tidb_vector.py
+core/rag/datasource/vdb/upstash/upstash_vector.py
+core/rag/datasource/vdb/vikingdb/vikingdb_vector.py
+core/rag/datasource/vdb/weaviate/weaviate_vector.py
+core/rag/extractor/csv_extractor.py
+core/rag/extractor/excel_extractor.py
+core/rag/extractor/firecrawl/firecrawl_app.py
+core/rag/extractor/firecrawl/firecrawl_web_extractor.py
+core/rag/extractor/html_extractor.py
+core/rag/extractor/jina_reader_extractor.py
+core/rag/extractor/markdown_extractor.py
+core/rag/extractor/notion_extractor.py
+core/rag/extractor/pdf_extractor.py
+core/rag/extractor/text_extractor.py
+core/rag/extractor/unstructured/unstructured_doc_extractor.py
+core/rag/extractor/unstructured/unstructured_eml_extractor.py
+core/rag/extractor/unstructured/unstructured_epub_extractor.py
+core/rag/extractor/unstructured/unstructured_markdown_extractor.py
+core/rag/extractor/unstructured/unstructured_msg_extractor.py
+core/rag/extractor/unstructured/unstructured_ppt_extractor.py
+core/rag/extractor/unstructured/unstructured_pptx_extractor.py
+core/rag/extractor/unstructured/unstructured_xml_extractor.py
+core/rag/extractor/watercrawl/client.py
+core/rag/extractor/watercrawl/extractor.py
+core/rag/extractor/watercrawl/provider.py
+core/rag/extractor/word_extractor.py
+core/rag/index_processor/processor/paragraph_index_processor.py
+core/rag/index_processor/processor/parent_child_index_processor.py
+core/rag/index_processor/processor/qa_index_processor.py
+core/rag/retrieval/router/multi_dataset_function_call_router.py
+core/rag/summary_index/summary_index.py
+core/repositories/sqlalchemy_workflow_execution_repository.py
+core/repositories/sqlalchemy_workflow_node_execution_repository.py
+core/tools/__base/tool.py
+core/tools/mcp_tool/provider.py
+core/tools/plugin_tool/provider.py
+core/tools/utils/message_transformer.py
+core/tools/utils/web_reader_tool.py
+core/tools/workflow_as_tool/provider.py
+core/trigger/debug/event_selectors.py
+core/trigger/entities/entities.py
+core/trigger/provider.py
+core/workflow/workflow_entry.py
+dify_graph/entities/workflow_execution.py
+dify_graph/file/file_manager.py
+dify_graph/graph_engine/error_handler.py
+dify_graph/graph_engine/layers/execution_limits.py
+dify_graph/nodes/agent/agent_node.py
+dify_graph/nodes/base/node.py
+dify_graph/nodes/code/code_node.py
+dify_graph/nodes/datasource/datasource_node.py
+dify_graph/nodes/document_extractor/node.py
+dify_graph/nodes/human_input/human_input_node.py
+dify_graph/nodes/if_else/if_else_node.py
+dify_graph/nodes/iteration/iteration_node.py
+dify_graph/nodes/knowledge_index/knowledge_index_node.py
+dify_graph/nodes/knowledge_retrieval/knowledge_retrieval_node.py
+dify_graph/nodes/list_operator/node.py
+dify_graph/nodes/llm/node.py
+dify_graph/nodes/loop/loop_node.py
+dify_graph/nodes/parameter_extractor/parameter_extractor_node.py
+dify_graph/nodes/question_classifier/question_classifier_node.py
+dify_graph/nodes/start/start_node.py
+dify_graph/nodes/template_transform/template_transform_node.py
+dify_graph/nodes/tool/tool_node.py
+dify_graph/nodes/trigger_plugin/trigger_event_node.py
+dify_graph/nodes/trigger_schedule/trigger_schedule_node.py
+dify_graph/nodes/trigger_webhook/node.py
+dify_graph/nodes/variable_aggregator/variable_aggregator_node.py
+dify_graph/nodes/variable_assigner/v1/node.py
+dify_graph/nodes/variable_assigner/v2/node.py
+dify_graph/variables/types.py
+extensions/ext_fastopenapi.py
+extensions/logstore/repositories/logstore_api_workflow_run_repository.py
+extensions/otel/instrumentation.py
+extensions/otel/runtime.py
+extensions/storage/aliyun_oss_storage.py
+extensions/storage/aws_s3_storage.py
+extensions/storage/azure_blob_storage.py
+extensions/storage/baidu_obs_storage.py
+extensions/storage/clickzetta_volume/clickzetta_volume_storage.py
+extensions/storage/clickzetta_volume/file_lifecycle.py
+extensions/storage/google_cloud_storage.py
+extensions/storage/huawei_obs_storage.py
+extensions/storage/opendal_storage.py
+extensions/storage/oracle_oci_storage.py
+extensions/storage/supabase_storage.py
+extensions/storage/tencent_cos_storage.py
+extensions/storage/volcengine_tos_storage.py
+factories/variable_factory.py
+libs/external_api.py
+libs/gmpy2_pkcs10aep_cipher.py
+libs/helper.py
+libs/login.py
+libs/module_loading.py
+libs/oauth.py
+libs/oauth_data_source.py
+models/trigger.py
+models/workflow.py
+repositories/sqlalchemy_api_workflow_node_execution_repository.py
+repositories/sqlalchemy_api_workflow_run_repository.py
+repositories/sqlalchemy_execution_extra_content_repository.py
+schedule/queue_monitor_task.py
+services/account_service.py
+services/audio_service.py
+services/auth/firecrawl/firecrawl.py
+services/auth/jina.py
+services/auth/jina/jina.py
+services/auth/watercrawl/watercrawl.py
+services/conversation_service.py
+services/dataset_service.py
+services/document_indexing_proxy/document_indexing_task_proxy.py
+services/document_indexing_proxy/duplicate_document_indexing_task_proxy.py
+services/external_knowledge_service.py
+services/plugin/plugin_migration.py
+services/recommend_app/buildin/buildin_retrieval.py
+services/recommend_app/database/database_retrieval.py
+services/recommend_app/remote/remote_retrieval.py
+services/summary_index_service.py
+services/tools/tools_transform_service.py
+services/trigger/trigger_provider_service.py
+services/trigger/trigger_subscription_builder_service.py
+services/trigger/webhook_service.py
+services/workflow_draft_variable_service.py
+services/workflow_event_snapshot_service.py
+services/workflow_service.py
+tasks/app_generate/workflow_execute_task.py
+tasks/regenerate_summary_index_task.py
+tasks/trigger_processing_tasks.py
+tasks/workflow_cfs_scheduler/cfs_scheduler.py
+tasks/workflow_execution_tasks.py

api/pyrefly.toml

@@ -1,8 +0,0 @@
-project-includes = ["."]
-project-excludes = [
-    ".venv",
-    "migrations/",
-]
-python-platform = "linux"
-python-version = "3.11.0"
-infer-with-first-use = false

api/services/account_service.py

@@ -74,6 +74,16 @@ from tasks.mail_reset_password_task import (
 logger = logging.getLogger(__name__)
 
 
+def _try_join_enterprise_default_workspace(account_id: str) -> None:
+    """Best-effort join to enterprise default workspace."""
+    if not dify_config.ENTERPRISE_ENABLED:
+        return
+
+    from services.enterprise.enterprise_service import try_join_default_workspace
+
+    try_join_default_workspace(account_id)
+
+
 class TokenPair(BaseModel):
     access_token: str
     refresh_token: str
@@ -287,13 +297,14 @@ class AccountService:
             email=email, name=name, interface_language=interface_language, password=password
         )
 
-        TenantService.create_owner_tenant_if_not_exist(account=account)
+        try:
+            TenantService.create_owner_tenant_if_not_exist(account=account)
+        except Exception:
+            # Enterprise-only side-effect should run independently from personal workspace creation.
+            _try_join_enterprise_default_workspace(str(account.id))
+            raise
 
-        # Enterprise-only: best-effort add the account to the default workspace (does not switch current workspace).
-        if dify_config.ENTERPRISE_ENABLED:
-            from services.enterprise.enterprise_service import try_join_default_workspace
-
-            try_join_default_workspace(str(account.id))
+        _try_join_enterprise_default_workspace(str(account.id))
 
         return account
 
@@ -1407,18 +1418,18 @@ class RegisterService:
                 and create_workspace_required
                 and FeatureService.get_system_features().license.workspaces.is_available()
             ):
-                tenant = TenantService.create_tenant(f"{account.name}'s Workspace")
-                TenantService.create_tenant_member(tenant, account, role="owner")
-                account.current_tenant = tenant
-                tenant_was_created.send(tenant)
+                try:
+                    tenant = TenantService.create_tenant(f"{account.name}'s Workspace")
+                    TenantService.create_tenant_member(tenant, account, role="owner")
+                    account.current_tenant = tenant
+                    tenant_was_created.send(tenant)
+                except Exception:
+                    _try_join_enterprise_default_workspace(str(account.id))
+                    raise
 
             db.session.commit()
 
-            # Enterprise-only: best-effort add the account to the default workspace (does not switch current workspace).
-            if dify_config.ENTERPRISE_ENABLED:
-                from services.enterprise.enterprise_service import try_join_default_workspace
-
-                try_join_default_workspace(str(account.id))
+            _try_join_enterprise_default_workspace(str(account.id))
         except WorkSpaceNotAllowedCreateError:
             db.session.rollback()
             logger.exception("Register failed")

api/services/app_dsl_service.py

@@ -4,6 +4,7 @@ import logging
 import uuid
 from collections.abc import Mapping
 from enum import StrEnum
+from typing import cast
 from urllib.parse import urlparse
 from uuid import uuid4
 
@@ -32,7 +33,7 @@ from extensions.ext_redis import redis_client
 from factories import variable_factory
 from libs.datetime_utils import naive_utc_now
 from models import Account, App, AppMode
-from models.model import AppModelConfig, IconType
+from models.model import AppModelConfig, AppModelConfigDict, IconType
 from models.workflow import Workflow
 from services.plugin.dependencies_analysis import DependenciesAnalysisService
 from services.workflow_draft_variable_service import WorkflowDraftVariableService
@@ -523,7 +524,7 @@ class AppDslService:
             if not app.app_model_config:
                 app_model_config = AppModelConfig(
                     app_id=app.id, created_by=account.id, updated_by=account.id
-                ).from_model_config_dict(model_config)
+                ).from_model_config_dict(cast(AppModelConfigDict, model_config))
                 app_model_config.id = str(uuid4())
                 app.app_model_config_id = app_model_config.id
 

api/services/app_generate_service.py

@@ -38,6 +38,13 @@ if TYPE_CHECKING:
 class AppGenerateService:
     @staticmethod
     def _build_streaming_task_on_subscribe(start_task: Callable[[], None]) -> Callable[[], None]:
+        """
+        Build a subscription callback that coordinates when the background task starts.
+
+        - streams transport: start immediately (events are durable; late subscribers can replay).
+        - pubsub/sharded transport: start on first subscribe, with a short fallback timer so the task
+          still runs if the client never connects.
+        """
         started = False
         lock = threading.Lock()
 
@@ -54,10 +61,18 @@ class AppGenerateService:
                 started = True
                 return True
 
-        # XXX(QuantumGhost): dirty hacks to avoid a race between publisher and SSE subscriber.
-        # The Celery task may publish the first event before the API side actually subscribes,
-        # causing an "at most once" drop with Redis Pub/Sub. We start the task on subscribe,
-        # but also use a short fallback timer so the task still runs if the client never consumes.
+        channel_type = dify_config.PUBSUB_REDIS_CHANNEL_TYPE
+        if channel_type == "streams":
+            # With Redis Streams, we can safely start right away; consumers can read past events.
+            _try_start()
+
+            # Keep return type Callable[[], None] consistent while allowing an extra (no-op) call.
+            def _on_subscribe_streams() -> None:
+                _try_start()
+
+            return _on_subscribe_streams
+
+        # Pub/Sub modes (at-most-once): subscribe-gated start with a tiny fallback.
         timer = threading.Timer(SSE_TASK_START_FALLBACK_MS / 1000.0, _try_start)
         timer.daemon = True
         timer.start()

api/services/app_model_config_service.py

@@ -1,12 +1,12 @@
 from core.app.apps.agent_chat.app_config_manager import AgentChatAppConfigManager
 from core.app.apps.chat.app_config_manager import ChatAppConfigManager
 from core.app.apps.completion.app_config_manager import CompletionAppConfigManager
-from models.model import AppMode
+from models.model import AppMode, AppModelConfigDict
 
 
 class AppModelConfigService:
     @classmethod
-    def validate_configuration(cls, tenant_id: str, config: dict, app_mode: AppMode):
+    def validate_configuration(cls, tenant_id: str, config: dict, app_mode: AppMode) -> AppModelConfigDict:
         if app_mode == AppMode.CHAT:
             return ChatAppConfigManager.config_validate(tenant_id, config)
         elif app_mode == AppMode.AGENT_CHAT:

api/services/app_service.py

@@ -1,6 +1,6 @@
 import json
 import logging
-from typing import TypedDict, cast
+from typing import Any, TypedDict, cast
 
 import sqlalchemy as sa
 from flask_sqlalchemy.pagination import Pagination
@@ -187,7 +187,7 @@ class AppService:
             for tool in agent_mode.get("tools") or []:
                 if not isinstance(tool, dict) or len(tool.keys()) <= 3:
                     continue
-                agent_tool_entity = AgentToolEntity(**tool)
+                agent_tool_entity = AgentToolEntity(**cast(dict[str, Any], tool))
                 # get tool
                 try:
                     tool_runtime = ToolManager.get_agent_tool_runtime(
@@ -388,7 +388,7 @@ class AppService:
             agent_config = app_model_config.agent_mode_dict
 
             # get all tools
-            tools = agent_config.get("tools", [])
+            tools = cast(list[dict[str, Any]], agent_config.get("tools", []))
 
         url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"
 

api/services/audio_service.py

@@ -2,6 +2,7 @@ import io
 import logging
 import uuid
 from collections.abc import Generator
+from typing import cast
 
 from flask import Response, stream_with_context
 from werkzeug.datastructures import FileStorage
@@ -106,7 +107,7 @@ class AudioService:
                         if not text_to_speech_dict.get("enabled"):
                             raise ValueError("TTS is not enabled")
 
-                        voice = text_to_speech_dict.get("voice")
+                        voice = cast(str | None, text_to_speech_dict.get("voice"))
 
             model_manager = ModelManager()
             model_instance = model_manager.get_default_model_instance(

api/services/human_input_service.py

@@ -130,7 +130,7 @@ class HumanInputService:
         if isinstance(session_factory, Engine):
             session_factory = sessionmaker(bind=session_factory)
         self._session_factory = session_factory
-        self._form_repository = form_repository or HumanInputFormSubmissionRepository(session_factory)
+        self._form_repository = form_repository or HumanInputFormSubmissionRepository()
 
     def get_form_by_token(self, form_token: str) -> Form | None:
         record = self._form_repository.get_by_token(form_token)

api/services/workflow_app_service.py

@@ -7,7 +7,7 @@ from sqlalchemy import and_, func, or_, select
 from sqlalchemy.orm import Session
 
 from dify_graph.enums import WorkflowExecutionStatus
-from models import Account, App, EndUser, WorkflowAppLog, WorkflowArchiveLog, WorkflowRun
+from models import Account, App, EndUser, TenantAccountJoin, WorkflowAppLog, WorkflowArchiveLog, WorkflowRun
 from models.enums import AppTriggerType, CreatorUserRole
 from models.trigger import WorkflowTriggerLog
 from services.plugin.plugin_service import PluginService
@@ -132,7 +132,14 @@ class WorkflowAppService:
                 ),
             )
         if created_by_account:
-            account = session.scalar(select(Account).where(Account.email == created_by_account))
+            account = session.scalar(
+                select(Account)
+                .join(TenantAccountJoin, TenantAccountJoin.account_id == Account.id)
+                .where(
+                    Account.email == created_by_account,
+                    TenantAccountJoin.tenant_id == app_model.tenant_id,
+                )
+            )
             if not account:
                 raise ValueError(f"Account not found: {created_by_account}")
 

api/services/workflow_service.py

@@ -11,13 +11,13 @@ from sqlalchemy.orm import Session, sessionmaker
 from configs import dify_config
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfigManager
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
-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.repositories import DifyCoreRepositoryFactory
 from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
 from core.workflow.workflow_entry import WorkflowEntry
 from dify_graph.entities import GraphInitParams, WorkflowNodeExecution
 from dify_graph.entities.pause_reason import HumanInputRequired
-from dify_graph.enums import ErrorStrategy, UserFrom, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from dify_graph.enums import ErrorStrategy, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from dify_graph.errors import WorkflowNodeRunFailedError
 from dify_graph.file import File
 from dify_graph.graph_events import GraphNodeEventBase, NodeRunFailedEvent, NodeRunSucceededEvent
@@ -1015,7 +1015,7 @@ class WorkflowService:
         rendered_content: str,
         resolved_default_values: Mapping[str, Any],
     ) -> tuple[str, list[DeliveryTestEmailRecipient]]:
-        repo = HumanInputFormRepositoryImpl(session_factory=db.engine, tenant_id=app_model.tenant_id)
+        repo = HumanInputFormRepositoryImpl(tenant_id=app_model.tenant_id)
         params = FormCreateParams(
             app_id=app_model.id,
             workflow_execution_id=None,
@@ -1063,13 +1063,15 @@ class WorkflowService:
         variable_pool: VariablePool,
     ) -> HumanInputNode:
         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=account.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=account.id,
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+            ),
             call_depth=0,
         )
         graph_runtime_state = GraphRuntimeState(
@@ -1081,6 +1083,7 @@ class WorkflowService:
             config=node_config,
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
+            form_repository=HumanInputFormRepositoryImpl(tenant_id=workflow.tenant_id),
         )
         return node
 

api/tasks/app_generate/workflow_execute_task.py

@@ -321,7 +321,13 @@ def _resume_app_execution(payload: dict[str, Any]) -> None:
                 return
 
             message = session.scalar(
-                select(Message).where(Message.workflow_run_id == workflow_run_id).order_by(Message.created_at.desc())
+                select(Message)
+                .where(
+                    Message.conversation_id == conversation.id,
+                    Message.workflow_run_id == workflow_run_id,
+                )
+                .order_by(Message.created_at.desc())
+                .limit(1)
             )
             if message is None:
                 logger.warning("Message not found for workflow run %s", workflow_run_id)

api/tasks/generate_summary_index_task.py

@@ -14,7 +14,7 @@ from services.summary_index_service import SummaryIndexService
 logger = logging.getLogger(__name__)
 
 
-@shared_task(queue="dataset")
+@shared_task(queue="dataset_summary")
 def generate_summary_index_task(dataset_id: str, document_id: str, segment_ids: list[str] | None = None):
     """
     Async generate summary index for document segments.

api/tasks/human_input_timeout_tasks.py

@@ -58,7 +58,7 @@ def check_and_handle_human_input_timeouts(limit: int = 100) -> None:
     """Scan for expired human input forms and resume or end workflows."""
 
     session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
-    form_repo = HumanInputFormSubmissionRepository(session_factory)
+    form_repo = HumanInputFormSubmissionRepository()
     service = HumanInputService(session_factory, form_repository=form_repo)
     now = naive_utc_now()
     global_timeout_seconds = dify_config.HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS

api/tasks/regenerate_summary_index_task.py

@@ -16,7 +16,7 @@ from services.summary_index_service import SummaryIndexService
 logger = logging.getLogger(__name__)
 
 
-@shared_task(queue="dataset")
+@shared_task(queue="dataset_summary")
 def regenerate_summary_index_task(
     dataset_id: str,
     regenerate_reason: str = "summary_model_changed",

api/tests/integration_tests/workflow/nodes/test_code.py

@@ -4,10 +4,9 @@ import uuid
 import pytest
 
 from configs import dify_config
-from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
 from core.workflow.node_factory import DifyNodeFactory
-from dify_graph.entities import GraphInitParams
-from dify_graph.enums import UserFrom, WorkflowNodeExecutionStatus
+from dify_graph.enums import WorkflowNodeExecutionStatus
 from dify_graph.graph import Graph
 from dify_graph.node_events import NodeRunResult
 from dify_graph.nodes.code.code_node import CodeNode
@@ -15,6 +14,7 @@ from dify_graph.nodes.code.limits import CodeNodeLimits
 from dify_graph.runtime import GraphRuntimeState, VariablePool
 from dify_graph.system_variable import SystemVariable
 from tests.integration_tests.workflow.nodes.__mock.code_executor import setup_code_executor_mock
+from tests.workflow_test_utils import build_test_graph_init_params
 
 CODE_MAX_STRING_LENGTH = dify_config.CODE_MAX_STRING_LENGTH
 
@@ -31,11 +31,11 @@ def init_code_node(code_config: dict):
         "nodes": [{"data": {"type": "start", "title": "Start"}, "id": "start"}, code_config],
     }
 
-    init_params = GraphInitParams(
-        tenant_id="1",
-        app_id="1",
+    init_params = build_test_graph_init_params(
         workflow_id="1",
         graph_config=graph_config,
+        tenant_id="1",
+        app_id="1",
         user_id="1",
         user_from=UserFrom.ACCOUNT,
         invoke_from=InvokeFrom.DEBUGGER,

api/tests/integration_tests/workflow/nodes/test_http.py

@@ -5,18 +5,18 @@ from urllib.parse import urlencode
 import pytest
 
 from configs import dify_config
-from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
 from core.helper.ssrf_proxy import ssrf_proxy
 from core.tools.tool_file_manager import ToolFileManager
 from core.workflow.node_factory import DifyNodeFactory
-from dify_graph.entities import GraphInitParams
-from dify_graph.enums import UserFrom, WorkflowNodeExecutionStatus
+from dify_graph.enums import WorkflowNodeExecutionStatus
 from dify_graph.file.file_manager import file_manager
 from dify_graph.graph import Graph
 from dify_graph.nodes.http_request import HttpRequestNode, HttpRequestNodeConfig
 from dify_graph.runtime import GraphRuntimeState, VariablePool
 from dify_graph.system_variable import SystemVariable
 from tests.integration_tests.workflow.nodes.__mock.http import setup_http_mock
+from tests.workflow_test_utils import build_test_graph_init_params
 
 HTTP_REQUEST_CONFIG = HttpRequestNodeConfig(
     max_connect_timeout=dify_config.HTTP_REQUEST_MAX_CONNECT_TIMEOUT,
@@ -41,11 +41,11 @@ def init_http_node(config: dict):
         "nodes": [{"data": {"type": "start", "title": "Start"}, "id": "start"}, config],
     }
 
-    init_params = GraphInitParams(
-        tenant_id="1",
-        app_id="1",
+    init_params = build_test_graph_init_params(
         workflow_id="1",
         graph_config=graph_config,
+        tenant_id="1",
+        app_id="1",
         user_id="1",
         user_from=UserFrom.ACCOUNT,
         invoke_from=InvokeFrom.DEBUGGER,
@@ -685,11 +685,11 @@ def test_nested_object_variable_selector(setup_http_mock):
         ],
     }
 
-    init_params = GraphInitParams(
-        tenant_id="1",
-        app_id="1",
+    init_params = build_test_graph_init_params(
         workflow_id="1",
         graph_config=graph_config,
+        tenant_id="1",
+        app_id="1",
         user_id="1",
         user_from=UserFrom.ACCOUNT,
         invoke_from=InvokeFrom.DEBUGGER,

api/tests/integration_tests/workflow/nodes/test_llm.py

@@ -4,17 +4,17 @@ import uuid
 from collections.abc import Generator
 from unittest.mock import MagicMock, patch
 
-from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
 from core.llm_generator.output_parser.structured_output import _parse_structured_output
 from core.model_manager import ModelInstance
-from dify_graph.entities import GraphInitParams
-from dify_graph.enums import UserFrom, WorkflowNodeExecutionStatus
+from dify_graph.enums import WorkflowNodeExecutionStatus
 from dify_graph.node_events import StreamCompletedEvent
 from dify_graph.nodes.llm.node import LLMNode
 from dify_graph.nodes.llm.protocols import CredentialsProvider, ModelFactory
 from dify_graph.runtime import GraphRuntimeState, VariablePool
 from dify_graph.system_variable import SystemVariable
 from extensions.ext_database import db
+from tests.workflow_test_utils import build_test_graph_init_params
 
 """FOR MOCK FIXTURES, DO NOT REMOVE"""
 
@@ -37,11 +37,11 @@ def init_llm_node(config: dict) -> LLMNode:
     workflow_id = "9d2074fc-6f86-45a9-b09d-6ecc63b9056d"
     user_id = "9d2074fc-6f86-45a9-b09d-6ecc63b9056e"
 
-    init_params = GraphInitParams(
-        tenant_id=tenant_id,
-        app_id=app_id,
+    init_params = build_test_graph_init_params(
         workflow_id=workflow_id,
         graph_config=graph_config,
+        tenant_id=tenant_id,
+        app_id=app_id,
         user_id=user_id,
         user_from=UserFrom.ACCOUNT,
         invoke_from=InvokeFrom.DEBUGGER,

api/tests/integration_tests/workflow/nodes/test_parameter_extractor.py

@@ -3,10 +3,9 @@ import time
 import uuid
 from unittest.mock import MagicMock
 
-from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
 from core.model_manager import ModelInstance
-from dify_graph.entities import GraphInitParams
-from dify_graph.enums import UserFrom, WorkflowNodeExecutionStatus
+from dify_graph.enums import WorkflowNodeExecutionStatus
 from dify_graph.model_runtime.entities import AssistantPromptMessage, UserPromptMessage
 from dify_graph.nodes.llm.protocols import CredentialsProvider, ModelFactory
 from dify_graph.nodes.parameter_extractor.parameter_extractor_node import ParameterExtractorNode
@@ -14,6 +13,7 @@ from dify_graph.runtime import GraphRuntimeState, VariablePool
 from dify_graph.system_variable import SystemVariable
 from extensions.ext_database import db
 from tests.integration_tests.workflow.nodes.__mock.model import get_mocked_fetch_model_instance
+from tests.workflow_test_utils import build_test_graph_init_params
 
 """FOR MOCK FIXTURES, DO NOT REMOVE"""
 from tests.integration_tests.model_runtime.__mock.plugin_daemon import setup_model_mock
@@ -43,11 +43,11 @@ def init_parameter_extractor_node(config: dict, memory=None):
         "nodes": [{"data": {"type": "start", "title": "Start"}, "id": "start"}, config],
     }
 
-    init_params = GraphInitParams(
-        tenant_id="1",
-        app_id="1",
+    init_params = build_test_graph_init_params(
         workflow_id="1",
         graph_config=graph_config,
+        tenant_id="1",
+        app_id="1",
         user_id="1",
         user_from=UserFrom.ACCOUNT,
         invoke_from=InvokeFrom.DEBUGGER,

api/tests/integration_tests/workflow/nodes/test_template_transform.py

@@ -1,15 +1,15 @@
 import time
 import uuid
 
-from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
 from core.workflow.node_factory import DifyNodeFactory
-from dify_graph.entities import GraphInitParams
-from dify_graph.enums import UserFrom, WorkflowNodeExecutionStatus
+from dify_graph.enums import WorkflowNodeExecutionStatus
 from dify_graph.graph import Graph
 from dify_graph.nodes.template_transform.template_renderer import TemplateRenderError
 from dify_graph.nodes.template_transform.template_transform_node import TemplateTransformNode
 from dify_graph.runtime import GraphRuntimeState, VariablePool
 from dify_graph.system_variable import SystemVariable
+from tests.workflow_test_utils import build_test_graph_init_params
 
 
 class _SimpleJinja2Renderer:
@@ -53,11 +53,11 @@ def test_execute_template_transform():
         "nodes": [{"data": {"type": "start", "title": "Start"}, "id": "start"}, config],
     }
 
-    init_params = GraphInitParams(
-        tenant_id="1",
-        app_id="1",
+    init_params = build_test_graph_init_params(
         workflow_id="1",
         graph_config=graph_config,
+        tenant_id="1",
+        app_id="1",
         user_id="1",
         user_from=UserFrom.ACCOUNT,
         invoke_from=InvokeFrom.DEBUGGER,