re-raising BreakPointException from pipeline.run()

Author: davidsbatista

haystack/core/pipeline/pipeline.py

@@ -10,7 +10,7 @@
 
 from haystack import logging, tracing
 from haystack.core.component import Component
-from haystack.core.errors import PipelineInvalidResumeStateError, PipelineRuntimeError
+from haystack.core.errors import BreakpointException, PipelineInvalidResumeStateError, PipelineRuntimeError
 from haystack.core.pipeline.base import (
     _COMPONENT_INPUT,
     _COMPONENT_OUTPUT,
@@ -71,8 +71,14 @@ def _run_component(
             logger.info("Running component {component_name}", component_name=component_name)
             try:
                 component_output = instance.run(**inputs)
+            except BreakpointException as error:
+                # Re-raise BreakpointException to preserve the original exception context
+                # This is important when Agent components internally use Pipeline._run_component
+                # and trigger breakpoints that need to bubble up to the main pipeline
+                raise error
             except Exception as error:
                 raise PipelineRuntimeError.from_exception(component_name, instance.__class__, error) from error
+
             component_visits[component_name] += 1
 
             if not isinstance(component_output, Mapping):

test/components/agents/test_agent_breakpoints_inside_pipeline.py

@@ -234,17 +234,6 @@ def test_chat_generator_breakpoint_in_pipeline_agent():
             assert e.state is not None
             assert "messages" in e.state
             assert e.results is not None
-        except PipelineRuntimeError as e:
-            # propagated exception to core Pipeline - assure that the cause is a PipelineBreakpointException
-            if hasattr(e, "__cause__") and isinstance(e.__cause__, BreakpointException):
-                original_exception = e.__cause__
-                assert original_exception.component == "chat_generator"
-                assert original_exception.state is not None
-                assert "messages" in original_exception.state
-                assert original_exception.results is not None
-            else:
-                # re-raise if it's a different PipelineRuntimeError - test failed
-                raise
 
         # verify that debug/state file was created
         chat_generator_state_files = list(Path(debug_path).glob("database_agent_chat_generator_*.json"))
@@ -269,17 +258,6 @@ def test_tool_breakpoint_in_pipeline_agent():
             assert e.state is not None
             assert "messages" in e.state
             assert e.results is not None
-        except PipelineRuntimeError as e:
-            # propagated exception to core Pipeline - assure that the cause is a PipelineBreakpointException
-            if hasattr(e, "__cause__") and isinstance(e.__cause__, BreakpointException):
-                original_exception = e.__cause__
-                assert original_exception.component == "tool_invoker"
-                assert original_exception.state is not None
-                assert "messages" in original_exception.state
-                assert original_exception.results is not None
-            else:
-                # re-raise if it's a different PipelineRuntimeError - test failed
-                raise
 
         # verify that debug/state file was created
         tool_invoker_state_files = list(Path(debug_path).glob("database_agent_tool_invoker_*.json"))
@@ -306,16 +284,6 @@ def test_agent_breakpoint_chat_generator_and_resume_pipeline():
             assert "messages" in e.state
             assert e.results is not None
 
-        except PipelineRuntimeError as e:
-            if hasattr(e, "__cause__") and isinstance(e.__cause__, BreakpointException):
-                original_exception = e.__cause__
-                assert original_exception.component == "chat_generator"
-                assert original_exception.state is not None
-                assert "messages" in original_exception.state
-                assert original_exception.results is not None
-            else:
-                raise
-
         # verify that the state file was created
         chat_generator_state_files = list(Path(debug_path).glob("database_agent_chat_generator_*.json"))
         assert len(chat_generator_state_files) > 0, f"No chat_generator state files found in {debug_path}"
@@ -368,16 +336,6 @@ def test_agent_breakpoint_tool_and_resume_pipeline():
             assert "messages" in e.state
             assert e.results is not None
 
-        except PipelineRuntimeError as e:
-            if hasattr(e, "__cause__") and isinstance(e.__cause__, BreakpointException):
-                original_exception = e.__cause__
-                assert original_exception.component == "tool_invoker"
-                assert original_exception.state is not None
-                assert "messages" in original_exception.state
-                assert original_exception.results is not None
-            else:
-                raise
-
         # verify that the state file was created
         tool_invoker_state_files = list(Path(debug_path).glob("database_agent_tool_invoker_*.json"))
         assert len(tool_invoker_state_files) > 0, f"No tool_invoker state files found in {debug_path}"