Report the failures a pipeline runs into

changelog/5618.added.md

@@ -0,0 +1 @@
+- Added `ErrorObserver`, which reports every error a pipeline raises through `on_error` as an `ErrorEvent`: the `message`, the `category` it was attributed to, the `exception_type` behind it, the `processor` that raised it, and whether that processor can still do its job (`processor_usable`). Errors are read where they are raised, so ones a processor answers for itself — a `ServiceSwitcher` failing over, or a service it holds in reserve — are reported too; `PipelineWorker`'s `on_pipeline_error` sees only the errors that reach the top of the pipeline.

src/pipecat/observers/error_observer.py

@@ -0,0 +1,115 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Observer reporting the failures a pipeline runs into.
+
+Errors travel upstream from the processor that raised them, and not all of
+them reach the end of that journey: a processor that answers for a failure
+itself — a service switcher that fails over to its next service, say — stops
+the error there. This observer reads each error where it is raised, so a
+session's failure history holds the ones that were recovered from as well as
+the ones that surfaced.
+"""
+
+import time
+from collections.abc import Callable
+
+from pydantic import BaseModel
+
+from pipecat.frames.frames import ErrorFrame
+from pipecat.observers.base_observer import BaseObserver, FramePushed
+from pipecat.utils.errors import ErrorCategory
+
+
+class ErrorEvent(BaseModel):
+    """One failure, as the processor that raised it described it.
+
+    Parameters:
+        message: What went wrong, in the words of the processor that failed.
+        category: Why it failed, drawn from :class:`ErrorCategory` and
+            independent of the provider that failed: rejected credentials, an
+            unreachable service, a malformed request and so on.
+        exception_type: The name of the exception behind the failure, where one
+            caused it. Failures group by this where a message, carrying the
+            particulars of a single occurrence, is too specific to group by.
+        processor: The name of the processor that raised the error.
+        processor_usable: Whether that processor can still do its job. A
+            processor that can't keeps failing for as long as it is given work,
+            so this separates a bad minute from the end of a capability.
+        timestamp: Unix timestamp of the failure.
+    """
+
+    message: str
+    category: ErrorCategory
+    exception_type: str | None = None
+    processor: str
+    processor_usable: bool
+    timestamp: float
+
+
+class ErrorObserver(BaseObserver):
+    """Reports each error a pipeline raises, once, where it is raised.
+
+    An error is reported at its origin rather than where it ends up, and named
+    for the processor that raised it rather than the one that passed it along.
+
+    Events:
+        on_error(observer, event): Emitted for each error, as an
+            :class:`ErrorEvent`.
+
+    Example::
+
+        observer = ErrorObserver()
+
+        @observer.event_handler("on_error")
+        async def on_error(observer, event):
+            logger.info(event.model_dump_json())
+    """
+
+    def __init__(self, *, time_source: Callable[[], float] = time.time, **kwargs):
+        """Initialize the error observer.
+
+        Args:
+            time_source: Reads the current time in seconds. Supplying one lets
+                a test place failures without waiting.
+            **kwargs: Additional arguments passed to parent class.
+        """
+        super().__init__(**kwargs)
+        self._now = time_source
+        self._reported: set[int] = set()
+
+        self._register_event_handler("on_error")
+
+    async def on_push_frame(self, data: FramePushed):
+        """Report an error frame, the first time it is seen.
+
+        An error is pushed again by every processor it travels through, and
+        only the first of those pushes comes from the processor that failed.
+
+        Args:
+            data: Frame push event containing the frame and direction.
+        """
+        frame = data.frame
+        if not isinstance(frame, ErrorFrame) or frame.id in self._reported:
+            return
+
+        self._reported.add(frame.id)
+
+        # An error assembled by hand rather than reported through `push_error`
+        # arrives without the processor and category that method settles, so
+        # attribute it to the processor pushing it and report its cause as unknown.
+        processor = frame.processor or data.source
+        await self._call_event_handler(
+            "on_error",
+            ErrorEvent(
+                message=frame.error,
+                category=frame.category or ErrorCategory.UNKNOWN,
+                exception_type=type(frame.exception).__name__ if frame.exception else None,
+                processor=processor.name,
+                processor_usable=processor.is_usable,
+                timestamp=self._now(),
+            ),
+        )

tests/test_error_observer.py

@@ -0,0 +1,154 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import unittest
+
+from pipecat.frames.frames import ErrorFrame, Frame, TextFrame
+from pipecat.observers.base_observer import FramePushed
+from pipecat.observers.error_observer import ErrorObserver
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.processors.filters.identity_filter import IdentityFilter
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.tests.utils import SleepFrame, run_test
+from pipecat.utils.asyncio.task_manager import TaskManager
+from pipecat.utils.errors import ErrorCategory
+
+
+class FailingProcessor(FrameProcessor):
+    """A processor that fails the way a service does, on being given work."""
+
+    def __init__(self, exception: Exception, category: ErrorCategory | None = None, **kwargs):
+        super().__init__(**kwargs)
+        self._exception = exception
+        self._category = category
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if isinstance(frame, TextFrame):
+            await self.push_error(
+                "the provider said no", exception=self._exception, category=self._category
+            )
+        else:
+            await self.push_frame(frame, direction)
+
+
+class TestErrorObserverInAPipeline(unittest.IsolatedAsyncioTestCase):
+    """Errors as they are raised, through a running pipeline."""
+
+    async def _errors_from(self, exception: Exception, category: ErrorCategory | None = None):
+        observer = ErrorObserver()
+        events = []
+
+        @observer.event_handler("on_error")
+        async def on_error(observer, event):
+            events.append(event)
+
+        failing = FailingProcessor(exception, category, name="stt")
+        await run_test(
+            Pipeline([failing]),
+            frames_to_send=[TextFrame("work"), SleepFrame(sleep=0.1)],
+            expected_down_frames=[],
+            observers=[observer],
+        )
+        return events
+
+    async def test_an_error_is_reported_where_it_is_raised(self):
+        """Named for the processor that failed, not the ones it travels through."""
+        (event,) = await self._errors_from(ConnectionError("no route to host"))
+
+        self.assertEqual(event.processor, "stt")
+        self.assertEqual(event.message, "the provider said no")
+        self.assertEqual(event.exception_type, "ConnectionError")
+        # Worked out from the exception, since the processor named no category.
+        self.assertEqual(event.category, ErrorCategory.CONNECTIVITY)
+
+    async def test_a_recoverable_failure_leaves_the_processor_usable(self):
+        """An unreachable service may well answer the next request."""
+        (event,) = await self._errors_from(ConnectionError("no route to host"))
+
+        self.assertTrue(event.processor_usable)
+
+    async def test_a_permanent_failure_costs_the_processor_its_usability(self):
+        """Rejected credentials stay rejected, so the capability is gone."""
+        (event,) = await self._errors_from(
+            Exception("invalid api key"), ErrorCategory.AUTHENTICATION
+        )
+
+        self.assertFalse(event.processor_usable)
+
+
+class TestErrorObserver(unittest.IsolatedAsyncioTestCase):
+    """What the observer makes of the frames it is shown."""
+
+    async def asyncSetUp(self):
+        self.clock = 1_000_000.0
+        self.observer = ErrorObserver(time_source=lambda: self.clock)
+        # Event handlers run as tasks, so the observer needs a task manager.
+        await self.observer.setup(TaskManager())
+        self.events = []
+
+        @self.observer.event_handler("on_error")
+        async def on_error(observer, event):
+            self.events.append(event)
+
+    async def _push(self, frame, source=None):
+        """Feed one frame to the observer, as a pipeline push would."""
+        await self.observer.on_push_frame(
+            FramePushed(
+                source=source or IdentityFilter(name="source"),
+                destination=IdentityFilter(name="destination"),
+                frame=frame,
+                direction=FrameDirection.UPSTREAM,
+                timestamp=0,
+            )
+        )
+        await self._settle()
+
+    async def _settle(self):
+        import asyncio
+
+        await asyncio.sleep(0.01)
+
+    async def test_an_error_is_reported_once_however_far_it_travels(self):
+        """Every processor it passes through pushes it again."""
+        failing = IdentityFilter(name="tts")
+        error = ErrorFrame(error="failed", processor=failing, category=ErrorCategory.SERVER)
+
+        await self._push(error, source=failing)
+        await self._push(error, source=IdentityFilter(name="passing it along"))
+        await self._push(error, source=IdentityFilter(name="and along"))
+
+        (event,) = self.events
+        self.assertEqual(event.processor, "tts")
+
+    async def test_each_error_is_its_own_event(self):
+        """A processor that fails twice failed twice."""
+        failing = IdentityFilter(name="tts")
+
+        await self._push(ErrorFrame(error="first", processor=failing), source=failing)
+        await self._push(ErrorFrame(error="second", processor=failing), source=failing)
+
+        self.assertEqual([event.message for event in self.events], ["first", "second"])
+
+    async def test_an_error_reports_when_it_happened(self):
+        await self._push(ErrorFrame(error="failed"))
+
+        (event,) = self.events
+        self.assertEqual(event.timestamp, 1_000_000.0)
+
+    async def test_an_error_assembled_by_hand_is_attributed_to_its_pusher(self):
+        """`push_error` settles both of these; a bare frame carries neither."""
+        await self._push(ErrorFrame(error="failed"), source=IdentityFilter(name="llm"))
+
+        (event,) = self.events
+        self.assertEqual(event.processor, "llm")
+        self.assertEqual(event.category, ErrorCategory.UNKNOWN)
+        self.assertIsNone(event.exception_type)
+
+    async def test_frames_that_are_not_errors_are_not_reported(self):
+        await self._push(TextFrame("hello"))
+
+        self.assertEqual(self.events, [])