SSE Stream Lifecycle#

Dify's streaming APIs use Server-Sent Events (SSE) over a Redis channel (pub/sub or streams) to deliver workflow execution events to clients. The lifecycle spans five stages: stream startup, event delivery, idle/ping, pause handling, and execution cancellation.

Architecture Overview#

1. Stream Startup#

The SSE endpoint is exposed at:

• Web app: GET /api/workflow/<task_id>/events
• Service API: GET /workflow/<task_id>/events

Both controllers instantiate a MessageGenerator and call retrieve_events() 1, which derives a Redis topic key from app_mode + workflow_run_id and prepares a subscription. For Redis Streams channels (SupportsPreparedSubscription), retrieve_events() calls prepare_subscription() to read the current stream tail and create a subscription that will resume from that checkpoint — ensuring events published after the checkpoint are delivered even if they arrive before the subscriber starts iterating. For other transports, it calls subscribe() directly. The subscription is then passed to stream_topic_events(). If the workflow run is already finished, the controller skips live streaming and immediately emits a single workflow_finished event 2.

stream_topic_events() accepts a subscription parameter (an AsyncIterator of events) instead of deriving one from a topic — this allows the caller to fix the delivery boundary before passing it in. The function sends an immediate PING event on connection to prevent the SSE request from appearing stuck in browser DevTools 3, then fires the optional on_subscribe callback once the subscription context manager is entered — this callback gates Celery task startup for all transport types to prevent race conditions.

2. Event Loop, Idle Timeout, and Pings#

The core event loop polls Redis with a 1-second timeout per iteration 4:

• Message received → yield it; if the event type matches a terminal event, return and close the stream.
• No message (None) → if current_time - last_msg_time > idle_timeout (default 300 s), return. Otherwise emit a PING if ping_interval has elapsed (default 10 s).

Terminal events are normalized by _normalize_terminal_events(). By default, workflow_finished and workflow_paused are the only terminal values; receiving either closes the stream 5.

3. close_on_pause and continue_on_pause#

The default behavior closes the SSE stream when the workflow pauses at a Human Input (HITL) node. Clients that need to stay subscribed across multiple sequential Human Input nodes can pass continue_on_pause=true as a query parameter 6:

• continue_on_pause=false (default): terminal_events=None → resolves to {workflow_finished, workflow_paused} — stream closes on first pause.
• continue_on_pause=true: terminal_events=[] (empty set) → stream only closes on workflow_finished or idle timeout.

When include_state_snapshot=true is also set, close_on_pause=not continue_on_pause is threaded into build_workflow_event_stream(), which passes it to an internal _is_terminal_event() check during snapshot replay 7.

4. Execution Cancellation and Listener Lifecycle#

Execution cancellation is managed by AppExecutionCoordinator, which owns the cancellation policy for one execution attempt. The coordinator is scoped to a single attempt: resumed workflows create a new coordinator even when reusing the stable task ID.

Cancellation Paths#

Two conditions trigger execution cancellation:

1. Timeout: An independent watchdog timer runs for APP_MAX_EXECUTION_TIME seconds. If the timer expires, the coordinator calls request_abort() and publishes a QueueStopEvent via the queue manager's _publish_timeout_stop() callback.
2. Manual stop: listen() polls _is_stopped() on every iteration, reading a generate_task_stopped:{task_id} key from Redis with a 1-second TTL cache. If the stop flag is detected, the listener calls AppExecutionCoordinator.request_abort().

When request_abort() is called, it:

1. Calls set_app_task_stop_flag(task_id) to write the Redis stop key (TTL 600 s).
2. Calls GraphEngineManager(redis_client).send_stop_command(task_id, reason=...) to signal the graphon GraphEngine directly.

Both the legacy Redis stop flag and the GraphEngine command path are updated to ensure cancellation propagates to all execution backends. The _lock and _abort_sent guards ensure the abort logic runs at most once per coordinator.

User-initiated stops call AppQueueManager.set_stop_flag(), which performs a user ownership check before writing the same Redis key that _is_stopped() reads.

Listener Detachment Is Not Cancellation#

When the listen() generator's finally block runs (stream closed by the client or terminal event), it calls AppExecutionCoordinator.listener_closed() with a segment_completed flag derived from _listener_segment_completed. This is an observation, not an implicit cancellation signal — streaming workflow execution may continue in a Celery worker and publish durable events that a later subscriber can retrieve via the /workflow/<task_id>/events endpoint. For Redis Streams, the durable nature of the transport allows late subscribers to replay events from a known checkpoint, ensuring no events are lost when the subscriber reconnects.

The coordinator logs listener detachment but does not issue stop commands unless the detachment was preceded by an explicit timeout or manual stop. This decouples HTTP transport lifecycle from execution policy.

State Transitions#

AppQueueManager.stop_listen() accepts an execution_state parameter to distinguish between PAUSED and TERMINAL listener completions:

• AppExecutionState.PAUSED: The workflow reached a Human Input node. The coordinator transitions to PAUSED state and cancels the watchdog timer. A subsequent resume creates a fresh coordinator and does not inherit cancellation from the earlier segment.
• AppExecutionState.TERMINAL: The workflow finished, failed, or encountered an error. The coordinator transitions to TERMINAL state and cancels the watchdog timer.

The _listener_segment_completed event is set when stop_listen() is called, gating the listener_closed() logic.

5. Pause/Resume and ResponseStreamFilter State#

On pause, PauseStatePersistenceLayer intercepts GraphRunPausedEvent and serializes two state components into WorkflowResumptionContext (persisted to the database) 11:

On resume, WorkflowResumptionContext.get_response_stream_filter() calls ResponseStreamFilter.loads() and the restored filter is threaded through iter_dify_graph_engine_events() in workflow_entry.py 11.

Before PR #38540 (Dify ≤ 1.15.0): Resume always instantiated a fresh ResponseStreamFilter. If a conditional-branch edge was traversed before the pause, the new filter had no record of it; the downstream Answer node's paths_map entry never emptied, permanently blocking token streaming — even though workflow_finished reported succeeded 12. Known affected issue reports: #38432, #38315, #38614, #38525.

6. Terminal Event Safety#

If the Celery worker fails before the generator emits any events, _publish_failed_terminal_event() synthesizes a workflow_finished(FAILED) so the SSE loop closes rather than hanging for 300 s 13. Similarly, if the generator exhausts without emitting a terminal event, a fallback failed event is published 13.

Key Source Files#