From 30ee86538f3d4a05ddbdb740acd6da080d6e87c5 Mon Sep 17 00:00:00 2001 From: Aditya Raut Date: Tue, 28 Jul 2026 14:47:39 +0530 Subject: [PATCH 1/3] fix: re-raise BreakpointException and PipelineRuntimeError in _run_component_async The async path had a bare 'except Exception' that wrapped both BreakpointException and PipelineRuntimeError into a new PipelineRuntimeError. This silently broke breakpoints in run_async / run_async_generator / stream and caused nested pipeline errors (e.g. from Agent) to lose their original context. Mirror the sync _run_component behaviour: re-raise both exception types as-is and only wrap unknown exceptions. Includes two regression tests and a release note. --- haystack/core/pipeline/pipeline.py | 10 +++ ...breakpoint-exception-980c389b99a94039.yaml | 10 +++ test/core/pipeline/test_async_pipeline.py | 73 ++++++++++++++++++- 3 files changed, 92 insertions(+), 1 deletion(-) create mode 100644 releasenotes/notes/fix-async-breakpoint-exception-980c389b99a94039.yaml diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py index 227f4a66594..7d760388af0 100644 --- a/haystack/core/pipeline/pipeline.py +++ b/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 diff --git a/releasenotes/notes/fix-async-breakpoint-exception-980c389b99a94039.yaml b/releasenotes/notes/fix-async-breakpoint-exception-980c389b99a94039.yaml new file mode 100644 index 00000000000..5d1bb4efb65 --- /dev/null +++ b/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. diff --git a/test/core/pipeline/test_async_pipeline.py b/test/core/pipeline/test_async_pipeline.py index f5bccdff004..fe1ad699f52 100644 --- a/test/core/pipeline/test_async_pipeline.py +++ b/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,73 @@ 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 bare ``except Exception`` in _run_component_async would catch the + BreakpointException raised at the start of that method (before _execute_component_async + is even called) and wrap it in a PipelineRuntimeError, silently breaking the breakpoint + feature for the async 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 From 75ef746c1280f8e16749ad908cd7ff36d7d4db3e Mon Sep 17 00:00:00 2001 From: Aditya Raut Date: Tue, 28 Jul 2026 15:03:53 +0530 Subject: [PATCH 2/3] docs: clarify docstring for test_run_component_async_propagates_breakpoint_exception --- test/core/pipeline/test_async_pipeline.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/test/core/pipeline/test_async_pipeline.py b/test/core/pipeline/test_async_pipeline.py index fe1ad699f52..f9b6df32ea7 100644 --- a/test/core/pipeline/test_async_pipeline.py +++ b/test/core/pipeline/test_async_pipeline.py @@ -558,10 +558,10 @@ async def test_run_component_async_propagates_breakpoint_exception(): """ Regression test: _run_component_async must re-raise BreakpointException as-is. - Before the fix, a bare ``except Exception`` in _run_component_async would catch the - BreakpointException raised at the start of that method (before _execute_component_async - is even called) and wrap it in a PipelineRuntimeError, silently breaking the breakpoint - feature for the async path. + 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 From 92c16c3942a6897e7e793fe4e0681d396538a62f Mon Sep 17 00:00:00 2001 From: Aditya Raut Date: Tue, 28 Jul 2026 15:10:04 +0530 Subject: [PATCH 3/3] style: format test_async_pipeline.py with ruff --- test/core/pipeline/test_async_pipeline.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/test/core/pipeline/test_async_pipeline.py b/test/core/pipeline/test_async_pipeline.py index f9b6df32ea7..85635d807af 100644 --- a/test/core/pipeline/test_async_pipeline.py +++ b/test/core/pipeline/test_async_pipeline.py @@ -602,9 +602,7 @@ async def test_run_component_async_does_not_rewrap_pipeline_runtime_error(): """ original_error = PipelineRuntimeError( - component_name="inner_comp", - component_type=type(None), - message="original detailed message", + component_name="inner_comp", component_type=type(None), message="original detailed message" ) @component