diff --git a/haystack/core/pipeline/base.py b/haystack/core/pipeline/base.py index 0784771745c..1e2093f48cd 100644 --- a/haystack/core/pipeline/base.py +++ b/haystack/core/pipeline/base.py @@ -1264,7 +1264,11 @@ 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. @@ -1272,11 +1276,16 @@ def _fill_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 diff --git a/haystack/core/pipeline/component_checks.py b/haystack/core/pipeline/component_checks.py index d05d373338a..fb6b6645a83 100644 --- a/haystack/core/pipeline/component_checks.py +++ b/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: diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py index 227f4a66594..b97f38e184a 100644 --- a/haystack/core/pipeline/pipeline.py +++ b/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) diff --git a/releasenotes/notes/fix-resume-loop-visit-blocked-b38766a042f86c42.yaml b/releasenotes/notes/fix-resume-loop-visit-blocked-b38766a042f86c42.yaml new file mode 100644 index 00000000000..ccd53100a66 --- /dev/null +++ b/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. diff --git a/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py b/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py index 99b9f5d3091..111135cd58e 100644 --- a/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py +++ b/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"