fix(pipeline): allow resuming a snapshot paused on a later loop visit

haystack/core/pipeline/base.py

@@ -1264,19 +1264,28 @@ def _consume_component_inputs(
         return consumed_inputs
 
     def _fill_queue(
-        self, component_names: list[str], inputs: InputsType, component_visits: dict[str, int]
+        self,
+        component_names: list[str],
+        inputs: InputsType,
+        component_visits: dict[str, int],
+        resume_component_name: str | None = None,
     ) -> FIFOPriorityQueue:
         """
         Calculates the execution priority for each component and inserts it into the priority queue.
 
         :param component_names: Names of the components to put into the queue.
         :param inputs: Inputs to the components.
         :param component_visits: Current state of component visits.
+        :param resume_component_name: When resuming from a snapshot, the name of the component the run
+            resumes at. Its restored inputs look like user inputs, so it is flagged as a resume target
+            to let it trigger once even if it was paused on a later visit inside a loop.
         :returns: A prioritized queue of component names.
         """
         priority_queue = FIFOPriorityQueue()
         for component_name in component_names:
             comp = self._get_component_with_graph_metadata_and_visits(component_name, component_visits[component_name])
+            if component_name == resume_component_name:
+                comp["is_resume"] = True
             priority = self._calculate_priority(comp, inputs.get(component_name, {}))
             priority_queue.push(component_name, priority)
         return priority_queue

haystack/core/pipeline/component_checks.py

@@ -45,8 +45,14 @@ def has_any_trigger(component: dict, inputs: dict) -> bool:
     trigger_from_predecessor = any_predecessors_provided_input(component, inputs)
     trigger_from_user = has_user_input(inputs) and component["visits"] == 0
     trigger_without_inputs = can_not_receive_inputs_from_pipeline(component) and component["visits"] == 0
-
-    return trigger_from_predecessor or trigger_from_user or trigger_without_inputs
+    # When resuming from a pipeline snapshot, the paused component's inputs are restored as user
+    # inputs (the sender information is not preserved through serialization). If the component was
+    # paused on a second-or-later visit inside a loop, ``visits`` is already > 0, so neither the
+    # user nor the predecessor trigger fires and the resume would be wrongly reported as blocked.
+    # The resume is an explicit, one-time trigger for exactly that component.
+    trigger_from_resume = component.get("is_resume", False) and has_user_input(inputs)
+
+    return trigger_from_predecessor or trigger_from_user or trigger_without_inputs or trigger_from_resume
 
 
 def are_all_sockets_ready(component: dict, inputs: dict, only_check_mandatory: bool = False) -> bool:

haystack/core/pipeline/pipeline.py

@@ -392,7 +392,10 @@ def run(  # noqa: PLR0915, PLR0912, C901
             },
         ) as span:
             inputs = self._convert_to_internal_format(pipeline_inputs=data)
-            priority_queue = self._fill_queue(ordered_component_names, inputs, component_visits)
+            resume_component_name = pipeline_snapshot.break_point.component_name if pipeline_snapshot else None
+            priority_queue = self._fill_queue(
+                ordered_component_names, inputs, component_visits, resume_component_name=resume_component_name
+            )
 
             # check if pipeline is blocked before execution
             self.validate_pipeline(priority_queue)

releasenotes/notes/fix-resume-loop-visit-blocked-b38766a042f86c42.yaml

@@ -0,0 +1,8 @@
+---
+fixes:
+  - |
+    Fixed resuming a pipeline from a snapshot when the breakpoint was hit on the second (or a later)
+    visit of a component inside a loop. Previously the resumed run immediately failed with
+    ``PipelineComponentsBlockedError`` because the paused component's restored inputs look like user
+    inputs, which only trigger a component on its first visit. The component the run resumes at is now
+    allowed to trigger once regardless of its visit count.

test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py

@@ -134,3 +134,52 @@ def test_pipeline_breakpoints_validation_loop(
         assert cities_data.cities[0].name == "Berlin"
         assert cities_data.cities[1].name == "Paris"
         assert cities_data.cities[2].name == "Lisbon"
+
+
+@component
+class _LoopOrDone:
+    """Emits ``retry`` until it receives a non-``start`` value, then emits ``done``."""
+
+    @component.output_types(retry=str, done=str)
+    def run(self, value: str) -> dict[str, str]:
+        if value == "start":
+            return {"retry": "start-retry"}
+        return {"done": value}
+
+
+def _build_loop_pipeline() -> Pipeline:
+    from haystack.components.joiners import BranchJoiner
+
+    pipe = Pipeline(max_runs_per_component=5)
+    pipe.add_component("joiner", BranchJoiner(str))
+    pipe.add_component("loop", _LoopOrDone())
+    pipe.connect("joiner.value", "loop.value")
+    pipe.connect("loop.retry", "joiner.value")
+    return pipe
+
+
+@pytest.mark.integration
+def test_resume_snapshot_taken_on_a_later_loop_visit():
+    """
+    Resuming from a snapshot that was taken when the breakpoint fired on a second-or-later visit of a
+    component inside a loop must continue the run instead of failing with PipelineComponentsBlockedError.
+
+    Regression test for https://github.com/deepset-ai/haystack/issues/12145
+    """
+    pipe = _build_loop_pipeline()
+
+    # visit_count=1 => the breakpoint fires the second time the joiner is about to run (inside the loop).
+    try:
+        pipe.run({"joiner": {"value": "start"}}, break_point=Breakpoint(component_name="joiner", visit_count=1))
+    except BreakpointException as exc:
+        snapshot = exc.pipeline_snapshot
+    else:
+        pytest.fail("Expected the breakpoint to trigger")
+
+    assert snapshot is not None
+    # The snapshot was taken when the joiner had already been visited once.
+    assert snapshot.pipeline_state.component_visits["joiner"] == 1
+
+    # Resuming must not raise PipelineComponentsBlockedError and must finish the loop.
+    result = _build_loop_pipeline().run(data={}, pipeline_snapshot=snapshot)
+    assert result["loop"]["done"] == "start-retry"