CodeVetter · Repo Unpack · Data flow
langflow-ai/langflow
The core data path is: flow JSON (nodes+edges) persisted in the `flow` SQLModel table -> loaded owner-scoped or share-aware -> optionally sanitized by caller-aware component policy -> tweaks applied -> Graph.from_payload compiles vertices/edges -> graph.arun executes vertices and emits events -> RunResponse/SSE to the caller while messages, transactions, and vertex_builds rows persist execution history. Secrets (global variables, API keys) are Fernet-encrypted at rest using LANGFLOW_SECRET_KEY. Files go through a local/S3 StorageService namespaced per flow. Components themselves are discovered at boot from built-ins, components_path dirs, extension bundles, and remote bundle URLs, and cached in component_cache.
- Flow persistence: Flow rows store the graph as `data` JSON (nodes/edges/viewport) with name/endpoint_name/access_type/webhook/mcp_enabled metadata; cascade_delete_flow manually deletes MessageTable, TransactionTable, VertexBuildTable, FlowVersion, SpanTable/TraceTable, AuthzShare and the Flow row because SQLite FK cascades are not guaranteed.
- Graph materialization: simple_run_flow loads flow.data, applies caller-aware component policy (prepare_flow_build_for_user), then either serves a warm-registry template (try_warm_run_graph) or runs process_tweaks + apply_global_variable_defaults + Graph.from_payload(payload, flow_id, user_id, context).
- Tweaks and targeted inputs mutate the graph payload pre-instantiation via lfx.processing.process.process_tweaks/validate_targeted_inputs; refused tweaks raise TweakRefusedError mapped to a structured 422 (code TWEAKS_REFUSED) by an app-level handler, and non-owners are blocked from tweaking stored flows by _enforce_owner_only_tweaks.
- Execution: run_graph_internal sets graph.session_id then awaits graph.arun(inputs..., outputs, stream, event_manager); results return as list[RunOutputs] -> RunResponse; v1 responses and SSE frames re-project content_blocks via project_payload_to_v1 to keep the 1.11 wire shape.
- Streaming events: build/run events are produced into an asyncio.Queue owned by JobQueueService.create_queue(job_id) and consumed by get_build_events/consume_and_yield; a RedisJobQueueService backend (LANGFLOW_JOB_QUEUE_TYPE=redis) shares queues across gunicorn workers and is connectivity-probed at startup.
- Credential storage: DatabaseVariableService stores global variables; CREDENTIAL_TYPE values are Fernet-encrypted/decrypted via auth service helpers keyed on LANGFLOW_SECRET_KEY — short secrets derive the key with SHA-256, and a MultiFernet retains a legacy PRNG-derived key for decrypting pre-1.10.1 ciphertext.
- File inputs for public flows are restricted to the `{source_flow_id}/{basename}` namespace by validate_public_files (mitigation for GHSA-rcjh-r59h-gq37), and session IDs for anonymous public runs are namespaced per (client_id, flow_id) by scope_session_to_namespace / compute_virtual_flow_id (CVE-2026-33017).
- Component discovery at boot: lifespan calls get_and_cache_all_types_dict (lfx.interface.components) which merges built-in lfx components, settings.components_path directories, extension components (import_extension_components over DiscoveredExtension records), and bundles downloaded from settings.bundle_urls into temp dirs by load_bundles_from_urls.
- Execution history is persisted per run: JobService.create_job writes a WORKFLOW job row, vertex builds and transactions land in VertexBuildTable/TransactionTable, and startup cleanup prunes both tables plus authz_audit_log to configured retention (max_transactions_to_keep, max_vertex_builds_to_keep, AUTHZ_AUDIT_RETENTION_DAYS).
- Per-request identity context: session cookie/API-key auth resolves through get_current_user_for_workflow/api_key_security; a serving-plane end-user header is merged into the effective session_id by resolve_serving_scope and stamped on graph.end_user_id while execution still runs as flow.user_id; execution_client/execution_protocol contextvars label the surface for tracing.
- Durable/background runs: POST /api/v2/workflows mode=background dispatches to LangflowWorkflowHost.submit_background -> submit_background_with_mapping, with HITL pause/resume supported by a CheckpointService (lfx.graph.checkpoint CheckpointStore) and orphan sweeping on startup via background_execution service.
- Outbound calls a boot makes: models.dev catalog refresh (daily task, disabled by LANGFLOW_MODELS_DEV_REFRESH=false), bundle URL fetches, store webhooks (LANGFLOW_DOWNLOAD_WEBHOOK_URL/LIKE_WEBHOOK_URL), GitHub archive downloads via detect_github_url, and version checks via fetch_latest_version.