Workflow Resume Architecture#

Dify workflows pause at Human Input (HITL) nodes and resume in a separate Celery task, process, and thread from the original execution. The resume path reuses _generate() — the same internal method as a fresh run — but supplies a restored GraphRuntimeState and a newly constructed AppQueueManager, skipping the input-preparation and SSE-subscribe handshake that a first-run requires 1.

Two invariants must hold across every pause boundary 2:

1. GraphRuntimeState — variable pool, LLM usage counters, execution context.
2. ResponseStreamFilter — graphon's streaming gate; its paths_map tracks which edges must be traversed before an Answer node may emit tokens. A stale or fresh filter will permanently block downstream Answer nodes after resume.

Pause: What Gets Persisted#

PauseStatePersistenceLayer is a GraphEngineLayer that fires synchronously on every GraphRunPausedEvent — the persistence runs in the execution thread to guarantee the pause record exists before the human-input form is surfaced 3. It:

1. Serializes GraphRuntimeState via GraphRuntimeState.dumps() 4.
2. Serializes ResponseStreamFilter via ResponseStreamFilter.dumps() — must be the exact same instance WorkflowEntry is using; a different instance silently persists empty state 5.
3. Wraps both plus the generate_entity (discriminated union of WorkflowAppGenerateEntity | AdvancedChatAppGenerateEntity) into a WorkflowResumptionContext Pydantic model 6.
4. Translates graphon session IDs to Dify form IDs via enrich_graph_pause_reasons() and writes the context to object storage; the DB WorkflowPause record stores only the state_object_key 7 8.

PauseStatePersistenceLayer is injected only when _generate() receives a PauseStateLayerConfig — absent in debug/single-iteration runs 9.

Resume Entry Points#

Both entry points converge on _resume_app_execution() 10:

_resume_app_execution() executes these steps 11:

1. Fetches WorkflowPause from DB → loads state bytes from object storage via pause_entity.get_state().
2. Deserializes via WorkflowResumptionContext.loads().
3. Restores GraphRuntimeState via GraphRuntimeState.from_snapshot(...).
4. Restores ResponseStreamFilter via resumption_context.get_response_stream_filter().
5. Re-fetches live ORM objects (WorkflowRun, Workflow, App, user) from the DB.
6. Dispatches to _resume_workflow() or _resume_advanced_chat() based on entity type.

Bypass of the Normal generate() Flow#

A fresh run goes generate() → _generate() → worker thread → WorkflowAppRunner.run(), with generate() preparing user inputs and establishing the SSE subscriber before events flow.

On resume, both WorkflowAppGenerator.resume() and AdvancedChatAppGenerator.resume() skip directly to _generate() with the pre-restored state 12. Key differences from a fresh run:

• _prepare_user_inputs() is not called. User inputs were already embedded in GraphRuntimeState.
• stream=True is forced via model_copy(update={"stream": True}) regardless of the persisted flag 13.
• trace_manager is rebuilt as a new TraceQueueManager — it is excluded from Pydantic serialization 14.
• Events publish immediately — _resume_workflow() and _resume_advanced_chat() call _publish_streaming_response() directly after the generator is returned, without waiting for an SSE subscriber 15.

New AppQueueManager on Resume#

_generate() always constructs a fresh AppQueueManager 16. The new instance uses the task_id preserved in WorkflowResumptionContext.generate_entity and wires to the same Redis topic key derived from workflow_run_id — so SSE consumers that subscribe to that key will receive events from the resumed execution.

The restored GraphRuntimeState is attached to the new queue manager before WorkflowEntry is initialized 17.

GraphRuntimeState Restoration in WorkflowAppRunner#

WorkflowAppRunner.run() branches on whether self._resume_graph_runtime_state is set 18:

• Resume path: uses the provided GraphRuntimeState directly; the variable pool comes from graph_runtime_state.variable_pool — no build_system_variables or VariablePool() construction.
• Fresh path: builds a new VariablePool, adds system + bootstrap variables, constructs a new GraphRuntimeState.

Both paths pass graph_runtime_state into WorkflowEntry, where the graph engine picks up execution from the paused position.

Known Issues and Active Work#

Race condition with CeleryWorkflowExecutionRepository (issue #40445): When CORE_WORKFLOW_EXECUTION_REPOSITORY=CeleryWorkflowExecutionRepository is set, save() enqueues save_workflow_execution_task asynchronously. If the workflow reaches a Human Input node before that task is processed, create_workflow_pause() queries for the WorkflowRun row synchronously and raises ValueError — leaving no durable pause record or resumption snapshot. The default SQLAlchemyWorkflowExecutionRepository commits synchronously and is not affected 19.

Durable-stream proposal (issue #41020): The existing Redis broadcast-channel abstraction does not carry a resume cursor, which means reconnecting API clients can miss or replay events. A maintainer-proposed DurableStreamTopic interface adds cursor-based subscriptions (subscribe_from_beginning, subscribe_from_cursor, subscribe_from_tail) with explicit CursorUnavailableError semantics. The first adapter will use Redis Streams; PostgreSQL is a future candidate 20.

Key Source Files#