fix: preserve BreakpointException and PipelineRuntimeError in _run_component_async

haystack/core/pipeline/pipeline.py

@@ -566,6 +566,16 @@ async def _run_component_async(
                 # For sync-only components, _run_component_async dispatches to a thread via asyncio.to_thread,
                 # which copies the current contextvars context — preserving e.g. the active tracing span.
                 outputs = await _execute_component_async(instance, **component_inputs_copy)
+            except BreakpointException:
+                # 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
+            except PipelineRuntimeError:
+                # Any components that internally use Pipeline._run_component could raise a PipelineRuntimeError
+                # with additional context (e.g. Agent raises an agent snapshot) so we re-raise here instead of
+                # wrapping it in another PipelineRuntimeError.
+                raise
             except Exception as error:
                 raise PipelineRuntimeError.from_exception(component_name, instance.__class__, error) from error
 

releasenotes/notes/fix-async-breakpoint-exception-980c389b99a94039.yaml

@@ -0,0 +1,10 @@
+---
+fixes:
+  - |
+    Fixed ``_run_component_async`` swallowing ``BreakpointException`` and ``PipelineRuntimeError``
+    instead of re-raising them. Previously, a bare ``except Exception`` clause wrapped both
+    exception types in a new ``PipelineRuntimeError``, which silently broke breakpoints in the
+    async pipeline path (``run_async``, ``run_async_generator``, ``stream``) and caused nested
+    pipeline errors from Agent components to lose their original context. The async path now
+    mirrors the sync ``_run_component`` behavior: ``BreakpointException`` and
+    ``PipelineRuntimeError`` are re-raised as-is, and only unknown exceptions are wrapped.

test/core/pipeline/test_async_pipeline.py

@@ -11,7 +11,8 @@
 
 from haystack import Document, Pipeline, component
 from haystack.components.joiners import BranchJoiner
-from haystack.core.errors import PipelineRuntimeError
+from haystack.core.errors import BreakpointException, PipelineRuntimeError
+from haystack.dataclasses.breakpoints import Breakpoint
 
 _test_context_var: contextvars.ContextVar[str] = contextvars.ContextVar("_test_context_var", default="unset")
 
@@ -550,3 +551,71 @@ def run(self, text: str) -> dict[str, str]:
 
     with pytest.raises(PipelineRuntimeError, match="Cannot unwrap a list of 3 items"):
         await pipe.run_async({})
+
+
+@pytest.mark.asyncio
+async def test_run_component_async_propagates_breakpoint_exception():
+    """
+    Regression test: _run_component_async must re-raise BreakpointException as-is.
+
+    Before the fix, a ``BreakpointException`` raised during component execution (or by a nested pipeline)
+    was caught by the generic ``except Exception`` handler in _run_component_async and wrapped in a
+    new ``PipelineRuntimeError``, preventing callers from handling breakpoints correctly in the
+    async pipeline path.
+    """
+
+    @component
+    class SimpleComponent:
+        @component.output_types(result=str)
+        def run(self, value: str) -> dict[str, str]:
+            return {"result": f"{value}_processed"}
+
+    pipeline = Pipeline()
+    pipeline.add_component("comp", SimpleComponent())
+
+    break_point = Breakpoint(component_name="comp", visit_count=0)
+    component_visits = {"comp": 0}
+    comp = pipeline._get_component_with_graph_metadata_and_visits("comp", 0)
+
+    # _run_component_async must raise BreakpointException, not wrap it in PipelineRuntimeError.
+    with pytest.raises(BreakpointException) as exc_info:
+        await pipeline._run_component_async(
+            component_name="comp",
+            component=comp,
+            component_inputs={"value": "hello"},
+            component_visits=component_visits,
+            break_point=break_point,
+        )
+
+    # The exception must carry the break_point back to the caller.
+    assert exc_info.value.break_point.component_name == "comp"
+
+
+@pytest.mark.asyncio
+async def test_run_component_async_does_not_rewrap_pipeline_runtime_error():
+    """
+    Regression test: _run_component_async must re-raise PipelineRuntimeError without wrapping it.
+
+    Components such as Agent internally call Pipeline._run_component and may raise a
+    PipelineRuntimeError that already contains a full snapshot / context. The async path must
+    not swallow that and replace it with a freshly constructed PipelineRuntimeError.
+    """
+
+    original_error = PipelineRuntimeError(
+        component_name="inner_comp", component_type=type(None), message="original detailed message"
+    )
+
+    @component
+    class ErroringComponent:
+        @component.output_types(result=str)
+        def run(self, value: str) -> dict[str, str]:
+            raise original_error
+
+    pipeline = Pipeline()
+    pipeline.add_component("erroring", ErroringComponent())
+
+    with pytest.raises(PipelineRuntimeError) as exc_info:
+        await pipeline.run_async({"erroring": {"value": "x"}})
+
+    # Must be the exact same instance — not a freshly constructed wrapper.
+    assert exc_info.value is original_error