feat: Update BreakpointException to include the pipeline snapshot and where it is saved (#9888)

* Add snapshot and the full file path it was saved to breakpoint exception

* Add checks to test and simplify test

* Make both tests unit tests

* remove unused import

* update reno

* Add docstrings

* Updated reno

* Simplify test

* Uupdate test

* Fix mypy

Author: sjrl

haystack/core/errors.py

@@ -18,10 +18,13 @@ def __init__(
         component_type: Optional[type],
         message: str,
         pipeline_snapshot: Optional[PipelineSnapshot] = None,
+        *,
+        pipeline_snapshot_file_path: Optional[str] = None,
     ) -> None:
         self.component_name = component_name
         self.component_type = component_type
         self.pipeline_snapshot = pipeline_snapshot
+        self.pipeline_snapshot_file_path = pipeline_snapshot_file_path
         super().__init__(message)
 
     @classmethod
@@ -109,13 +112,43 @@ def __init__(
         self,
         message: str,
         component: Optional[str] = None,
-        inputs: Optional[dict[str, Any]] = None,
-        results: Optional[dict[str, Any]] = None,
+        pipeline_snapshot: Optional[PipelineSnapshot] = None,
+        pipeline_snapshot_file_path: Optional[str] = None,
     ):
         super().__init__(message)
         self.component = component
-        self.inputs = inputs
-        self.results = results
+        self.pipeline_snapshot = pipeline_snapshot
+        self.pipeline_snapshot_file_path = pipeline_snapshot_file_path
+
+    @property
+    def inputs(self):
+        """
+        Returns the inputs of the pipeline or agent at the breakpoint.
+
+        If an AgentBreakpoint caused this exception, returns the inputs of the agent's internal components.
+        Otherwise, returns the current inputs of the pipeline.
+        """
+        if not self.pipeline_snapshot:
+            return None
+
+        if self.pipeline_snapshot.agent_snapshot:
+            return self.pipeline_snapshot.agent_snapshot.component_inputs
+        return self.pipeline_snapshot.pipeline_state.inputs
+
+    @property
+    def results(self):
+        """
+        Returns the results of the pipeline or agent at the breakpoint.
+
+        If an AgentBreakpoint caused this exception, returns the current results of the agent.
+        Otherwise, returns the current outputs of the pipeline.
+        """
+        if not self.pipeline_snapshot:
+            return None
+
+        if self.pipeline_snapshot.agent_snapshot:
+            return self.pipeline_snapshot.agent_snapshot.component_inputs["tool_invoker"]["serialized_data"]["state"]
+        return self.pipeline_snapshot.pipeline_state.pipeline_outputs
 
 
 class PipelineInvalidPipelineSnapshotError(Exception):

haystack/core/pipeline/breakpoint.py

@@ -144,7 +144,7 @@ def load_pipeline_snapshot(file_path: Union[str, Path]) -> PipelineSnapshot:
     return pipeline_snapshot
 
 
-def _save_pipeline_snapshot(pipeline_snapshot: PipelineSnapshot, raise_on_failure: bool = True) -> None:
+def _save_pipeline_snapshot(pipeline_snapshot: PipelineSnapshot, raise_on_failure: bool = True) -> Optional[str]:
     """
     Save the pipeline snapshot dictionary to a JSON file.
 
@@ -158,6 +158,8 @@ def _save_pipeline_snapshot(pipeline_snapshot: PipelineSnapshot, raise_on_failur
     :param pipeline_snapshot: The pipeline snapshot to save.
     :param raise_on_failure: If True, raises an exception if saving fails. If False, logs the error and returns.
 
+    :returns:
+        The full path to the saved JSON file, or None if `snapshot_file_path` is None.
     :raises:
         Exception: If saving the JSON snapshot fails.
     """
@@ -169,7 +171,7 @@ def _save_pipeline_snapshot(pipeline_snapshot: PipelineSnapshot, raise_on_failur
     )
 
     if snapshot_file_path is None:
-        return
+        return None
 
     dt = pipeline_snapshot.timestamp or datetime.now()
     snapshot_dir = Path(snapshot_file_path)
@@ -201,6 +203,8 @@ def _save_pipeline_snapshot(pipeline_snapshot: PipelineSnapshot, raise_on_failur
         if raise_on_failure:
             raise
 
+    return str(full_path)
+
 
 def _create_pipeline_snapshot(
     *,
@@ -304,9 +308,9 @@ def _trigger_break_point(*, pipeline_snapshot: PipelineSnapshot) -> None:
     Trigger a breakpoint by saving a snapshot and raising exception.
 
     :param pipeline_snapshot: The current pipeline snapshot containing the state and break point
-    :raises PipelineBreakpointException: When breakpoint is triggered
+    :raises BreakpointException: When breakpoint is triggered
     """
-    _save_pipeline_snapshot(pipeline_snapshot=pipeline_snapshot)
+    full_file_path = _save_pipeline_snapshot(pipeline_snapshot=pipeline_snapshot)
 
     if isinstance(pipeline_snapshot.break_point, Breakpoint):
         component_name = pipeline_snapshot.break_point.component_name
@@ -318,8 +322,8 @@ def _trigger_break_point(*, pipeline_snapshot: PipelineSnapshot) -> None:
     raise BreakpointException(
         message=msg,
         component=component_name,
-        inputs=pipeline_snapshot.pipeline_state.inputs,
-        results=pipeline_snapshot.pipeline_state.pipeline_outputs,
+        pipeline_snapshot=pipeline_snapshot,
+        pipeline_snapshot_file_path=full_file_path,
     )
 
 
@@ -502,17 +506,16 @@ def _trigger_chat_generator_breakpoint(*, pipeline_snapshot: PipelineSnapshot) -
         raise ValueError("PipelineSnapshot must contain an AgentSnapshot to trigger a chat generator breakpoint.")
 
     break_point = pipeline_snapshot.break_point.break_point
-    _save_pipeline_snapshot(pipeline_snapshot=pipeline_snapshot)
+    full_file_path = _save_pipeline_snapshot(pipeline_snapshot=pipeline_snapshot)
     msg = (
         f"Breaking at {break_point.component_name} visit count "
         f"{pipeline_snapshot.agent_snapshot.component_visits[break_point.component_name]}"
     )
-    logger.info(msg)
     raise BreakpointException(
         message=msg,
         component=break_point.component_name,
-        inputs=pipeline_snapshot.agent_snapshot.component_inputs,
-        results=pipeline_snapshot.agent_snapshot.component_inputs["tool_invoker"]["serialized_data"]["state"],
+        pipeline_snapshot=pipeline_snapshot,
+        pipeline_snapshot_file_path=full_file_path,
     )
 
 
@@ -546,7 +549,7 @@ def _trigger_tool_invoker_breakpoint(*, llm_messages: list[ChatMessage], pipelin
     if not should_break:
         return  # No breakpoint triggered
 
-    _save_pipeline_snapshot(pipeline_snapshot=pipeline_snapshot)
+    full_file_path = _save_pipeline_snapshot(pipeline_snapshot=pipeline_snapshot)
 
     msg = (
         f"Breaking at {tool_breakpoint.component_name} visit count "
@@ -559,6 +562,6 @@ def _trigger_tool_invoker_breakpoint(*, llm_messages: list[ChatMessage], pipelin
     raise BreakpointException(
         message=msg,
         component=tool_breakpoint.component_name,
-        inputs=pipeline_snapshot.agent_snapshot.component_inputs,
-        results=pipeline_snapshot.agent_snapshot.component_inputs["tool_invoker"]["serialized_data"]["state"],
+        pipeline_snapshot=pipeline_snapshot,
+        pipeline_snapshot_file_path=full_file_path,
     )

haystack/core/pipeline/pipeline.py

@@ -390,8 +390,6 @@ def run(  # noqa: PLR0915, PLR0912, C901, pylint: disable=too-many-branches
                         parent_span=span,
                     )
                 except PipelineRuntimeError as error:
-                    # TODO Wrap creation of the pipeline snapshot with try-except in case it fails
-                    #      (e.g. serialization issue)
                     out_dir = _get_output_dir("pipeline_snapshot")
                     break_point = Breakpoint(
                         component_name=component_name,
@@ -420,7 +418,10 @@ def run(  # noqa: PLR0915, PLR0912, C901, pylint: disable=too-many-branches
 
                     # Attach the pipeline snapshot to the error before re-raising
                     error.pipeline_snapshot = pipeline_snapshot
-                    _save_pipeline_snapshot(pipeline_snapshot=pipeline_snapshot, raise_on_failure=False)
+                    full_file_path = _save_pipeline_snapshot(
+                        pipeline_snapshot=pipeline_snapshot, raise_on_failure=False
+                    )
+                    error.pipeline_snapshot_file_path = full_file_path
                     raise error
 
                 # Updates global input state with component outputs and returns outputs that should go to

releasenotes/notes/add-pipeline-snapshot-and-file-path-breakpoint-5b57687210a6516e.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    - Added `pipeline_snapshot` and `pipeline_snapshot_file_path` parameters to `BreakpointException` to provide more context when a pipeline breakpoint is triggered.
+    - Added `pipeline_snapshot_file_path` parameter to `PipelineRuntimeError` to include a reference to the stored pipeline snapshot so it can be easily found.