Workflow Resume Architecture#

Overview#

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 the same _generate() method as a fresh run but supplies a restored GraphRuntimeState and a new AppQueueManager, bypassing the initial input-preparation and SSE-subscribe handshake that a first-run goes through.

Two invariants must hold across the pause boundary:

1. GraphRuntimeState (variable pool, LLM usage counters, execution context) must be fully restored.
2. ResponseStreamFilter (graphon streaming gate, paths_map edge tracking) must be restored — otherwise downstream Answer nodes are permanently blocked from streaming.

Pause: What Gets Persisted#

PauseStatePersistenceLayer is a GraphEngineLayer that fires on every GraphRunPausedEvent 1. On each pause it:

1. Serializes GraphRuntimeState via self.graph_runtime_state.dumps() 2.
2. Serializes the ResponseStreamFilter via self._response_stream_filter.dumps() 3. The filter must be the same instance that WorkflowEntry is using — the layer's docstring explicitly warns that a different instance would silently persist the wrong (empty) state 4.
3. Wraps both plus the generate_entity (discriminated union of WorkflowAppGenerateEntity or AdvancedChatAppGenerateEntity) into a WorkflowResumptionContext.
4. Translates graphon session IDs to Dify form IDs via enrich_graph_pause_reasons() and calls repo.create_workflow_pause() 5.

WorkflowResumptionContext is a Pydantic model with:

• generate_entity — discriminated union (type-discriminated by AppMode)
• serialized_graph_runtime_state — GraphRuntimeState.dumps() output
• serialized_response_stream_filter_state — optional (None for runs paused before PR #38540; degrades to fresh-filter on resume for those stale runs)

PauseStatePersistenceLayer is injected as a GraphEngineLayer only when a PauseStateLayerConfig is provided to _generate() — it is absent during single-iteration/debug runs 6.

Resume Entry Points#

There are two resume entry points, both calling _resume_app_execution() in workflow_execute_task.py:

1. Celery task (resume_app_execution)#

resume_app_execution is a @shared_task on the workflow_based_app_execution queue. It delegates directly to _resume_app_execution().

2. UI-driven human-input resume#

The same _resume_app_execution() function is called from the human-input controller when a user submits a form.

In _resume_app_execution():

1. Fetches the persisted pause entity from the DB 7.
2. Deserializes via WorkflowResumptionContext.loads() 8.
3. Restores GraphRuntimeState via GraphRuntimeState.from_snapshot(resumption_context.serialized_graph_runtime_state) 9.
4. Restores ResponseStreamFilter via resumption_context.get_response_stream_filter() 10.
5. Re-fetches live ORM objects (WorkflowRun, Workflow, App, user) from the DB 11.
6. Dispatches to either _resume_advanced_chat() or _resume_workflow() based on entity type 12.

Bypass of the Normal generate() Flow#

A fresh execution goes through generate() → _generate() → new worker thread → WorkflowAppRunner.run(). The generate() call prepares user inputs and — in the streaming path dispatched via Celery — the SSE subscriber must attach before events are consumed.

On resume, both WorkflowAppGenerator.resume() and AdvancedChatAppGenerator.resume() skip directly to _generate(), passing the pre-restored graph_runtime_state and response_stream_filter. Crucially:

• Events are published immediately: _resume_advanced_chat() and _resume_workflow() call _publish_streaming_response() directly after the generator is returned, pushing events to the Redis topic without waiting for an SSE subscriber to attach first 13 14.
• No user-input preparation: resume() does not call _prepare_user_inputs().
• trace_manager is rebuilt: excluded from Pydantic serialization, it is reconstructed as a new TraceQueueManager on resume before _generate() is called 15 16.
• stream=True is forced: resumed_generate_entity is built with model_copy(update={"stream": True}) regardless of the persisted stream flag 17 18.

New AppQueueManager on Resume#

_generate() always constructs a fresh AppQueueManager 19 20. This new instance uses the task_id from the deserialized generate entity — preserved across the pause boundary as part of WorkflowResumptionContext.generate_entity. The queue manager is wired to the same Redis topic key derived from workflow_run_id, so SSE consumers can subscribe to events from the resumed execution.

The restored graph_runtime_state is attached to the new queue manager before WorkflowEntry is initialized 21.

GraphRuntimeState Restoration in WorkflowAppRunner#

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

• Resume path: uses the provided graph_runtime_state directly; 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 23.

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

Key Source Files#