CodeVetter · Repo Unpack · Data flow
langgenius/dify
Every generate request follows the same spine: blueprint → flask-restx Resource → auth decorator (JWT passport, session, API token, or inner-api key) → Pydantic payload validation → `services/` orchestration → `core/app/apps/*` generator → `graphon` GraphEngine events → Redis pub/sub → SSE stream to the caller. Writes go through SQLAlchemy sessions (Postgres) with explicit tenant scoping; large payloads and files go through `ext_storage` backends; model/tool/plugin invocations egress through the plugin daemon over HTTP; arbitrary outbound HTTP from nodes/tools is forced through the Squid SSRF proxy. Config flows from pydantic-settings (env/.env/TOML/remote) into `app.config` at boot.
- Web-app requests authenticate via `WebApiResource.method_decorators = [validate_jwt_token]` → `decode_jwt_token()` verifies a `PassportService` JWT carrying `app_code`/`app_id`/`end_user_id`, then loads `App`, `Site`, and `EndUser` rows and rejects disabled sites or missing entities.
- `ChatApi.post` (POST `/api/chat-messages`) gates on app mode (`CHAT|AGENT_CHAT|ADVANCED_CHAT|AGENT`), validates `ChatMessagePayload` (inputs/query/files/response_mode/conversation_id/parent_message_id), eagerly resolves the conversation, then calls `AppGenerateService.generate` and streams via `helper.compact_generate_response`.
- `AppGenerateService.generate` wraps dispatch in `_run_with_guardrails`: on CLOUD edition it reserves `QuotaService.reserve(QuotaType.WORKFLOW)` with commit/refund semantics, then `RateLimit(app_model.id, max_active_requests)` bounds concurrent executions where the cap is `min(app.max_active_requests|APP_DEFAULT_ACTIVE_REQUESTS, APP_MAX_ACTIVE_REQUESTS)`.
- `_dispatch_generate` is a `match` on effective `AppMode` routing to `CompletionAppGenerator`, `AgentChatAppGenerator`, `AgentAppGenerator`, `ChatAppGenerator`, `AdvancedChatAppGenerator`, or `WorkflowAppGenerator` — each returning an event stream wrapped by `rate_limit.generate`.
- Streaming workflow/advanced-chat runs cross a process boundary: the request path serializes `AppExecutionParams` (discriminated `Account`/`EndUser` union, app_id, workflow_id, tenant_id, args, call_depth, workflow_run_id) to JSON and enqueues `workflow_based_app_execution_task.delay(payload_json)` on the `workflow_based_app_execution` queue only when the first SSE subscriber attaches (`on_subscribe` + 200ms fallback timer; streams channel type skips the timer).
- Inside the worker, `workflow_execute_task` rehydrates trusted state — reloads `App`, `Workflow`, and the user from Postgres via fresh sessions (never trusting the payload for identity beyond ids) — then drives `WorkflowAppGenerator`/`AdvancedChatAppGenerator` → `WorkflowEntry` → `graphon.GraphEngine.run()`, persisting `WorkflowRun`/`WorkflowNodeExecution` via `DifyCoreRepositoryFactory`/`DifyAPIRepositoryFactory`.
- Graph events are filtered before reaching the client: `iter_dify_graph_engine_events` pipes `engine.run()` through `filter_graph_events` with `HumanInputFormEventFilter` (backed by `HumanInputFormSubmissionRepository`) and `ResponseStreamFilter` which is deliberately shared with the pause-persistence layer so resumed runs keep consistent stream state.
- Streaming responses are plain SSE: `compact_generate_response` returns a `Response` with `mimetype="text/event-stream"` wrapping `_stream_with_request_context(generate())` — blocking mode returns a JSON `Mapping` response instead.
- A separate `length_prefixed_response` binary framing (1-byte magic + header/data length fields, little-endian) exists specifically for plugin-daemon compatibility, avoiding line-based streaming memory issues.
- Model/tool invocations egress to the external plugin daemon via `BasePluginClient._request_with_plugin_daemon_response` (e.g. `PluginAgentClient.fetch_agent_strategy_providers` GETs `plugin/{tenant_id}/management/agent_strategies`); streamed blobs are reassembled by `merge_blob_chunks`.
- Untrusted outbound HTTP (HTTP Request nodes, provider integrations, auth discovery, custom tools) is forced through `core/helper/ssrf_proxy.py`: pooled httpx clients mounted per-scheme on `SSRF_PROXY_HTTP(S)_URL` (Squid), with retry-on-[429,5xx] backoff and response-size guards (`ResponseTooLargeError`, `UnsupportedResponseEncodingError`).
- Enterprise license enforcement is a `before_request` concern: for `DEPLOYMENT_EDITION == ENTERPRISE`, `_match_license_gated_surface` gates `/console/api`, `/api`, `/v1`, `/mcp`, `/triggers` with surface-specific errors (session surfaces force logout 401, bearer surfaces 403, webhook surfaces 503 so senders retry); `/files`, `/inner/api`, `/health` are deliberately ungated, and bootstrap endpoints (system-features, setup, init, login, features, profile, workspaces/current, version, activate/check) are exempted to avoid signin reload loops.
- Files flow through `factories/file_factory` (builders, remote fetch, storage keys, validation) into `ext_storage` backends; `core/app/file_access` provides `DatabaseFileAccessController` scoping file access, and remote-file retrieval is split off from the generic SSRF client into `core/file/remote_fetcher` so signed file URLs resolve via DB+storage first.
- Async state crossing the broker is minimal and revalidated: `AppExecutionParams` carries only ids+args; `_ensure_workflow_service_mode_available` re-walks `workflow.walk_nodes()` for trigger nodes on manual invokes; `WORKFLOW_CALL_MAX_DEPTH` is enforced inside `WorkflowEntry.__init__` for nested workflow calls.
- Stop/cancel flows go through Redis: `POST /chat-messages/<task_id>/stop` → `AppTaskService.stop_task` sets a cancellation signal that the running generator polls, raising `GenerateTaskStoppedError`; `clear_app_task_cancellation_signals` cleans up on worker completion.