refactor(backend): Improve Block Error Handling (#11366)

We need a way to differentiate between serious errors that cause on call
alerts and block errors.
This PR address this need by ensuring all errors that occur during
execution of a block are of Subtype BlockError

### Changes 🏗️

- Introduced BlockErrors and its subtypes
- Updated current errors that are emitted by blocks to use BlockError
- Update executor manager, to errors emitted when running a block that
are not of type BlockError to BlockUnknownError

### Checklist 📋

#### For code changes:
- [x] I have clearly listed my changes in the PR description
- [x] I have made a test plan
- [x] I have tested my changes according to the test plan:
  <!-- Put your test plan here: -->
  - [x] checked tests still work
  - [x] Ensured block error message is readable and useful

autogpt_platform/backend/backend/data/block.py

@@ -29,6 +29,13 @@ from backend.data.model import NodeExecutionStats
 from backend.integrations.providers import ProviderName
 from backend.util import json
 from backend.util.cache import cached
+from backend.util.exceptions import (
+    BlockError,
+    BlockExecutionError,
+    BlockInputError,
+    BlockOutputError,
+    BlockUnknownError,
+)
 from backend.util.settings import Config
 
 from .model import (
@@ -542,9 +549,25 @@ class Block(ABC, Generic[BlockSchemaInputType, BlockSchemaOutputType]):
         )
 
     async def execute(self, input_data: BlockInput, **kwargs) -> BlockOutput:
+        try:
+            async for output_name, output_data in self._execute(input_data, **kwargs):
+                yield output_name, output_data
+        except Exception as ex:
+            if not isinstance(ex, BlockError):
+                raise BlockUnknownError(
+                    message=str(ex),
+                    block_name=self.name,
+                    block_id=self.id,
+                ) from ex
+            else:
+                raise ex
+
+    async def _execute(self, input_data: BlockInput, **kwargs) -> BlockOutput:
         if error := self.input_schema.validate_data(input_data):
-            raise ValueError(
-                f"Unable to execute block with invalid input data: {error}"
+            raise BlockInputError(
+                message=f"Unable to execute block with invalid input data: {error}",
+                block_name=self.name,
+                block_id=self.id,
             )
 
         async for output_name, output_data in self.run(
@@ -552,11 +575,17 @@ class Block(ABC, Generic[BlockSchemaInputType, BlockSchemaOutputType]):
             **kwargs,
         ):
             if output_name == "error":
-                raise RuntimeError(output_data)
+                raise BlockExecutionError(
+                    message=output_data, block_name=self.name, block_id=self.id
+                )
             if self.block_type == BlockType.STANDARD and (
                 error := self.output_schema.validate_field(output_name, output_data)
             ):
-                raise ValueError(f"Block produced an invalid output data: {error}")
+                raise BlockOutputError(
+                    message=f"Block produced an invalid output data: {error}",
+                    block_name=self.name,
+                    block_id=self.id,
+                )
             yield output_name, output_data
 
     def is_triggered_by_event_type(

autogpt_platform/backend/backend/util/exceptions.py

@@ -1,6 +1,35 @@
 from typing import Mapping
 
 
+class BlockError(Exception):
+    """An error occurred during the running of a block"""
+
+    def __init__(self, message: str, block_name: str, block_id: str) -> None:
+        super().__init__(message)
+        self.message = message
+        self.block_name = block_name
+        self.block_id = block_id
+
+    def __str__(self):
+        return f"raised by {self.block_name} with message: {self.message}. block_id: {self.block_id}"
+
+
+class BlockInputError(BlockError, ValueError):
+    """The block had incorrect inputs, resulting in an error condition"""
+
+
+class BlockOutputError(BlockError, ValueError):
+    """The block had incorrect outputs, resulting in an error condition"""
+
+
+class BlockExecutionError(BlockError, ValueError):
+    """The block failed to execute at runtime, resulting in a handled error"""
+
+
+class BlockUnknownError(BlockError):
+    """Critical unknown error with block handling"""
+
+
 class MissingConfigError(Exception):
     """The attempted operation requires configuration which is not available"""