Trace Task Pipeline#

The trace task pipeline is Dify's observability system for shipping LLM run data to external tracing providers (Langfuse, LangSmith, OpenTelemetry, etc.). It converts runtime execution events into structured trace objects, batches them using a background threading timer, and delivers them asynchronously via Celery.

Full pipeline flow:

App invocation
  → Queue events (QueueErrorEvent / QueueStopEvent / QueueMessageEndEvent)
      → EasyUI / Workflow persistence layer
          → TraceTask enqueued in TraceQueueManager
              → background threading.Timer fires every N seconds
                  → TraceTask.execute() builds trace info (MessageTraceInfo / WorkflowTraceInfo)
                      → TaskData serialized to storage (JSON file)
                          → process_trace_tasks Celery task
                              → tracing provider (Langfuse, LangSmith, …)

Key source files#

Event-Based Trigger Points#

The pipeline is driven by events dequeued from AppQueueManager. Three event types determine whether a trace task is created:

• QueueErrorEvent — signals an LLM or runtime error. The base pipeline's handle_error() sets Message.status = ERROR and returns an ErrorStreamResponse. A MESSAGE_TRACE task is added in the chat pipeline's error handling path.
• QueueStopEvent — user-initiated stop or moderation halt. Handled alongside QueueMessageEndEvent in the EasyUI pipeline; triggers _save_message() which enqueues a MESSAGE_TRACE task 2.
• QueueMessageEndEvent — normal LLM completion. Also handled by _save_message(), which calls trace_manager.add_trace_task() 3.

The TraceQueueManager.add_trace_task() method is a guarded entry point: tasks are only enqueued when either enterprise telemetry is enabled or a third-party trace instance (e.g., Langfuse) is configured for the app 4.

Batched Dispatch via Background Timer#

TraceQueueManager owns a process-level threading.Timer and a queue.Queue shared across all manager instances 5.

• Interval and batch size are controlled by env vars TRACE_QUEUE_MANAGER_INTERVAL (default: 5 s) and TRACE_QUEUE_MANAGER_BATCH_SIZE (default: 100) 6.
• Each add_trace_task() call puts a task on the shared queue and ensures the timer is alive; if the timer is dead, a new one is started 4.
• When the timer fires, TraceQueueManager.run() calls collect_tasks() (drains up to batch_size items) then send_to_celery().
• send_to_celery() calls persist_trace_task() for each task, then enqueue_persisted_trace().
  • persist_trace_task() uses the ops_trace_payload_path() helper instead of hardcoding the file path format, automatically assigns operation_id to trace info if not set, and returns a file_info dict with file_id and app_id.
  • enqueue_persisted_trace() uses process_trace_tasks.apply_async() with a retry policy (max_retries=3, interval_start=0, interval_step=1, interval_max=2) instead of .delay().

The timer is a non-daemon thread (daemon=False), so it will finish its current run on process shutdown before the Python interpreter exits 7.

Asymmetric Error Behavior: Chat/Completion vs. Workflow Apps#

This is the most important behavioral difference to understand when debugging missing traces.

Chat / Completion apps (EasyUIBasedGenerateTaskPipeline)#

When QueueErrorEvent arrives, the pipeline calls handle_error(), then explicitly adds a MESSAGE_TRACE task via trace_manager.add_trace_task() before yielding an ErrorStreamResponse and breaking out of the event loop 8. _save_message() is never called, but the error trace task ensures LLM errors are captured alongside successful completions.

Workflow apps (WorkflowPersistenceLayer)#

Trace tasks are not tied to the queue event loop at all. They are emitted by the persistence layer's _enqueue_trace_task(), which is called from every terminal graph event handler:

Practical implication: Both workflow apps and chat/completion apps now emit trace tasks on error events, ensuring observability parity across app types.

Celery Task: process_trace_tasks#

The process_trace_tasks task runs on the "ops_trace" Celery queue.

Execution steps 13:

1. Resolve the JSON file path from file_info (app_id + file_id) and load it from storage.
2. Reconstitute typed model objects (Message, WorkflowRun, Document from raw dicts).
3. Instantiate the correct TraceInfo Pydantic model via trace_info_info_map.
4. Dispatch to enterprise telemetry (if enabled) — only once, guarded by _enterprise_trace_dispatched flag.
5. Dispatch to the configured provider trace instance (Langfuse, LangSmith, etc.) via trace_instance.trace(trace_info).
6. Delete the storage file in finally.

Retry contract 14: if a provider raises RetryableTraceDispatchError (e.g., Phoenix waiting for a parent span context), the task updates the stored file with the _enterprise_trace_dispatched flag and schedules self.retry(). Retry limits and delay are configured via OPS_TRACE_RETRYABLE_DISPATCH_MAX_RETRIES (default 780) and OPS_TRACE_RETRYABLE_DISPATCH_DELAY_SECONDS in dify_config 15. Terminal failures increment a Redis counter at {OPS_TRACE_FAILED_KEY}_{app_id} 16.

Trace Types and TraceTask Preprocessing#

TraceTask 17 is initialized with a trace_type from TraceTaskName. When execute() is called inside send_to_celery(), it dispatches to the appropriate builder method:

Each builder performs DB lookups to hydrate the trace info object. For MESSAGE_TRACE, app_id is read directly from Message.app_id (no join required), and tenant_id is resolved via a single App query 18. For WORKFLOW_TRACE, token counts are split by summing usage.prompt_tokens / usage.completion_tokens from all node execution outputs columns 19.

The TraceTask.app_id is None at construction; it must be set on the TraceTask instance before calling persist_trace_task(), preserving the originating application context. The app_id is used as the storage key prefix for the serialized JSON. This ensures that trace tasks retain the correct application context boundary even when processed by different queue managers.

Configuration Variables#

Unified Tracing Mode#

When OPS_TRACE_UNIFIED_ENABLED=true, providers registered in the unified registry use a provider-neutral tracing path that assembles Dify trace semantics once in Core and delegates only provider-specific translation and transport to lightweight adapters.

Registered unified providers:

• Phoenix (TracingProviderEnum.PHOENIX)
• LangSmith (TracingProviderEnum.LANGSMITH)

Providers not registered in the unified registry continue to use their existing implementations.

Parent context retention: The system validates that OPS_TRACE_PARENT_CONTEXT_TTL_SECONDS must be large enough to cover the retry window, calculated as OPS_TRACE_RETRYABLE_DISPATCH_MAX_RETRIES * OPS_TRACE_RETRYABLE_DISPATCH_DELAY_SECONDS. If the retry window exceeds the TTL, the application will raise a ValueError at startup.