Agent App Event Architecture#

Agent Apps in Dify use a distinct event emission pattern compared to Chat, Completion, and Advanced Chat (workflow-based) apps. Instead of running an in-process LLM loop, an Agent App delegates execution to the Agent backend service, consumes its streamed event flow, and re-publishes results through the standard EasyUI chat task pipeline using a narrow set of queue events 1.

The central implementation lives in AgentAppRunner (api/core/app/apps/agent_app/app_runner.py). Two companion helpers handle the pipeline's two "channels":

Process metadata (tool calls, thinking, observations) is recorded directly to the database via _AgentProcessRecorder, which also fires QueueAgentThoughtEvent for each created or updated MessageAgentThought row 2.

Event Flow: From Agent Backend to SSE#

Agent Backend (dify-agent service)
   │ SSE events (run_started, pydantic_ai_event, run_succeeded / run_failed)
   ▼
AgentAppRunner._consume_stream() ← api/core/app/apps/agent_app/app_runner.py
   │ adapts via AgentBackendRunEventAdapter
   ├─ AgentBackendAgentMessageDeltaInternalEvent → publish_agent_message_delta() → QueueAgentMessageEvent
   ├─ AgentBackendStreamInternalEvent → _AgentProcessRecorder → QueueAgentThoughtEvent
   └─ terminal (run_succeeded / run_failed) → publish_text_answer() → QueueLLMChunkEvent + QueueMessageEndEvent
               │
               ▼
EasyUIBasedGenerateTaskPipeline ← api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py
   ├─ QueueAgentMessageEvent → AgentMessageStreamResponse (streamed to client)
   ├─ QueueAgentThoughtEvent → AgentThoughtStreamResponse (DB fetch + streamed)
   ├─ QueueLLMChunkEvent → MessageStreamResponse (final answer chunk)
   └─ QueueMessageEndEvent → MessageEndStreamResponse (persists message)

The _consume_stream() loop runs until a terminal event is received. Text deltas from the agent are debounced by _TextDeltaDebouncer before being published to avoid a flood of single-character SSE frames. The loop tracks the last event ID (after) and passes the session_scope and binding_id to _cancel_run() when the queue manager signals a stop 3.

When cancellation is requested, _cancel_run() calls cancel_run_and_wait() on the Agent backend client to wait for backend cleanup to finish. It receives the terminal run_cancelled event, adapts it to AgentBackendRunCancelledInternalEvent, saves any included session_snapshot, and persists terminal usage to the Message record via _persist_message_usage() before raising GenerateTaskStoppedError. This ensures compositor cleanup completes, resumable session state is preserved, and usage accounting is captured even when the queue manager signals a stop.

The EasyUI pipeline dispatches QueueAgentMessageEvent by extracting the delta text and yielding an AgentMessageStreamResponse. It dispatches QueueAgentThoughtEvent by querying MessageAgentThought from the database and yielding an AgentThoughtStreamResponse.

Bypass of the Retriever Resources System#

Agent Apps never emit QueueRetrieverResourcesEvent. The AgentAppRunner imports only four queue event types 4: QueueAgentMessageEvent, QueueAgentThoughtEvent, QueueLLMChunkEvent, and QueueMessageEndEvent. QueueRetrieverResourcesEvent is absent by design.

In contrast, the advanced chat (workflow) pipeline does handle this event. MessageCycleManager.handle_retriever_resources() merges knowledge-base citations into task_state.metadata.retriever_resources (deduplicating by (dataset_id, document_id)), but only when the show_retrieve_source feature flag is enabled 5. The advanced chat pipeline registers a handler for it in its event dispatch map 6.

Practical implication: Agent Apps do not surface knowledge-base source citations in the EasyUI SSE stream. If retrieval traceability is needed, it must be implemented at the Agent backend layer or through a workflow-wrapped agent node (AgentNode) instead of an Agent App.

Process Recording: _AgentProcessRecorder#

_AgentProcessRecorder bridges the agent backend's raw stream events to the legacy MessageAgentThought DB model used by the EasyUI pipeline. It translates pydantic-ai event kinds into DB writes:

Tool thoughts are tracked by both index and tool_call_id via internal dictionaries 7, so that tool results can be correlated back to the correct open call even when events arrive out of order.

At the end of a successful run, trim_answer_suffix() removes any overlap between the incrementally streamed answer text and the final authoritative answer returned in the terminal event, preventing duplicate text in the DB row.

Session and Terminal Event Handling#

Terminal events from the Agent backend carry the final output plus an optional session_snapshot. The runner saves the session snapshot via _save_session() to preserve multi-turn state across conversation turns. Session management is handled through the Agent Working Environment architecture (Home Snapshots, Workspaces, and Bindings) introduced in the PR; each conversation-owned Agent App session now maintains a persistent AgentWorkspaceBinding that stores the resumable session snapshot rather than the previous cleanup-based model.

Success (AgentBackendRunSucceededInternalEvent) always carries a session_snapshot produced after compositor exit. Failed (AgentBackendRunFailedInternalEvent) and cancelled (AgentBackendRunCancelledInternalEvent) runs include an optional session_snapshot field. The snapshot is present only when compositor entry succeeded and layer exit completed; if the snapshot is None, the runner leaves the previously stored session snapshot untouched rather than copying the request snapshot or clearing session state.

All three terminal event types (AgentBackendRunSucceededInternalEvent, AgentBackendRunFailedInternalEvent, AgentBackendRunCancelledInternalEvent) now carry usage fields (as of PR #40937). The runner extracts provider-reported usage from each terminal event and persists it directly to the Message record via _persist_message_usage() before publishing the answer to the queue. This ensures usage accounting is retained regardless of whether the client SSE stream stays connected or the run succeeds—failed and cancelled runs now capture accumulated token usage that was previously lost.

The agent_session_scope_config_version_id parameter (previously agent_runtime_session_snapshot_id) identifies the draft or immutable config version whose Binding should be reused for this session.

If the terminal event is a AgentBackendDeferredToolCallInternalEvent (the agent invoked dify.ask_human), the runner calls _pause_for_ask_human(): it creates a HITL form, saves the pause correlation, and echoes the agent's question as the chat answer. The next turn resumes by threading the human's reply as deferred_tool_results 9.

Failure events are translated to standard InvokeError subclasses via _agent_backend_failure_to_exception(). When the RunFailedEvent includes an error_type field (a RunFailureType enum such as AGENT_RUN_LIMIT_EXCEEDED), the function uses that type directly to classify the error. Otherwise, it falls back to a reason-code lookup table. The resulting AgentBackendRunFailedError includes both the stable error_type field (when present) and the legacy reason field, ensuring consistent error propagation through the existing error-handling middleware.

Error Type Classification: RunFailureType is a string enum defined in dify_agent.protocol that provides stable, machine-readable categories for failed agent runs. The first defined type, AGENT_RUN_LIMIT_EXCEEDED, classifies execution budget violations (Pydantic AI request or step limits enforced by Dify Agent); it does not cover wall-clock run timeouts or provider/connection failures. The error_type field is optional in both RunFailedEvent.data and AgentBackendRunFailedInternalEvent, and is exposed through the public run status, event polling, and SSE APIs.