From e22c80610ed52661df1c01de85df180ad51a6217 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Fri, 16 Aug 2024 23:17:31 -0700
Subject: [PATCH] frames: add new FatalErrorFrame

---
 src/pipecat/frames/frames.py              | 17 +++++++++++++++--
 src/pipecat/pipeline/task.py              | 11 ++++++-----
 src/pipecat/processors/frame_processor.py |  8 +++++++-
 3 files changed, 28 insertions(+), 8 deletions(-)

diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py
index 68ec1ec38..9cd4288fe 100644
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -238,11 +238,24 @@ class CancelFrame(SystemFrame):
 @dataclass
 class ErrorFrame(SystemFrame):
     """This is used notify upstream that an error has occurred downstream the
-    pipeline."""
+    pipeline. A fatal error indicates the error is unrecoverable and that the
+    bot should exit.
+
+    """
     error: str
+    fatal: bool = False
 
     def __str__(self):
-        return f"{self.name}(error: {self.error})"
+        return f"{self.name}(error: {self.error}, fatal: {self.fatal})"
+
+
+@dataclass
+class FatalErrorFrame(ErrorFrame):
+    """This is used notify upstream that an unrecoverable error has occurred and
+    that the bot should exit.
+
+    """
+    fatal: bool = field(default=True, init=False)
 
 
 @dataclass
diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 1f09ad233..102b4528b 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -50,11 +50,12 @@ class Source(FrameProcessor):
 
     async def _handle_upstream_frame(self, frame: Frame):
         if isinstance(frame, ErrorFrame):
-            logger.error(f"Error running app: {frame.error}")
-            # Cancel all tasks downstream.
-            await self.push_frame(CancelFrame())
-            # Tell the task we should stop.
-            await self._up_queue.put(StopTaskFrame())
+            logger.error(f"Error running app: {frame}")
+            if frame.fatal:
+                # Cancel all tasks downstream.
+                await self.push_frame(CancelFrame())
+                # Tell the task we should stop.
+                await self._up_queue.put(StopTaskFrame())
 
 
 class PipelineTask:
diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index 5bc47fd03..156e1c0ae 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/src/pipecat/processors/frame_processor.py
@@ -9,7 +9,13 @@ import time
 
 from enum import Enum
 
-from pipecat.frames.frames import ErrorFrame, Frame, MetricsFrame, StartFrame, StartInterruptionFrame, UserStoppedSpeakingFrame
+from pipecat.frames.frames import (
+    ErrorFrame,
+    Frame,
+    MetricsFrame,
+    StartFrame,
+    StartInterruptionFrame,
+    UserStoppedSpeakingFrame)
 from pipecat.utils.utils import obj_count, obj_id
 
 from loguru import logger