Bug/dg 314 show nicer errors webrtc

inference/core/interfaces/http/error_handlers.py

@@ -54,6 +54,7 @@
 )
 from inference.core.workflows.errors import (
     ClientCausedStepExecutionError,
+    DynamicBlockCodeError,
     DynamicBlockError,
     ExecutionGraphStructureError,
     InvalidReferenceTargetError,
@@ -88,6 +89,41 @@
 )
 
 
+def _build_execution_error_response(
+    error: "DynamicBlockCodeError | StepExecutionError",
+) -> "WorkflowErrorResponse":
+    """Build a WorkflowErrorResponse for execution errors."""
+    if isinstance(error, DynamicBlockCodeError):
+        block_id = error.block_type_name or "Dynamic Block"
+        block_type = error.block_type_name
+        property_name = "Python code"
+        property_details = error.public_message
+    elif isinstance(error, StepExecutionError):
+        block_id = error.block_id
+        block_type = error.block_type
+        property_name = None
+        property_details = str(error.inner_error)
+    else:
+        raise ValueError(f"Unsupported error type: {type(error)}")
+
+    return WorkflowErrorResponse(
+        message=error.public_message,
+        error_type=error.__class__.__name__,
+        context=error.context,
+        inner_error_type=error.inner_error_type,
+        inner_error_message=str(error.inner_error) if error.inner_error else None,
+        blocks_errors=[
+            WorkflowBlockError(
+                block_id=block_id,
+                block_type=block_type,
+                property_name=property_name,
+                property_details=property_details,
+                block_traceback=error.block_traceback,
+            ),
+        ],
+    )
+
+
 def with_route_exceptions(route):
     """
     A decorator that wraps a FastAPI route to handle specific exceptions. If an exception
@@ -180,6 +216,10 @@ def wrapped_route(*args, **kwargs):
                 blocks_errors=error.blocks_errors,
             )
             resp = JSONResponse(status_code=400, content=content.model_dump())
+        except DynamicBlockCodeError as error:
+            logger.exception("%s: %s", type(error).__name__, error)
+            content = _build_execution_error_response(error)
+            resp = JSONResponse(status_code=400, content=content.model_dump())
         except (
             WorkflowDefinitionError,
             ReferenceTypeError,
@@ -450,19 +490,7 @@ def wrapped_route(*args, **kwargs):
             )
         except StepExecutionError as error:
             logger.exception("%s: %s", type(error).__name__, error)
-            content = WorkflowErrorResponse(
-                message=str(error.public_message),
-                error_type=error.__class__.__name__,
-                context=str(error.context),
-                inner_error_type=str(error.inner_error_type),
-                inner_error_message=str(error.inner_error),
-                blocks_errors=[
-                    WorkflowBlockError(
-                        block_id=error.block_id,
-                        block_type=error.block_type,
-                    ),
-                ],
-            )
+            content = _build_execution_error_response(error)
             resp = JSONResponse(
                 status_code=500,
                 content=content.model_dump(),
@@ -619,6 +647,10 @@ async def wrapped_route(*args, **kwargs):
                 blocks_errors=error.blocks_errors,
             )
             resp = JSONResponse(status_code=400, content=content.model_dump())
+        except DynamicBlockCodeError as error:
+            logger.exception("%s: %s", type(error).__name__, error)
+            content = _build_execution_error_response(error)
+            resp = JSONResponse(status_code=400, content=content.model_dump())
         except (
             WorkflowDefinitionError,
             ReferenceTypeError,
@@ -889,19 +921,7 @@ async def wrapped_route(*args, **kwargs):
             )
         except StepExecutionError as error:
             logger.exception("%s: %s", type(error).__name__, error)
-            content = WorkflowErrorResponse(
-                message=str(error.public_message),
-                error_type=error.__class__.__name__,
-                context=str(error.context),
-                inner_error_type=str(error.inner_error_type),
-                inner_error_message=str(error.inner_error),
-                blocks_errors=[
-                    WorkflowBlockError(
-                        block_id=error.block_id,
-                        block_type=error.block_type,
-                    ),
-                ],
-            )
+            content = _build_execution_error_response(error)
             resp = JSONResponse(
                 status_code=500,
                 content=content.model_dump(),

inference/core/interfaces/http/http_api.py

@@ -279,7 +279,11 @@
 from inference.core.utils.container import is_docker_socket_mounted
 from inference.core.utils.notebooks import start_notebook
 from inference.core.workflows.core_steps.common.entities import StepExecutionMode
-from inference.core.workflows.errors import WorkflowError, WorkflowSyntaxError
+from inference.core.workflows.errors import (
+    WorkflowBlockError,
+    WorkflowError,
+    WorkflowSyntaxError,
+)
 from inference.core.workflows.execution_engine.core import (
     ExecutionEngine,
     get_available_versions,
@@ -1800,10 +1804,30 @@ async def initialise_webrtc_worker(
                 )
                 if worker_result.exception_type is not None:
                     if worker_result.exception_type == "WorkflowSyntaxError":
+                        # Reconstruct exception from serialized worker result.
+                        # We dynamically create an exception class to preserve
+                        # the original type name (e.g., "ValidationError") for
+                        # the inner_error_type property, since exceptions can't
+                        # be pickled across the worker process boundary.
+                        inner_error = None
+                        if worker_result.inner_error and worker_result.inner_error_type:
+                            inner_error = type(
+                                worker_result.inner_error_type,
+                                (Exception,),
+                                {},
+                            )(worker_result.inner_error)
+
+                        blocks_errors = None
+                        if worker_result.blocks_errors:
+                            blocks_errors = [
+                                WorkflowBlockError(**be)
+                                for be in worker_result.blocks_errors
+                            ]
                         raise WorkflowSyntaxError(
                             public_message=worker_result.error_message,
                             context=worker_result.error_context,
-                            inner_error=worker_result.inner_error,
+                            inner_error=inner_error,
+                            blocks_errors=blocks_errors,
                         )
                     if worker_result.exception_type == "WorkflowError":
                         raise WorkflowError(

inference/core/interfaces/webrtc_worker/entities.py

@@ -87,6 +87,8 @@ class WebRTCWorkerResult(BaseModel):
     error_message: Optional[str] = None
     error_context: Optional[str] = None
     inner_error: Optional[str] = None
+    inner_error_type: Optional[str] = None
+    blocks_errors: Optional[List[Dict[str, Any]]] = None
 
 
 class WebRTCSessionHeartbeatRequest(BaseModel):

inference/core/interfaces/webrtc_worker/webrtc.py

@@ -986,12 +986,19 @@ async def init_rtc_peer_connection_with_loop(
     except WorkflowSyntaxError as error:
         if heartbeat_callback:
             heartbeat_callback()
+        blocks_errors_serialized = None
+        if error.blocks_errors:
+            blocks_errors_serialized = [
+                block_error.model_dump() for block_error in error.blocks_errors
+            ]
         send_answer(
             WebRTCWorkerResult(
                 exception_type=WorkflowSyntaxError.__name__,
-                error_message=str(error),
+                error_message=error.public_message,
                 error_context=str(error.context),
-                inner_error=str(error.inner_error),
+                inner_error=str(error.inner_error) if error.inner_error else None,
+                inner_error_type=error.inner_error_type,
+                blocks_errors=blocks_errors_serialized,
             )
         )
         return

inference/core/workflows/errors.py

@@ -3,12 +3,21 @@
 from pydantic import BaseModel, Field
 
 
+class BlockTraceback(BaseModel):
+    traceback: Optional[str] = None
+    error_line: Optional[int] = None
+    code_snippet: Optional[str] = None
+    stdout: Optional[str] = None
+    stderr: Optional[str] = None
+
+
 class WorkflowBlockError(BaseModel):
     block_id: Optional[str] = None
     block_type: Optional[str] = None
     block_details: Optional[str] = None
     property_name: Optional[str] = None
     property_details: Optional[str] = None
+    block_traceback: Optional[BlockTraceback] = None
 
 
 class WorkflowError(Exception):
@@ -152,6 +161,45 @@ class WorkflowExecutionEngineError(WorkflowError):
     pass
 
 
+class DynamicBlockCodeError(WorkflowExecutionEngineError):
+    """Exception for dynamic block code execution errors (errors provoked by user's code)."""
+
+    def __init__(
+        self,
+        public_message: str,
+        context: str = "dynamic_block_code_execution",
+        inner_error: Optional[Exception] = None,
+        block_type_name: Optional[str] = None,
+        error_line: Optional[int] = None,
+        code_snippet: Optional[str] = None,
+        traceback_str: Optional[str] = None,
+        stdout: Optional[str] = None,
+        stderr: Optional[str] = None,
+    ):
+        super().__init__(
+            public_message=public_message, context=context, inner_error=inner_error
+        )
+        self.block_type_name = block_type_name
+        self.error_line = error_line
+        self.code_snippet = code_snippet
+        self.traceback_str = traceback_str
+        self.stdout = stdout
+        self.stderr = stderr
+
+    @property
+    def block_traceback(self) -> Optional[BlockTraceback]:
+        """Construct BlockTraceback from error fields if any are present."""
+        if not any([self.error_line, self.traceback_str, self.stdout, self.stderr]):
+            return None
+        return BlockTraceback(
+            error_line=self.error_line,
+            code_snippet=self.code_snippet,
+            traceback=self.traceback_str,
+            stdout=self.stdout,
+            stderr=self.stderr,
+        )
+
+
 class NotSupportedExecutionEngineError(WorkflowExecutionEngineError):
     pass
 
@@ -165,12 +213,14 @@ def __init__(
         self,
         block_id: str,
         block_type: str,
+        block_traceback: Optional[BlockTraceback] = None,
         *args,
         **kwargs,
     ):
         super().__init__(*args, **kwargs)
         self.block_id = block_id
         self.block_type = block_type
+        self.block_traceback = block_traceback
 
 
 class ClientCausedStepExecutionError(WorkflowExecutionEngineError):

inference/core/workflows/execution_engine/v1/dynamic_blocks/block_scaffolding.py

@@ -1,6 +1,5 @@
-import traceback
 import types
-from typing import List, Optional, Type
+from typing import Any, Dict, List, Optional, Type
 
 from inference.core.env import (
     ALLOW_CUSTOM_PYTHON_EXECUTION_IN_WORKFLOWS,
@@ -10,6 +9,7 @@
 from inference.core.exceptions import WorkspaceLoadError
 from inference.core.roboflow_api import get_roboflow_workspace
 from inference.core.workflows.errors import (
+    DynamicBlockCodeError,
     DynamicBlockError,
     WorkflowEnvironmentConfigurationError,
 )
@@ -40,6 +40,12 @@
 # Shared globals dict for all custom python blocks in local mode
 _LOCAL_SHARED_GLOBALS = {}
 
+from inference.core.workflows.execution_engine.v1.dynamic_blocks.error_utils import (
+    capture_output,
+    create_dynamic_block_code_error,
+    extract_code_snippet,
+)
+
 
 def assembly_custom_python_block(
     block_type_name: str,
@@ -49,13 +55,15 @@ def assembly_custom_python_block(
     api_key: Optional[str] = None,
     skip_class_eval: Optional[bool] = False,
 ) -> Type[WorkflowBlock]:
+
     code_module = create_dynamic_module(
         block_type_name=block_type_name,
         python_code=python_code,
         module_name=f"dynamic_module_{unique_identifier}",
         api_key=api_key,
         skip_class_eval=skip_class_eval,
     )
+
     if not hasattr(code_module, python_code.run_function_name):
         raise DynamicBlockError(
             public_message=f"Cannot find function: {python_code.run_function_name} in declared code for "
@@ -97,20 +105,19 @@ def run(self, *args, **kwargs) -> BlockResult:
                     "`ALLOW_CUSTOM_PYTHON_EXECUTION_IN_WORKFLOWS=True`",
                     context="workflow_execution | step_execution | dynamic_step",
                 )
+            import_lines_count = len(_get_python_code_imports(python_code).splitlines())
             try:
-                return run_function(self, *args, **kwargs)
+                with capture_output() as (stdout_buf, stderr_buf):
+                    return run_function(self, *args, **kwargs)
             except Exception as error:
-                tb = traceback.extract_tb(error.__traceback__)
-                if tb:
-                    frame = tb[-1]
-                    line_number = frame.lineno - len(
-                        _get_python_code_imports(python_code).splitlines()
-                    )
-                    function_name = frame.name
-                    message = f"Error in line {line_number}, in {function_name}: {error.__class__.__name__}: {error}"
-                else:
-                    message = f"{error.__class__.__name__}: {error}"
-                raise Exception(message) from error
+                raise create_dynamic_block_code_error(
+                    error=error,
+                    user_code=python_code.run_function_code or "",
+                    import_lines_count=import_lines_count,
+                    stdout=stdout_buf.getvalue() or None,
+                    stderr=stderr_buf.getvalue() or None,
+                    block_type_name=block_type_name,
+                ) from error
 
     if python_code.init_function_code is not None and not hasattr(
         code_module, python_code.init_function_name
@@ -213,9 +220,23 @@ def create_dynamic_module(
             exec(code, dynamic_module.__dict__)
             return dynamic_module
         except Exception as error:
-            raise DynamicBlockError(
-                public_message=f"Error of type `{error.__class__.__name__}` encountered while attempting to "
-                f"create Python module with code for block: {block_type_name}. Error message: {error}. Full code:\n{code}",
-                context="workflow_compilation | dynamic_block_compilation | dynamic_module_creation",
+            error_line = getattr(error, "lineno", None)
+            code_snippet = None
+            if error_line and python_code.run_function_code:
+                import_lines_offset = len(
+                    _get_python_code_imports(python_code).splitlines()
+                )
+                error_line -= import_lines_offset
+                snippet = extract_code_snippet(
+                    python_code.run_function_code, error_line
+                )
+                code_snippet = snippet.lstrip("\n") if snippet else None
+
+            raise DynamicBlockCodeError(
+                public_message=f"{error.__class__.__name__}: {error}",
+                context="dynamic_block_code_compilation",
                 inner_error=error,
+                block_type_name=block_type_name,
+                error_line=error_line,
+                code_snippet=code_snippet,
             ) from error