# Data flow — langgenius/dify

> Canonical page: https://codevetter.com/unpack/dify/data-flow

# Data flow — langgenius/dify

Scanned commit 3a017e8cccb316f23f601af65d2e8ce92e68f788 (2026-09-22). Generated by codevetter codevetter 1.12.0 on 2026-09-22.

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. — [api/controllers/web/wraps.py#L26-44](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/controllers/web/wraps.py#L26-L44) [api/controllers/web/wraps.py#L47-102](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/controllers/web/wraps.py#L47-L102) [api/libs/passport.py](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/libs/passport.py)
- `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`. — [api/controllers/web/completion.py#L191-263](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/controllers/web/completion.py#L191-L263)
- `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)`. — [api/services/app_generate_service.py#L137-167](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/services/app_generate_service.py#L137-L167) [api/services/app_generate_service.py#L370-389](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/services/app_generate_service.py#L370-L389) [api/services/quota_service.py](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/services/quota_service.py)
- `_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`. — [api/services/app_generate_service.py#L169-358](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/services/app_generate_service.py#L169-L358)
- 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). — [api/services/app_generate_service.py#L249-278](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/services/app_generate_service.py#L249-L278) [api/services/app_generate_service.py#L55-97](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/services/app_generate_service.py#L55-L97) [api/tasks/app_generate/workflow_execute_task.py#L42-120](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/tasks/app_generate/workflow_execute_task.py#L42-L120)
- 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`. — [api/tasks/app_generate/workflow_execute_task.py#L1-60](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/tasks/app_generate/workflow_execute_task.py#L1-L60) [api/repositories/factory.py](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/repositories/factory.py) [api/core/app/apps/workflow/app_generator.py#L72-120](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/core/app/apps/workflow/app_generator.py#L72-L120)
- 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. — [api/core/workflow/workflow_entry.py#L56-79](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/core/workflow/workflow_entry.py#L56-L79)
- 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. — [api/libs/helper.py#L416-435](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/libs/helper.py#L416-L435)
- 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. — [api/libs/helper.py#L438-499](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/libs/helper.py#L438-L499)
- 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`. — [api/core/plugin/impl/base.py](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/core/plugin/impl/base.py) [api/core/plugin/impl/agent.py#L13-40](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/core/plugin/impl/agent.py#L13-L40)
- 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`). — [api/core/helper/ssrf_proxy.py#L1-80](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/core/helper/ssrf_proxy.py#L1-L80) [docker/ssrf_proxy/squid.conf.template](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/docker/ssrf_proxy/squid.conf.template)
- 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. — [api/app_factory.py#L24-51](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/app_factory.py#L24-L51) [api/app_factory.py#L70-94](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/app_factory.py#L70-L94) [api/app_factory.py#L113-130](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/app_factory.py#L113-L130)
- 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. — [api/factories/file_factory](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/factories/file_factory) [api/core/app/file_access/controller.py](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/core/app/file_access/controller.py) [api/core/helper/ssrf_proxy.py#L1-8](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/core/helper/ssrf_proxy.py#L1-L8)
- 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. — [api/tasks/app_generate/workflow_execute_task.py#L85-120](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/tasks/app_generate/workflow_execute_task.py#L85-L120) [api/services/app_generate_service.py#L360-368](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/services/app_generate_service.py#L360-L368) [api/core/workflow/workflow_entry.py#L138-141](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/core/workflow/workflow_entry.py#L138-L141)
- 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. — [api/controllers/web/completion.py#L266-294](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/controllers/web/completion.py#L266-L294) [api/services/app_task_service.py](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/services/app_task_service.py) [api/tasks/app_generate/workflow_execute_task.py#L15](https://github.com/langgenius/dify/blob/3a017e8cccb316f23f601af65d2e8ce92e68f788/api/tasks/app_generate/workflow_execute_task.py#L15)

Every claim links to its cited files at the pinned commit. Citations identify source evidence; they do not establish runtime correctness. Independent CodeVetter scan — not reviewed by upstream maintainers.

## Public product links

- [CodeVetter](https://codevetter.com/)
- [Download](https://codevetter.com/download)
- [Documentation](https://codevetter.com/docs/)
- [Source](https://github.com/Codevetter/codevetter)