Error Frames

def __str__(self):
    """
    Returns formatted string with error message and fatal status
    """

@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)

# Handle non-fatal error
async def handle_service_error(error: Exception):
    await pipeline.push_frame(ErrorFrame(
        error=f"Service error: {str(error)}",
        fatal=False
    ))

# Handle fatal error
async def handle_critical_error(error: Exception):
    await pipeline.push_frame(FatalErrorFrame(
        error=f"Critical error: {str(error)}"
    ))

async def process_frame(self, frame: Frame):
    if isinstance(frame, ErrorFrame):
        if frame.fatal:
            await self.shutdown_pipeline()
        else:
            await self.attempt_recovery()

# Error-aware pipeline
pipeline = Pipeline([
    input_processor,
    error_handler,        # Processes ErrorFrames
    recovery_processor,   # Handles recovery
    output_processor
])

# Error handler implementation
class ErrorHandler(Processor):
    async def process_frame(self, frame: Frame):
        if isinstance(frame, ErrorFrame):
            if frame.fatal:
                logger.critical(f"Fatal error: {frame.error}")
                await self.push_frame(CancelFrame())
            else:
                logger.error(f"Error: {frame.error}")
                await self.handle_error(frame)

async def handle_service_error(self, error: Exception):
    if is_temporary_error(error):
        # Non-fatal error with retry
        await self.push_frame(ErrorFrame(
            error="Temporary service disruption",
            fatal=False
        ))
        await self.retry_operation()
    else:
        # Fatal error requiring shutdown
        await self.push_frame(FatalErrorFrame(
            error="Service permanently unavailable"
        ))

class ErrorAggregator(Processor):
    def __init__(self):
        self.errors = []

    async def process_frame(self, frame: Frame):
        if isinstance(frame, ErrorFrame):
            self.errors.append({
                'timestamp': time.time(),
                'message': frame.error,
                'fatal': frame.fatal
            })
            await self.update_error_metrics()