Report the failures a pipeline runs into

[markbackman]

Summary

Adds ErrorObserver, which reports every error a pipeline raises through on_error as an ErrorEvent.

PipelineWorker already reports errors through on_pipeline_error, and hands its handler the same frame this observer reads: the message, the category, the exception, and the processor that raised it. The difference is which errors get there. An error travels upstream from the processor that raised it, and a processor that answers for a failure itself stops it on the way: ServiceSwitcher swallows errors from the services it holds in reserve, and swallows an active service's error whenever the strategy responds by failing over. A bot that spent a session failing over between three TTS providers looks, through on_pipeline_error, like a bot that never had a problem.

This observer reads each error at the push that raises it, before anything downstream can answer for it, so a session's failure history holds the recoveries as well as the failures that surfaced. Use on_pipeline_error to hear about the errors the pipeline as a whole had to deal with, and this to hear about all of them.

• Reported once. An error is pushed again by every processor it travels through, and only the first of those pushes comes from the processor that failed, so that push is the one that becomes an event.
• Recovery is reported too. When a switcher can't recover, it re-reports the failure under its own name through push_error — a separate error, and a second event, which is what happened.
• Attribution and cause are read as push_error settled them. An ErrorFrame assembled by hand carries neither, so it is attributed to the processor pushing it, with ErrorCategory.UNKNOWN for its cause. The field is never None, so a query never has to case on it.
• processor_usable is read after the verdict. push_error_frame settles usability before the frame travels, so the event carries the state the error left the processor in. This is what separates a bad minute from the end of a capability, and it is the non-deprecated answer to the question fatal used to ask — fatal is removed in 2.0.0 and deliberately absent from the record.

{
  "message": "the provider said no",
  "category": "connectivity",
  "exception_type": "ConnectionError",
  "processor": "stt",
  "processor_usable": true,
  "timestamp": 1756944000.0
}

exception_type is there to group by: messages carry the particulars of a single occurrence, and are usually too specific to aggregate.

An exception that crashes a processor task never becomes an ErrorFrame, and so is outside what this observer or on_pipeline_error can see.

Testing

uv run pytest tests/test_error_observer.py — 8 tests, covering the observer in a running pipeline (origin attribution, category inferred from the exception, usability after a permanent failure) and against frames pushed directly (reported once however far it travels, each error its own event, hand-assembled frames, non-error frames ignored).

observer = ErrorObserver()

@observer.event_handler("on_error")
async def on_error(observer, event):
    logger.info(event.model_dump_json())

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

Files with missing lines	Coverage Δ
src/pipecat/observers/error_observer.py	100.00% <100.00%> (ø)

... and 2 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[aconchillo]

LGTM!