From b6d51a9d6adef0ba030e4f2b37d93b8341cefb1d Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 10:55:44 +0200 Subject: [PATCH 01/13] Update pipeline snpahsots to serialize the internal inputs format of pipelines. This fixes a bug when using a pipeline snapshots with pipelines that contain loops. --- haystack/core/component/sockets.py | 2 +- haystack/core/component/types.py | 28 ++- haystack/core/pipeline/breakpoint.py | 61 +++++- haystack/core/pipeline/pipeline.py | 43 ++-- haystack/dataclasses/breakpoints.py | 3 + ...shot-resume-in-loops-0ab593c343026abb.yaml | 36 ++++ test/core/pipeline/test_breakpoint.py | 204 ++++++++++++++---- 7 files changed, 316 insertions(+), 61 deletions(-) create mode 100644 releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml diff --git a/haystack/core/component/sockets.py b/haystack/core/component/sockets.py index 6a7d7c5db03..f940616e5b3 100644 --- a/haystack/core/component/sockets.py +++ b/haystack/core/component/sockets.py @@ -46,7 +46,7 @@ class Sockets: # noqa: PLW1641 # >> - documents: Any inputs.question - # >> InputSocket(name='question', type=typing.Any, default_value=, ... + # >> InputSocket(name='question', type=typing.Any, default_value=_empty, ... ``` """ diff --git a/haystack/core/component/types.py b/haystack/core/component/types.py index 84753b3191c..ecf75e74560 100644 --- a/haystack/core/component/types.py +++ b/haystack/core/component/types.py @@ -31,8 +31,32 @@ GreedyVariadic: TypeAlias = Annotated[Iterable[T], HAYSTACK_GREEDY_VARIADIC_ANNOTATION] -class _empty: - """Custom object for marking InputSocket.default_value as not set.""" +class _Empty: + """ + Type of the `_empty` sentinel, which marks an `InputSocket.default_value` as not set. + + The sentinel is also reused by the pipeline to mark a socket whose sender ran without producing a value for it. + It is compared by identity, so `from_dict` returns the singleton rather than a new instance. + """ + + def __repr__(self) -> str: + return "_empty" + + def to_dict(self) -> dict[str, Any]: + """ + Serialize the sentinel. It carries no state, so the payload is empty. + """ + return {} + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "_Empty": # noqa: ARG003 + """ + Deserialize the sentinel back to the singleton, so that identity comparisons keep working. + """ + return _empty + + +_empty = _Empty() @dataclass diff --git a/haystack/core/pipeline/breakpoint.py b/haystack/core/pipeline/breakpoint.py index 932c6190ca6..dddd7b1e801 100644 --- a/haystack/core/pipeline/breakpoint.py +++ b/haystack/core/pipeline/breakpoint.py @@ -14,6 +14,7 @@ from haystack import logging from haystack.core.errors import PipelineInvalidPipelineSnapshotError from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot, PipelineState +from haystack.utils import _deserialize_value_with_schema from haystack.utils.base_serialization import _serialize_with_field_fallback logger = logging.getLogger(__name__) @@ -25,6 +26,11 @@ # The callback receives a PipelineSnapshot and optionally returns a file path string SnapshotCallback = Callable[[PipelineSnapshot], str | None] +# Value of `PipelineState.inputs_format` for snapshots that store the pipeline's internal inputs state, where each +# input keeps the component that sent it. Snapshots without it store flattened `{component: {socket: value}}` inputs. +# See `_serialize_internal_inputs` for the exact shape. +INTERNAL_INPUTS_FORMAT = "internal" + def _is_snapshot_save_enabled() -> bool: """ @@ -211,6 +217,48 @@ def _save_pipeline_snapshot( return str(full_path) +def _serialize_internal_inputs(inputs: dict[str, Any]) -> dict[str, Any]: + """ + Serialize the pipeline's internal inputs state, keeping the `sender` of every input intact. + + The `sender` tells a resumed run which inputs came from a predecessor and which from outside the pipeline, and + therefore whether the paused component can be triggered again. + + The inputs a socket received are stored keyed by position rather than as a list, because a list gets a single + schema derived from its first item. A socket with several senders can hold values of different types, for + example a value from one sender and the `_NO_OUTPUT_PRODUCED` sentinel from another, and those need one schema + each to survive the round trip. + + :param inputs: The pipeline's internal inputs state. + :returns: A dict of the form `{"serialization_schema": ..., "serialized_data": ...}`. + """ + positional_inputs = { + component_name: { + socket_name: {str(position): entry for position, entry in enumerate(socket_inputs)} + for socket_name, socket_inputs in socket_dict.items() + } + for component_name, socket_dict in inputs.items() + } + + return _serialize_with_field_fallback(positional_inputs, description="the inputs of the current pipeline state") + + +def _deserialize_internal_inputs(serialized_inputs: dict[str, Any]) -> dict[str, Any]: + """ + Restore the pipeline's internal inputs state from a snapshot written by `_serialize_internal_inputs`. + + :param serialized_inputs: The `inputs` of a `PipelineState` whose `inputs_format` is `INTERNAL_INPUTS_FORMAT`. + :returns: The pipeline's internal inputs state. + """ + return { + component_name: { + socket_name: [socket_inputs[position] for position in sorted(socket_inputs, key=int)] + for socket_name, socket_inputs in socket_dict.items() + } + for component_name, socket_dict in _deserialize_value_with_schema(serialized_inputs).items() + } + + def _create_pipeline_snapshot( *, inputs: dict[str, Any], @@ -226,7 +274,8 @@ def _create_pipeline_snapshot( Create a snapshot of the pipeline at the point where the breakpoint was triggered. :param inputs: The current pipeline snapshot inputs. - :param component_inputs: The inputs to the component that triggered the breakpoint. + :param component_inputs: The inputs of the component that triggered the breakpoint, as they were before the + component consumed them. :param break_point: The breakpoint that triggered the snapshot. :param component_visits: The visit count of the component that triggered the breakpoint. :param original_input_data: The original input data. @@ -239,11 +288,8 @@ def _create_pipeline_snapshot( component_name = break_point.component_name transformed_original_input_data = _transform_json_structure(original_input_data) - transformed_inputs = _transform_json_structure({**inputs, component_name: component_inputs}) - serialized_inputs = _serialize_with_field_fallback( - transformed_inputs, description="the inputs of the current pipeline state" - ) + serialized_inputs = _serialize_internal_inputs({**inputs, component_name: component_inputs}) serialized_original_input_data = _serialize_with_field_fallback( transformed_original_input_data, description="original input data for `pipeline.run`" ) @@ -253,7 +299,10 @@ def _create_pipeline_snapshot( return PipelineSnapshot( pipeline_state=PipelineState( - inputs=serialized_inputs, component_visits=component_visits, pipeline_outputs=serialized_pipeline_outputs + inputs=serialized_inputs, + component_visits=component_visits, + pipeline_outputs=serialized_pipeline_outputs, + inputs_format=INTERNAL_INPUTS_FORMAT, ), timestamp=datetime.now(), break_point=break_point, diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py index 227f4a66594..a71eb255a01 100644 --- a/haystack/core/pipeline/pipeline.py +++ b/haystack/core/pipeline/pipeline.py @@ -19,8 +19,10 @@ _validate_component_output_keys, ) from haystack.core.pipeline.breakpoint import ( + INTERNAL_INPUTS_FORMAT, SnapshotCallback, _create_pipeline_snapshot, + _deserialize_internal_inputs, _save_pipeline_snapshot, _validate_break_point_against_pipeline, _validate_pipeline_snapshot_against_pipeline, @@ -347,6 +349,9 @@ def run( # noqa: PLR0915, PLR0912, C901 include_outputs_from = set() pipeline_outputs: dict[str, Any] = {} + # Set when resuming from a snapshot that predates `INTERNAL_INPUTS_FORMAT` and therefore lost the sender of + # each input. Cleared as soon as the paused component has run. + legacy_resume_component: str | None = None if not pipeline_snapshot: # normalize `data` @@ -369,7 +374,9 @@ def run( # noqa: PLR0915, PLR0912, C901 # Handle resuming the pipeline from a snapshot component_visits = pipeline_snapshot.pipeline_state.component_visits ordered_component_names = pipeline_snapshot.ordered_component_names - data = _deserialize_value_with_schema(pipeline_snapshot.pipeline_state.inputs) + data = _deserialize_value_with_schema(pipeline_snapshot.original_input_data) + if pipeline_snapshot.pipeline_state.inputs_format != INTERNAL_INPUTS_FORMAT: + legacy_resume_component = pipeline_snapshot.break_point.component_name # include_outputs_from from the snapshot when resuming include_outputs_from = pipeline_snapshot.include_outputs_from @@ -391,7 +398,17 @@ def run( # noqa: PLR0915, PLR0912, C901 "haystack.pipeline.execution_mode": "sync", }, ) as span: - inputs = self._convert_to_internal_format(pipeline_inputs=data) + if pipeline_snapshot is None: + inputs = self._convert_to_internal_format(pipeline_inputs=data) + elif legacy_resume_component is not None: + # Legacy snapshots stored flattened values, so the only thing we can do is treat them as if they + # came from outside the pipeline. + inputs = self._convert_to_internal_format( + pipeline_inputs=_deserialize_value_with_schema(pipeline_snapshot.pipeline_state.inputs) + ) + else: + inputs = _deserialize_internal_inputs(pipeline_snapshot.pipeline_state.inputs) + priority_queue = self._fill_queue(ordered_component_names, inputs, component_visits) # check if pipeline is blocked before execution @@ -431,10 +448,15 @@ def run( # noqa: PLR0915, PLR0912, C901 component_name, component_visits[component_name] ) - if pipeline_snapshot: - is_resume = pipeline_snapshot.break_point.component_name == component_name - else: - is_resume = False + is_resume = legacy_resume_component == component_name + if is_resume: + # Only the first execution of the paused component needs the legacy handling. Later visits of the + # same component inside a loop receive regular inputs and must consume them normally. + legacy_resume_component = None + + # A snapshot has to store the component's inputs as they were before the component consumed them, so + # that resuming re-triggers the component the same way this run did. + component_inputs_before_consume = inputs.get(component_name, {}) component_inputs = self._consume_component_inputs( component_name=component_name, component=component, inputs=inputs, is_resume=is_resume ) @@ -444,13 +466,6 @@ def run( # noqa: PLR0915, PLR0912, C901 # initialization component_inputs = self._add_missing_input_defaults(component_inputs, component["input_sockets"]) - # Scenario 1: Pipeline snapshot is provided to resume the pipeline at a specific component - # Deserialize the component_inputs if they are passed in the pipeline_snapshot. - # this check will prevent other component_inputs generated at runtime from being deserialized - if pipeline_snapshot and component_name in pipeline_snapshot.pipeline_state.inputs.keys(): - for key, value in component_inputs.items(): - component_inputs[key] = _deserialize_value_with_schema(value) - try: component_outputs = self._run_component( component_name=component_name, @@ -474,7 +489,7 @@ def run( # noqa: PLR0915, PLR0912, C901 # Create a snapshot of the state of the pipeline before the error occurred. pipeline_snapshot = _create_pipeline_snapshot( inputs=_deepcopy_with_exceptions(inputs), - component_inputs=_deepcopy_with_exceptions(component_inputs), + component_inputs=_deepcopy_with_exceptions(component_inputs_before_consume), break_point=saved_break_point, component_visits=component_visits, original_input_data=data, diff --git a/haystack/dataclasses/breakpoints.py b/haystack/dataclasses/breakpoints.py index 70748146a5a..e6b4098f25d 100644 --- a/haystack/dataclasses/breakpoints.py +++ b/haystack/dataclasses/breakpoints.py @@ -53,11 +53,14 @@ class PipelineState: :param component_visits: A dictionary mapping component names to their visit counts. :param inputs: The inputs processed by the pipeline at the time of the snapshot. :param pipeline_outputs: Dictionary containing the final outputs of the pipeline up to the breakpoint. + :param inputs_format: Which format `inputs` are stored in. `None` marks snapshots taken before Haystack + recorded the sender of each input; those can only be resumed on a component's first visit. """ inputs: dict[str, Any] component_visits: dict[str, int] pipeline_outputs: dict[str, Any] + inputs_format: str | None = None def to_dict(self) -> dict[str, Any]: """ diff --git a/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml new file mode 100644 index 00000000000..099fa2da95b --- /dev/null +++ b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml @@ -0,0 +1,36 @@ +--- +upgrade: + - | + ``PipelineSnapshot.pipeline_state.inputs`` changed shape. It now stores the pipeline's internal inputs, + ``{component: {socket: [{"sender": ..., "value": ...}]}}``, instead of the flattened + ``{component: {socket: value}}``, and ``PipelineState`` gained an ``inputs_format`` field that records which + of the two a snapshot uses. The same change applies to ``BreakpointException.inputs``, which returns that field. + + You are affected if you read ``pipeline_state.inputs`` (or ``BreakpointException.inputs``) directly, for example + to display or post-process a snapshot. Resuming a pipeline with ``Pipeline.run(pipeline_snapshot=...)`` is not + affected, and neither is any code that only passes snapshots around or persists them. + + To adapt, unwrap the sender when reading a value: + + .. code-block:: python + + from haystack.utils import _deserialize_value_with_schema + + inputs = _deserialize_value_with_schema(snapshot.pipeline_state.inputs) + value = inputs["my_component"]["my_socket"][0]["value"] + + Snapshots written by earlier versions of Haystack have ``inputs_format`` set to ``None`` and keep the flattened + shape, so branch on that field if you need to handle both. +fixes: + - | + Fixed resuming a ``Pipeline`` from a ``pipeline_snapshot`` that was taken inside a loop. A snapshot used to + store only the values of the pipeline's inputs, dropping the information about which component had sent each + one. On resume, every restored input therefore looked like it came from outside the pipeline, and such inputs + can only trigger a component on its first visit. Resuming from a breakpoint on a second or later visit failed + with ``PipelineComponentsBlockedError: Cannot run pipeline - all components are blocked``, and a component that + was visited again after the resume could receive its inputs in the wrong shape, for example + ``TypeError: object of type 'int' has no len()`` from a ``BranchJoiner``. + + Snapshots now record the sender of each input, and ``PipelineState`` has a new ``inputs_format`` field marking + which format the inputs are stored in. Snapshots created by earlier versions of Haystack can still be resumed + on a component's first visit, as before; re-create them to resume from a later visit of a loop. diff --git a/test/core/pipeline/test_breakpoint.py b/test/core/pipeline/test_breakpoint.py index 956595da3d9..4dd5fd7ba12 100644 --- a/test/core/pipeline/test_breakpoint.py +++ b/test/core/pipeline/test_breakpoint.py @@ -4,15 +4,23 @@ import json import logging +from dataclasses import replace +from typing import Any import pytest from haystack import component +from haystack.components.joiners import BranchJoiner +from haystack.components.routers import ConditionalRouter +from haystack.components.routers.conditional_router import Route +from haystack.core.component.types import _empty from haystack.core.errors import BreakpointException, PipelineInvalidPipelineSnapshotError from haystack.core.pipeline import Pipeline from haystack.core.pipeline.breakpoint import ( HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, + INTERNAL_INPUTS_FORMAT, _create_pipeline_snapshot, + _deserialize_internal_inputs, _is_snapshot_save_enabled, _save_pipeline_snapshot, _transform_json_structure, @@ -21,6 +29,7 @@ from haystack.dataclasses import ChatMessage from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot, PipelineState from haystack.utils import _deserialize_value_with_schema +from haystack.utils.base_serialization import _serialize_value_with_schema _EMPTY_OBJECT_PAYLOAD = {"serialization_schema": {"type": "object", "properties": {}}, "serialized_data": {}} @@ -142,6 +151,25 @@ def run(self, input_value: str) -> dict[str, str]: return {"result": f"{input_value}_processed"} +@component +class _CountUpTo: + def __init__(self, limit: int) -> None: + self.limit = limit + + @component.output_types(retry=int, done=str) + def run(self, value: int) -> dict[str, Any]: + if value < self.limit: + return {"retry": value + 1} + return {"done": f"finished at {value}"} + + +@component +class _CollectBranches: + @component.output_types(result=str) + def run(self, a: str | None = None, b: str | None = None) -> dict[str, str]: + return {"result": f"a={a} b={b}"} + + def _three_component_pipeline() -> Pipeline: pipeline = Pipeline() pipeline.add_component("comp1", _AppendingComponent()) @@ -152,50 +180,132 @@ def _three_component_pipeline() -> Pipeline: return pipeline -def test_break_point_with_pipeline_snapshot_steps_through_pipeline(): - pipeline = _three_component_pipeline() +def _looping_pipeline() -> Pipeline: + pipeline = Pipeline(max_runs_per_component=20) + pipeline.add_component("joiner", BranchJoiner(int)) + pipeline.add_component("counter", _CountUpTo(limit=5)) + pipeline.connect("joiner.value", "counter.value") + pipeline.connect("counter.retry", "joiner.value") + return pipeline + + +class TestResumeFromPipelineSnapshot: + def test_break_point_with_pipeline_snapshot_steps_through_pipeline(self): + pipeline = _three_component_pipeline() - # run until the breakpoint on comp2 - with pytest.raises(BreakpointException) as exc_info: - pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) - first_snapshot = exc_info.value.pipeline_snapshot - assert first_snapshot is not None - assert first_snapshot.pipeline_state.component_visits == {"comp1": 1, "comp2": 0, "comp3": 0} + # run until the breakpoint on comp2 + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) + first_snapshot = exc_info.value.pipeline_snapshot + assert first_snapshot is not None + assert first_snapshot.pipeline_state.component_visits == {"comp1": 1, "comp2": 0, "comp3": 0} - # step: resume from the snapshot and pause again at comp3 - with pytest.raises(BreakpointException) as exc_info: - pipeline.run(data={}, pipeline_snapshot=first_snapshot, break_point=Breakpoint(component_name="comp3")) - second_snapshot = exc_info.value.pipeline_snapshot - assert second_snapshot is not None - assert second_snapshot.pipeline_state.component_visits == {"comp1": 1, "comp2": 1, "comp3": 0} + # step: resume from the snapshot and pause again at comp3 + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={}, pipeline_snapshot=first_snapshot, break_point=Breakpoint(component_name="comp3")) + second_snapshot = exc_info.value.pipeline_snapshot + assert second_snapshot is not None + assert second_snapshot.pipeline_state.component_visits == {"comp1": 1, "comp2": 1, "comp3": 0} - # resume from the second snapshot and run to completion - result = pipeline.run(data={}, pipeline_snapshot=second_snapshot) - assert result["comp3"]["result"] == "test_processed_processed_processed" + # resume from the second snapshot and run to completion + result = pipeline.run(data={}, pipeline_snapshot=second_snapshot) + assert result["comp3"]["result"] == "test_processed_processed_processed" + def test_break_point_on_earlier_component_than_pipeline_snapshot_never_triggers(self): + pipeline = _three_component_pipeline() -def test_break_point_on_earlier_component_than_pipeline_snapshot_never_triggers(): - pipeline = _three_component_pipeline() + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) + snapshot = exc_info.value.pipeline_snapshot - with pytest.raises(BreakpointException) as exc_info: - pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) - snapshot = exc_info.value.pipeline_snapshot + # comp1 already ran before the snapshot was taken, so a breakpoint on it never triggers + # and the resumed run completes normally + result = pipeline.run(data={}, pipeline_snapshot=snapshot, break_point=Breakpoint(component_name="comp1")) + assert result["comp3"]["result"] == "test_processed_processed_processed" - # comp1 already ran before the snapshot was taken, so a breakpoint on it never triggers - # and the resumed run completes normally - result = pipeline.run(data={}, pipeline_snapshot=snapshot, break_point=Breakpoint(component_name="comp1")) - assert result["comp3"]["result"] == "test_processed_processed_processed" + def test_break_point_matching_pipeline_snapshot_break_point_raises(self): + pipeline = _three_component_pipeline() + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) + snapshot = exc_info.value.pipeline_snapshot -def test_break_point_matching_pipeline_snapshot_break_point_raises(): - pipeline = _three_component_pipeline() + with pytest.raises(PipelineInvalidPipelineSnapshotError, match="different component or visit count"): + pipeline.run( + data={}, pipeline_snapshot=snapshot, break_point=Breakpoint(component_name="comp2", visit_count=0) + ) - with pytest.raises(BreakpointException) as exc_info: - pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) - snapshot = exc_info.value.pipeline_snapshot + @pytest.mark.parametrize("visit_count", [0, 1, 2, 3]) + def test_break_point_in_loop_resumes_on_any_visit(self, visit_count): + """A component paused on a later visit of a loop must still resume and finish the loop.""" + expected = _looping_pipeline().run({"joiner": {"value": 0}}) - with pytest.raises(PipelineInvalidPipelineSnapshotError, match="different component or visit count"): - pipeline.run(data={}, pipeline_snapshot=snapshot, break_point=Breakpoint(component_name="comp2", visit_count=0)) + with pytest.raises(BreakpointException) as exc_info: + _looping_pipeline().run( + {"joiner": {"value": 0}}, break_point=Breakpoint(component_name="joiner", visit_count=visit_count) + ) + snapshot = exc_info.value.pipeline_snapshot + assert snapshot is not None + assert snapshot.pipeline_state.component_visits["joiner"] == visit_count + + assert _looping_pipeline().run(data={}, pipeline_snapshot=snapshot) == expected + + def test_snapshot_preserves_the_sender_of_each_input(self): + pipeline = _three_component_pipeline() + + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) + snapshot = exc_info.value.pipeline_snapshot + assert snapshot is not None + + assert snapshot.pipeline_state.inputs_format == INTERNAL_INPUTS_FORMAT + restored = _deserialize_internal_inputs(snapshot.pipeline_state.inputs) + # comp2's input came from comp1, not from outside the pipeline + assert restored["comp2"] == {"input_value": [{"sender": "comp1", "value": "test_processed"}]} + + def test_snapshot_preserves_sockets_whose_sender_produced_no_output(self): + """A router that leaves a branch inactive puts the `_NO_OUTPUT_PRODUCED` sentinel in the pipeline inputs.""" + routes: list[Route] = [ + {"condition": "{{ n > 5 }}", "output": "{{ 'big' }}", "output_name": "big", "output_type": str}, + {"condition": "{{ n <= 5 }}", "output": "{{ 'small' }}", "output_name": "small", "output_type": str}, + ] + pipeline = Pipeline() + pipeline.add_component("router", ConditionalRouter(routes=routes)) + pipeline.add_component("collect", _CollectBranches()) + pipeline.connect("router.big", "collect.a") + pipeline.connect("router.small", "collect.b") + + expected = pipeline.run({"router": {"n": 9}}) + + with pytest.raises(BreakpointException) as exc_info: + pipeline.run({"router": {"n": 9}}, break_point=Breakpoint(component_name="collect")) + snapshot = exc_info.value.pipeline_snapshot + assert snapshot is not None + + restored = _deserialize_internal_inputs(snapshot.pipeline_state.inputs) + assert restored["collect"]["b"] == [{"sender": "router", "value": _empty}] + + assert pipeline.run(data={}, pipeline_snapshot=snapshot) == expected + + def test_resume_from_legacy_snapshot_without_sender_information(self): + """Snapshots taken before the sender was recorded store flattened values and must still resume.""" + pipeline = _three_component_pipeline() + + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) + snapshot = exc_info.value.pipeline_snapshot + assert snapshot is not None + + legacy_inputs = _serialize_value_with_schema( + _transform_json_structure(_deserialize_internal_inputs(snapshot.pipeline_state.inputs)) + ) + assert legacy_inputs["serialized_data"]["comp2"] == {"input_value": "test_processed"} + legacy_snapshot = replace( + snapshot, pipeline_state=replace(snapshot.pipeline_state, inputs=legacy_inputs, inputs_format=None) + ) + + result = pipeline.run(data={}, pipeline_snapshot=legacy_snapshot) + assert result["comp3"]["result"] == "test_processed_processed_processed" class TestCreatePipelineSnapshot: @@ -206,7 +316,7 @@ def test_create_pipeline_snapshot_all_fields(self): snapshot = _create_pipeline_snapshot( inputs={"comp1": {"input_value": [{"sender": None, "value": "test"}]}, "comp2": {}}, - component_inputs={"input_value": "processed_test"}, + component_inputs={"input_value": [{"sender": "comp1", "value": "processed_test"}]}, break_point=break_point, component_visits={"comp1": 1, "comp2": 0}, original_input_data={"comp1": {"input_value": "test"}}, @@ -225,16 +335,32 @@ def test_create_pipeline_snapshot_all_fields(self): assert snapshot.ordered_component_names == ordered_component_names assert snapshot.break_point == break_point assert snapshot.include_outputs_from == include_outputs_from + + # Each input a socket received is stored under its position, so that it carries its own schema. + def socket_schema(sender_type: str) -> dict[str, Any]: + return { + "type": "object", + "properties": { + "0": { + "type": "object", + "properties": {"sender": {"type": sender_type}, "value": {"type": "string"}}, + } + }, + } + assert snapshot.pipeline_state == PipelineState( inputs={ "serialization_schema": { "type": "object", "properties": { - "comp1": {"type": "object", "properties": {"input_value": {"type": "string"}}}, - "comp2": {"type": "object", "properties": {"input_value": {"type": "string"}}}, + "comp1": {"type": "object", "properties": {"input_value": socket_schema("null")}}, + "comp2": {"type": "object", "properties": {"input_value": socket_schema("string")}}, }, }, - "serialized_data": {"comp1": {"input_value": "test"}, "comp2": {"input_value": "processed_test"}}, + "serialized_data": { + "comp1": {"input_value": {"0": {"sender": None, "value": "test"}}}, + "comp2": {"input_value": {"0": {"sender": "comp1", "value": "processed_test"}}}, + }, }, component_visits={"comp1": 1, "comp2": 0}, pipeline_outputs={ @@ -244,6 +370,7 @@ def test_create_pipeline_snapshot_all_fields(self): }, "serialized_data": {"comp1": {"result": "processed_test"}}, }, + inputs_format=INTERNAL_INPUTS_FORMAT, ) def test_create_pipeline_snapshot_with_dataclasses_in_pipeline_outputs(self): @@ -281,6 +408,7 @@ def test_create_pipeline_snapshot_with_dataclasses_in_pipeline_outputs(self): "comp1": {"result": {"role": "user", "meta": {}, "name": None, "content": [{"text": "hello"}]}} }, }, + inputs_format=INTERNAL_INPUTS_FORMAT, ) def test_create_pipeline_snapshot_non_serializable_inputs(self, caplog): @@ -337,7 +465,7 @@ def to_dict(self): # The non-serializable comp1 field is omitted while the serializable siblings are preserved. assert "comp1" not in deserialized_inputs - assert deserialized_inputs["comp2"] == {"input_value": "keep me"} + assert deserialized_inputs["comp2"] == {"input_value": {"0": {"sender": None, "value": "keep me"}}} assert deserialized_inputs["comp3"] == {} # original_input_data and pipeline_outputs degrade to empty-but-valid payloads. assert deserialized_original_input_data == {} From c6a1517bc3cac551e055294fcbdb303a8e1ba5dd Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 11:17:33 +0200 Subject: [PATCH 02/13] updates --- haystack/core/component/types.py | 23 ++++++++- haystack/core/pipeline/base.py | 7 +-- haystack/core/pipeline/component_checks.py | 15 +++--- ...shot-resume-in-loops-0ab593c343026abb.yaml | 47 ++++++++++++------- 4 files changed, 63 insertions(+), 29 deletions(-) diff --git a/haystack/core/component/types.py b/haystack/core/component/types.py index ecf75e74560..3f83c8c142b 100644 --- a/haystack/core/component/types.py +++ b/haystack/core/component/types.py @@ -36,12 +36,18 @@ class _Empty: Type of the `_empty` sentinel, which marks an `InputSocket.default_value` as not set. The sentinel is also reused by the pipeline to mark a socket whose sender ran without producing a value for it. - It is compared by identity, so `from_dict` returns the singleton rather than a new instance. + Use `_is_empty_sentinel` to test for it. The single `_empty` instance is the one to use everywhere, so `from_dict` + and `__reduce__` both hand it back rather than building a second one. """ def __repr__(self) -> str: return "_empty" + def __reduce__(self) -> str: + # Returning the global's name keeps `copy`, `deepcopy` and `pickle` from building a second instance, which + # would silently fail the identity comparisons. + return "_empty" + def to_dict(self) -> dict[str, Any]: """ Serialize the sentinel. It carries no state, so the payload is empty. @@ -59,6 +65,19 @@ def from_dict(cls, data: dict[str, Any]) -> "_Empty": # noqa: ARG003 _empty = _Empty() +def _is_empty_sentinel(value: Any) -> bool: + """ + Check whether a value is the `_empty` sentinel. + + Compares by type rather than by identity or equality. `_Empty` carries no state, so every instance means the same + thing, and unlike `==` this never runs `__eq__` on an arbitrary value, which for values such as numpy arrays does + not return a bool. + + :param value: The value to check. + """ + return isinstance(value, _Empty) + + @dataclass class InputSocket: """ @@ -97,7 +116,7 @@ def is_variadic(self) -> bool: @property def is_mandatory(self) -> bool: """Check if the input is mandatory.""" - return self.default_value == _empty + return _is_empty_sentinel(self.default_value) def __post_init__(self) -> None: try: diff --git a/haystack/core/pipeline/base.py b/haystack/core/pipeline/base.py index 0784771745c..ee1947d2152 100644 --- a/haystack/core/pipeline/base.py +++ b/haystack/core/pipeline/base.py @@ -28,6 +28,7 @@ ) from haystack.core.pipeline.component_checks import ( _NO_OUTPUT_PRODUCED, + _no_output_produced, all_predecessors_executed, are_all_sockets_ready, can_component_run, @@ -1223,7 +1224,7 @@ def _consume_component_inputs( greedy_inputs_to_remove = set() for socket_name, socket in component["input_sockets"].items(): socket_inputs = component_inputs.get(socket_name, []) - socket_inputs_values = [sock["value"] for sock in socket_inputs if sock["value"] is not _NO_OUTPUT_PRODUCED] + socket_inputs_values = [sock["value"] for sock in socket_inputs if not _no_output_produced(sock["value"])] # if we are resuming a component, the inputs are already consumed, so we just return the first input if is_resume: @@ -1551,7 +1552,7 @@ def _write_component_outputs( # This allows us to track if a predecessor already ran but did not produce an output. value = component_outputs.get(sender_socket.name, _NO_OUTPUT_PRODUCED) - if value is not _NO_OUTPUT_PRODUCED and conversion_strategy: + if not _no_output_produced(value) and conversion_strategy: try: value = _convert_value(value=value, conversion_strategy=conversion_strategy) except Exception as e: @@ -1861,5 +1862,5 @@ def _write_to_standard_socket( current_value = inputs[receiver_name].get(receiver_socket_name) # Only overwrite if there's no existing value, or we have a new value to provide - if current_value is None or value is not _NO_OUTPUT_PRODUCED: + if current_value is None or not _no_output_produced(value): inputs[receiver_name][receiver_socket_name] = [{"sender": component_name, "value": value}] diff --git a/haystack/core/pipeline/component_checks.py b/haystack/core/pipeline/component_checks.py index d05d373338a..fa4b19341bb 100644 --- a/haystack/core/pipeline/component_checks.py +++ b/haystack/core/pipeline/component_checks.py @@ -4,10 +4,13 @@ from typing import Any -from haystack.core.component.types import InputSocket, _empty +from haystack.core.component.types import InputSocket, _empty, _is_empty_sentinel _NO_OUTPUT_PRODUCED = _empty +# A socket input holding the sentinel records that its sender ran without producing a value for that socket. +_no_output_produced = _is_empty_sentinel + def can_component_run(component: dict, inputs: dict) -> bool: """ @@ -103,7 +106,7 @@ def any_socket_value_from_predecessor_received(socket_inputs: list[dict[str, Any :param socket_inputs: Inputs for the component's socket. """ # When sender is None, the input was provided from outside the pipeline. - return any(inp["value"] is not _NO_OUTPUT_PRODUCED and inp["sender"] is not None for inp in socket_inputs) + return any(not _no_output_produced(inp["value"]) and inp["sender"] is not None for inp in socket_inputs) def has_user_input(inputs: dict) -> bool: @@ -142,7 +145,7 @@ def any_socket_input_received(socket_inputs: list[dict]) -> bool: :param socket_inputs: Inputs for the socket. """ - return any(inp["value"] is not _NO_OUTPUT_PRODUCED for inp in socket_inputs) + return any(not _no_output_produced(inp["value"]) for inp in socket_inputs) def has_lazy_variadic_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict]) -> bool: @@ -156,7 +159,7 @@ def has_lazy_variadic_socket_received_all_inputs(socket: InputSocket, socket_inp actual_senders = { sock["sender"] for sock in socket_inputs - if sock["value"] is not _NO_OUTPUT_PRODUCED and sock["sender"] is not None + if not _no_output_produced(sock["value"]) and sock["sender"] is not None } return expected_senders == actual_senders @@ -176,7 +179,7 @@ def has_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict if ( socket.is_variadic and socket.is_greedy - and any(sock["value"] is not _NO_OUTPUT_PRODUCED for sock in socket_inputs) + and any(not _no_output_produced(sock["value"]) for sock in socket_inputs) ): return True @@ -185,7 +188,7 @@ def has_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict return True # The socket is not variadic and the only expected input is complete. - return not socket.is_variadic and socket_inputs[0]["value"] is not _NO_OUTPUT_PRODUCED + return not socket.is_variadic and not _no_output_produced(socket_inputs[0]["value"]) def all_predecessors_executed(component: dict, inputs: dict) -> bool: diff --git a/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml index 099fa2da95b..65550166d7c 100644 --- a/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml +++ b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml @@ -1,36 +1,47 @@ --- upgrade: - | - ``PipelineSnapshot.pipeline_state.inputs`` changed shape. It now stores the pipeline's internal inputs, - ``{component: {socket: [{"sender": ..., "value": ...}]}}``, instead of the flattened - ``{component: {socket: value}}``, and ``PipelineState`` gained an ``inputs_format`` field that records which - of the two a snapshot uses. The same change applies to ``BreakpointException.inputs``, which returns that field. + ``PipelineSnapshot.pipeline_state.inputs`` changed shape. It used to store one flattened value per socket, + ``{component: {socket: value}}``. It now stores the pipeline's internal inputs, keeping the component that sent + each one and keying the inputs a socket received by their position: + ``{component: {socket: {"0": {"sender": ..., "value": ...}}}}``. Positions are used instead of a list so that + each input carries its own type schema, which a list cannot do. ``PipelineState`` gained an ``inputs_format`` + field recording which of the two shapes a snapshot uses. The same applies to ``BreakpointException.inputs``, + which returns that field. You are affected if you read ``pipeline_state.inputs`` (or ``BreakpointException.inputs``) directly, for example to display or post-process a snapshot. Resuming a pipeline with ``Pipeline.run(pipeline_snapshot=...)`` is not - affected, and neither is any code that only passes snapshots around or persists them. + affected, and neither is code that only passes snapshots around or persists them. - To adapt, unwrap the sender when reading a value: + To adapt, read through the position and unwrap the sender: .. code-block:: python from haystack.utils import _deserialize_value_with_schema inputs = _deserialize_value_with_schema(snapshot.pipeline_state.inputs) - value = inputs["my_component"]["my_socket"][0]["value"] + value = inputs["my_component"]["my_socket"]["0"]["value"] Snapshots written by earlier versions of Haystack have ``inputs_format`` set to ``None`` and keep the flattened shape, so branch on that field if you need to handle both. fixes: - | - Fixed resuming a ``Pipeline`` from a ``pipeline_snapshot`` that was taken inside a loop. A snapshot used to - store only the values of the pipeline's inputs, dropping the information about which component had sent each - one. On resume, every restored input therefore looked like it came from outside the pipeline, and such inputs - can only trigger a component on its first visit. Resuming from a breakpoint on a second or later visit failed - with ``PipelineComponentsBlockedError: Cannot run pipeline - all components are blocked``, and a component that - was visited again after the resume could receive its inputs in the wrong shape, for example - ``TypeError: object of type 'int' has no len()`` from a ``BranchJoiner``. - - Snapshots now record the sender of each input, and ``PipelineState`` has a new ``inputs_format`` field marking - which format the inputs are stored in. Snapshots created by earlier versions of Haystack can still be resumed - on a component's first visit, as before; re-create them to resume from a later visit of a loop. + Fixed resuming a ``Pipeline`` from a ``pipeline_snapshot`` that was taken on a component's second or later visit, + which failed with ``PipelineComponentsBlockedError: Cannot run pipeline - all components are blocked``. A snapshot + stored only the values of the pipeline's inputs and dropped the information about which component had sent each + one, so on resume every restored input looked like it came from outside the pipeline, and such an input can only + trigger a component on its first visit. Snapshots now record the sender of each input. Snapshots created by + earlier versions of Haystack can still be resumed on a first visit, as before; re-create them to resume from a + later visit of a loop. + - | + Fixed a resumed ``Pipeline`` passing malformed inputs to the component the snapshot was taken on, whenever that + component ran more than once after the resume, for example inside a loop. Every visit after the first reused the + handling meant only for the paused visit and skipped the regular input consumption, so a variadic component could + receive a bare value where it expected a list, raising errors such as + ``TypeError: object of type 'int' has no len()`` from a ``BranchJoiner``. This affected snapshots taken at any + visit count, including the first. + - | + The ``_empty`` sentinel, which marks an unset ``InputSocket.default_value`` and a socket whose sender produced no + output, is now recognized by type rather than by identity or equality. Testing for it no longer runs ``__eq__`` + on the value being checked, which for values such as numpy arrays does not return a bool, and the sentinel now + survives ``copy``, ``deepcopy``, ``pickle``, and serialization as the same object. From 10efc59f10d5b81a62cab5a5b771caa63d8ac77c Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 11:22:09 +0200 Subject: [PATCH 03/13] remove unnecessary abstraction --- haystack/core/component/types.py | 21 +++++---------------- haystack/core/pipeline/base.py | 8 ++++---- haystack/core/pipeline/component_checks.py | 22 +++++++--------------- 3 files changed, 16 insertions(+), 35 deletions(-) diff --git a/haystack/core/component/types.py b/haystack/core/component/types.py index 3f83c8c142b..f8837d8e1c6 100644 --- a/haystack/core/component/types.py +++ b/haystack/core/component/types.py @@ -36,8 +36,10 @@ class _Empty: Type of the `_empty` sentinel, which marks an `InputSocket.default_value` as not set. The sentinel is also reused by the pipeline to mark a socket whose sender ran without producing a value for it. - Use `_is_empty_sentinel` to test for it. The single `_empty` instance is the one to use everywhere, so `from_dict` - and `__reduce__` both hand it back rather than building a second one. + Test for it with `isinstance(value, _Empty)`: the sentinel carries no state, so every instance means the same + thing, and unlike `==` this never runs `__eq__` on a value such as a numpy array, which does not return a bool. + The single `_empty` instance is still the one to use everywhere, so `from_dict` and `__reduce__` both hand it back + rather than building a second one. """ def __repr__(self) -> str: @@ -65,19 +67,6 @@ def from_dict(cls, data: dict[str, Any]) -> "_Empty": # noqa: ARG003 _empty = _Empty() -def _is_empty_sentinel(value: Any) -> bool: - """ - Check whether a value is the `_empty` sentinel. - - Compares by type rather than by identity or equality. `_Empty` carries no state, so every instance means the same - thing, and unlike `==` this never runs `__eq__` on an arbitrary value, which for values such as numpy arrays does - not return a bool. - - :param value: The value to check. - """ - return isinstance(value, _Empty) - - @dataclass class InputSocket: """ @@ -116,7 +105,7 @@ def is_variadic(self) -> bool: @property def is_mandatory(self) -> bool: """Check if the input is mandatory.""" - return _is_empty_sentinel(self.default_value) + return isinstance(self.default_value, _Empty) def __post_init__(self) -> None: try: diff --git a/haystack/core/pipeline/base.py b/haystack/core/pipeline/base.py index ee1947d2152..398d59cbbe5 100644 --- a/haystack/core/pipeline/base.py +++ b/haystack/core/pipeline/base.py @@ -16,6 +16,7 @@ from haystack import logging, tracing from haystack.core.component import Component, InputSocket, OutputSocket, component +from haystack.core.component.types import _Empty from haystack.core.errors import ( DeserializationError, PipelineComponentsBlockedError, @@ -28,7 +29,6 @@ ) from haystack.core.pipeline.component_checks import ( _NO_OUTPUT_PRODUCED, - _no_output_produced, all_predecessors_executed, are_all_sockets_ready, can_component_run, @@ -1224,7 +1224,7 @@ def _consume_component_inputs( greedy_inputs_to_remove = set() for socket_name, socket in component["input_sockets"].items(): socket_inputs = component_inputs.get(socket_name, []) - socket_inputs_values = [sock["value"] for sock in socket_inputs if not _no_output_produced(sock["value"])] + socket_inputs_values = [sock["value"] for sock in socket_inputs if not isinstance(sock["value"], _Empty)] # if we are resuming a component, the inputs are already consumed, so we just return the first input if is_resume: @@ -1552,7 +1552,7 @@ def _write_component_outputs( # This allows us to track if a predecessor already ran but did not produce an output. value = component_outputs.get(sender_socket.name, _NO_OUTPUT_PRODUCED) - if not _no_output_produced(value) and conversion_strategy: + if not isinstance(value, _Empty) and conversion_strategy: try: value = _convert_value(value=value, conversion_strategy=conversion_strategy) except Exception as e: @@ -1862,5 +1862,5 @@ def _write_to_standard_socket( current_value = inputs[receiver_name].get(receiver_socket_name) # Only overwrite if there's no existing value, or we have a new value to provide - if current_value is None or not _no_output_produced(value): + if current_value is None or not isinstance(value, _Empty): inputs[receiver_name][receiver_socket_name] = [{"sender": component_name, "value": value}] diff --git a/haystack/core/pipeline/component_checks.py b/haystack/core/pipeline/component_checks.py index fa4b19341bb..95d10037191 100644 --- a/haystack/core/pipeline/component_checks.py +++ b/haystack/core/pipeline/component_checks.py @@ -4,13 +4,11 @@ from typing import Any -from haystack.core.component.types import InputSocket, _empty, _is_empty_sentinel +from haystack.core.component.types import InputSocket, _Empty, _empty +# A socket input holding this sentinel records that its sender ran without producing a value for that socket. _NO_OUTPUT_PRODUCED = _empty -# A socket input holding the sentinel records that its sender ran without producing a value for that socket. -_no_output_produced = _is_empty_sentinel - def can_component_run(component: dict, inputs: dict) -> bool: """ @@ -106,7 +104,7 @@ def any_socket_value_from_predecessor_received(socket_inputs: list[dict[str, Any :param socket_inputs: Inputs for the component's socket. """ # When sender is None, the input was provided from outside the pipeline. - return any(not _no_output_produced(inp["value"]) and inp["sender"] is not None for inp in socket_inputs) + return any(not isinstance(inp["value"], _Empty) and inp["sender"] is not None for inp in socket_inputs) def has_user_input(inputs: dict) -> bool: @@ -145,7 +143,7 @@ def any_socket_input_received(socket_inputs: list[dict]) -> bool: :param socket_inputs: Inputs for the socket. """ - return any(not _no_output_produced(inp["value"]) for inp in socket_inputs) + return any(not isinstance(inp["value"], _Empty) for inp in socket_inputs) def has_lazy_variadic_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict]) -> bool: @@ -157,9 +155,7 @@ def has_lazy_variadic_socket_received_all_inputs(socket: InputSocket, socket_inp """ expected_senders = set(socket.senders) actual_senders = { - sock["sender"] - for sock in socket_inputs - if not _no_output_produced(sock["value"]) and sock["sender"] is not None + sock["sender"] for sock in socket_inputs if not isinstance(sock["value"], _Empty) and sock["sender"] is not None } return expected_senders == actual_senders @@ -176,11 +172,7 @@ def has_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict return False # The socket is greedy variadic and at least one input was produced, it is complete. - if ( - socket.is_variadic - and socket.is_greedy - and any(not _no_output_produced(sock["value"]) for sock in socket_inputs) - ): + if socket.is_variadic and socket.is_greedy and any(not isinstance(sock["value"], _Empty) for sock in socket_inputs): return True # The socket is lazy variadic and all expected inputs were produced. @@ -188,7 +180,7 @@ def has_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict return True # The socket is not variadic and the only expected input is complete. - return not socket.is_variadic and not _no_output_produced(socket_inputs[0]["value"]) + return not socket.is_variadic and not isinstance(socket_inputs[0]["value"], _Empty) def all_predecessors_executed(component: dict, inputs: dict) -> bool: From 7b862e9ffaf9de4b1f2c3356817341a31c38b9fd Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 11:22:59 +0200 Subject: [PATCH 04/13] add new test file --- test/core/component/test_types.py | 82 +++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) create mode 100644 test/core/component/test_types.py diff --git a/test/core/component/test_types.py b/test/core/component/test_types.py new file mode 100644 index 00000000000..bd88a463df9 --- /dev/null +++ b/test/core/component/test_types.py @@ -0,0 +1,82 @@ +# SPDX-FileCopyrightText: 2022-present deepset GmbH +# +# SPDX-License-Identifier: Apache-2.0 + +import copy +import pickle + +import pytest + +from haystack.core.component.types import GreedyVariadic, InputSocket, _Empty, _empty +from haystack.core.pipeline.component_checks import ( + _NO_OUTPUT_PRODUCED, + any_socket_input_received, + has_socket_received_all_inputs, +) +from haystack.core.pipeline.utils import _deepcopy_with_exceptions +from haystack.utils.base_serialization import _deserialize_value_with_schema, _serialize_value_with_schema + + +class _AlwaysUnequal: # noqa: PLW1641 # __hash__ is irrelevant; this only needs a hostile __eq__ + """Stands in for values such as numpy arrays, whose `__eq__` does not return a bool.""" + + def __eq__(self, other): + raise AssertionError("__eq__ must not be called when testing for the sentinel") + + +class TestEmptySentinel: + def test_the_sentinel_alias_is_the_singleton(self): + assert _NO_OUTPUT_PRODUCED is _empty + assert repr(_empty) == "_empty" + + @pytest.mark.parametrize("copier", [copy.copy, copy.deepcopy, lambda v: pickle.loads(pickle.dumps(v))]) + def test_copying_the_sentinel_returns_the_singleton(self, copier): + assert copier(_empty) is _empty + + def test_the_sentinel_survives_a_deepcopy_of_the_pipeline_inputs(self): + """`Pipeline.run` deep-copies its inputs, which carry the sentinel, before snapshotting them.""" + inputs = {"comp": {"socket": [{"sender": "other", "value": _empty}]}} + + copied = _deepcopy_with_exceptions(inputs) + + assert copied["comp"]["socket"][0]["value"] is _empty + + def test_the_sentinel_survives_serialization(self): + restored = _deserialize_value_with_schema(_serialize_value_with_schema({"value": _empty})) + + assert restored["value"] is _empty + + +class TestSocketChecksAgainstTheSentinel: + """The socket checks compare by type, so they hold for any `_Empty` instance and never run `__eq__`.""" + + @pytest.mark.parametrize("sentinel", [_empty, _Empty()], ids=["singleton", "separate instance"]) + def test_a_socket_holding_only_the_sentinel_received_nothing(self, sentinel): + socket = InputSocket(name="value", type=GreedyVariadic[int], senders=["sender"]) + socket_inputs = [{"sender": "sender", "value": sentinel}] + + assert not any_socket_input_received(socket_inputs) + assert not has_socket_received_all_inputs(socket, socket_inputs) + + def test_a_socket_holding_a_value_received_it(self): + socket = InputSocket(name="value", type=GreedyVariadic[int], senders=["sender"]) + socket_inputs = [{"sender": "sender", "value": 5}] + + assert any_socket_input_received(socket_inputs) + assert has_socket_received_all_inputs(socket, socket_inputs) + + def test_a_value_that_cannot_be_compared_counts_as_received(self): + socket = InputSocket(name="value", type=GreedyVariadic[int], senders=["sender"]) + socket_inputs = [{"sender": "sender", "value": _AlwaysUnequal()}] + + assert any_socket_input_received(socket_inputs) + assert has_socket_received_all_inputs(socket, socket_inputs) + + +class TestInputSocketIsMandatory: + def test_a_socket_without_a_default_is_mandatory(self): + assert InputSocket(name="value", type=str).is_mandatory + + @pytest.mark.parametrize("default", [None, 0, "", [], False, _AlwaysUnequal()], ids=repr) + def test_a_socket_with_any_default_is_not_mandatory(self, default): + assert not InputSocket(name="value", type=str, default_value=default).is_mandatory From a8be66040ed5052fe9b7d8294dd7a6be01ea288b Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 11:36:20 +0200 Subject: [PATCH 05/13] refactoring --- haystack/core/pipeline/breakpoint.py | 7 +-- haystack/core/pipeline/pipeline.py | 3 +- haystack/dataclasses/breakpoints.py | 11 +++- test/core/component/test_types.py | 59 ++++++--------------- test/core/pipeline/test_breakpoint.py | 3 +- test/core/pipeline/test_component_checks.py | 19 ++++++- 6 files changed, 45 insertions(+), 57 deletions(-) diff --git a/haystack/core/pipeline/breakpoint.py b/haystack/core/pipeline/breakpoint.py index dddd7b1e801..e3afd50c85e 100644 --- a/haystack/core/pipeline/breakpoint.py +++ b/haystack/core/pipeline/breakpoint.py @@ -13,7 +13,7 @@ from haystack import logging from haystack.core.errors import PipelineInvalidPipelineSnapshotError -from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot, PipelineState +from haystack.dataclasses.breakpoints import INTERNAL_INPUTS_FORMAT, Breakpoint, PipelineSnapshot, PipelineState from haystack.utils import _deserialize_value_with_schema from haystack.utils.base_serialization import _serialize_with_field_fallback @@ -26,11 +26,6 @@ # The callback receives a PipelineSnapshot and optionally returns a file path string SnapshotCallback = Callable[[PipelineSnapshot], str | None] -# Value of `PipelineState.inputs_format` for snapshots that store the pipeline's internal inputs state, where each -# input keeps the component that sent it. Snapshots without it store flattened `{component: {socket: value}}` inputs. -# See `_serialize_internal_inputs` for the exact shape. -INTERNAL_INPUTS_FORMAT = "internal" - def _is_snapshot_save_enabled() -> bool: """ diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py index a71eb255a01..f1ec4880a62 100644 --- a/haystack/core/pipeline/pipeline.py +++ b/haystack/core/pipeline/pipeline.py @@ -19,7 +19,6 @@ _validate_component_output_keys, ) from haystack.core.pipeline.breakpoint import ( - INTERNAL_INPUTS_FORMAT, SnapshotCallback, _create_pipeline_snapshot, _deserialize_internal_inputs, @@ -29,7 +28,7 @@ ) from haystack.core.pipeline.utils import _deepcopy_with_exceptions from haystack.dataclasses import AsyncStreamingCallbackT, StreamingCallbackT, StreamingChunk, select_streaming_callback -from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot +from haystack.dataclasses.breakpoints import INTERNAL_INPUTS_FORMAT, Breakpoint, PipelineSnapshot from haystack.dataclasses.streaming_chunk import _invoke_streaming_callback from haystack.telemetry import pipeline_running from haystack.utils import _deserialize_value_with_schema diff --git a/haystack/dataclasses/breakpoints.py b/haystack/dataclasses/breakpoints.py index e6b4098f25d..f41f3f335b4 100644 --- a/haystack/dataclasses/breakpoints.py +++ b/haystack/dataclasses/breakpoints.py @@ -8,6 +8,10 @@ from haystack.utils.dataclasses import _warn_on_inplace_mutation +# Value of `PipelineState.inputs_format` used by every snapshot Haystack creates now. See that field for the shape +# each format implies. +INTERNAL_INPUTS_FORMAT = "internal" + @dataclass(frozen=True) class Breakpoint: @@ -53,8 +57,11 @@ class PipelineState: :param component_visits: A dictionary mapping component names to their visit counts. :param inputs: The inputs processed by the pipeline at the time of the snapshot. :param pipeline_outputs: Dictionary containing the final outputs of the pipeline up to the breakpoint. - :param inputs_format: Which format `inputs` are stored in. `None` marks snapshots taken before Haystack - recorded the sender of each input; those can only be resumed on a component's first visit. + :param inputs_format: Which format `inputs` are stored in. `"internal"` means each input records the component + that sent it, keyed by the position it arrived in, as in + `{component: {socket: {"0": {"sender": ..., "value": ...}}}}`. `None` marks snapshots taken before Haystack + recorded the sender, which hold one flattened value per socket, `{component: {socket: value}}`, and can only + be resumed on a component's first visit. """ inputs: dict[str, Any] diff --git a/test/core/component/test_types.py b/test/core/component/test_types.py index bd88a463df9..cdfca35c0d8 100644 --- a/test/core/component/test_types.py +++ b/test/core/component/test_types.py @@ -7,12 +7,7 @@ import pytest -from haystack.core.component.types import GreedyVariadic, InputSocket, _Empty, _empty -from haystack.core.pipeline.component_checks import ( - _NO_OUTPUT_PRODUCED, - any_socket_input_received, - has_socket_received_all_inputs, -) +from haystack.core.component.types import InputSocket, _empty from haystack.core.pipeline.utils import _deepcopy_with_exceptions from haystack.utils.base_serialization import _deserialize_value_with_schema, _serialize_value_with_schema @@ -25,13 +20,18 @@ def __eq__(self, other): class TestEmptySentinel: - def test_the_sentinel_alias_is_the_singleton(self): - assert _NO_OUTPUT_PRODUCED is _empty + """`_empty` is only ever recognized as itself, so anything that could duplicate it has to hand it back.""" + + def test_repr_names_the_sentinel(self): assert repr(_empty) == "_empty" - @pytest.mark.parametrize("copier", [copy.copy, copy.deepcopy, lambda v: pickle.loads(pickle.dumps(v))]) - def test_copying_the_sentinel_returns_the_singleton(self, copier): - assert copier(_empty) is _empty + @pytest.mark.parametrize( + "duplicate", + [copy.copy, copy.deepcopy, lambda v: pickle.loads(pickle.dumps(v))], + ids=["copy", "deepcopy", "pickle"], + ) + def test_duplicating_the_sentinel_returns_the_singleton(self, duplicate): + assert duplicate(_empty) is _empty def test_the_sentinel_survives_a_deepcopy_of_the_pipeline_inputs(self): """`Pipeline.run` deep-copies its inputs, which carry the sentinel, before snapshotting them.""" @@ -42,41 +42,12 @@ def test_the_sentinel_survives_a_deepcopy_of_the_pipeline_inputs(self): assert copied["comp"]["socket"][0]["value"] is _empty def test_the_sentinel_survives_serialization(self): + """A pipeline snapshot serializes the inputs, sentinel included.""" restored = _deserialize_value_with_schema(_serialize_value_with_schema({"value": _empty})) assert restored["value"] is _empty -class TestSocketChecksAgainstTheSentinel: - """The socket checks compare by type, so they hold for any `_Empty` instance and never run `__eq__`.""" - - @pytest.mark.parametrize("sentinel", [_empty, _Empty()], ids=["singleton", "separate instance"]) - def test_a_socket_holding_only_the_sentinel_received_nothing(self, sentinel): - socket = InputSocket(name="value", type=GreedyVariadic[int], senders=["sender"]) - socket_inputs = [{"sender": "sender", "value": sentinel}] - - assert not any_socket_input_received(socket_inputs) - assert not has_socket_received_all_inputs(socket, socket_inputs) - - def test_a_socket_holding_a_value_received_it(self): - socket = InputSocket(name="value", type=GreedyVariadic[int], senders=["sender"]) - socket_inputs = [{"sender": "sender", "value": 5}] - - assert any_socket_input_received(socket_inputs) - assert has_socket_received_all_inputs(socket, socket_inputs) - - def test_a_value_that_cannot_be_compared_counts_as_received(self): - socket = InputSocket(name="value", type=GreedyVariadic[int], senders=["sender"]) - socket_inputs = [{"sender": "sender", "value": _AlwaysUnequal()}] - - assert any_socket_input_received(socket_inputs) - assert has_socket_received_all_inputs(socket, socket_inputs) - - -class TestInputSocketIsMandatory: - def test_a_socket_without_a_default_is_mandatory(self): - assert InputSocket(name="value", type=str).is_mandatory - - @pytest.mark.parametrize("default", [None, 0, "", [], False, _AlwaysUnequal()], ids=repr) - def test_a_socket_with_any_default_is_not_mandatory(self, default): - assert not InputSocket(name="value", type=str, default_value=default).is_mandatory +def test_input_socket_with_an_incomparable_default_is_not_mandatory(): + """`is_mandatory` tests for the sentinel by type, so it never compares the default it was given.""" + assert not InputSocket(name="value", type=str, default_value=_AlwaysUnequal()).is_mandatory diff --git a/test/core/pipeline/test_breakpoint.py b/test/core/pipeline/test_breakpoint.py index 4dd5fd7ba12..3787e567b5b 100644 --- a/test/core/pipeline/test_breakpoint.py +++ b/test/core/pipeline/test_breakpoint.py @@ -18,7 +18,6 @@ from haystack.core.pipeline import Pipeline from haystack.core.pipeline.breakpoint import ( HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, - INTERNAL_INPUTS_FORMAT, _create_pipeline_snapshot, _deserialize_internal_inputs, _is_snapshot_save_enabled, @@ -27,7 +26,7 @@ load_pipeline_snapshot, ) from haystack.dataclasses import ChatMessage -from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot, PipelineState +from haystack.dataclasses.breakpoints import INTERNAL_INPUTS_FORMAT, Breakpoint, PipelineSnapshot, PipelineState from haystack.utils import _deserialize_value_with_schema from haystack.utils.base_serialization import _serialize_value_with_schema diff --git a/test/core/pipeline/test_component_checks.py b/test/core/pipeline/test_component_checks.py index e60e7c2bd10..1b2ba07faab 100644 --- a/test/core/pipeline/test_component_checks.py +++ b/test/core/pipeline/test_component_checks.py @@ -7,7 +7,7 @@ import pytest from pandas import DataFrame -from haystack.core.component.types import GreedyVariadic, InputSocket, OutputSocket, Variadic +from haystack.core.component.types import GreedyVariadic, InputSocket, OutputSocket, Variadic, _Empty from haystack.core.pipeline.component_checks import ( _NO_OUTPUT_PRODUCED, all_predecessors_executed, @@ -26,6 +26,13 @@ ) +class _IncomparableValue: # noqa: PLW1641 # __hash__ is irrelevant; this only needs a hostile __eq__ + """Stands in for values such as numpy arrays, whose `__eq__` does not return a bool.""" + + def __eq__(self, other): + raise AssertionError("__eq__ must not be called when testing for the sentinel") + + @pytest.fixture def basic_component(): """Basic component with one mandatory and one optional input.""" @@ -477,6 +484,16 @@ def test_any_socket_input_received_empty_list(self): """Empty list: no input received.""" assert any_socket_input_received([]) is False + def test_any_socket_input_received_with_a_separate_sentinel_instance(self): + """The sentinel is recognized by type, so a second instance means no output just as the singleton does.""" + socket_inputs = [{"sender": "component1", "value": _Empty()}] + assert any_socket_input_received(socket_inputs) is False + + def test_any_socket_input_received_with_an_incomparable_value(self): + """Recognizing the sentinel by type means a value that cannot be compared is still counted as received.""" + socket_inputs = [{"sender": "component1", "value": _IncomparableValue()}] + assert any_socket_input_received(socket_inputs) is True + class TestLazyVariadicSocket: def test_lazy_variadic_all_inputs_received(self, variadic_socket_with_senders): From d19ee3fc71671c0155ae89caeea9b34f6619e15b Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 11:51:12 +0200 Subject: [PATCH 06/13] Refactoring to make _empty and _NO_OUTPUT_PRODUCED separate things --- haystack/core/component/sockets.py | 2 +- haystack/core/component/types.py | 40 ++----------- haystack/core/pipeline/base.py | 10 ++-- haystack/core/pipeline/component_checks.py | 57 ++++++++++++++++--- ...shot-resume-in-loops-0ab593c343026abb.yaml | 5 -- test/core/component/test_types.py | 44 ++------------ test/core/pipeline/test_breakpoint.py | 4 +- test/core/pipeline/test_component_checks.py | 57 ++++++++++++++----- 8 files changed, 109 insertions(+), 110 deletions(-) diff --git a/haystack/core/component/sockets.py b/haystack/core/component/sockets.py index f940616e5b3..6a7d7c5db03 100644 --- a/haystack/core/component/sockets.py +++ b/haystack/core/component/sockets.py @@ -46,7 +46,7 @@ class Sockets: # noqa: PLW1641 # >> - documents: Any inputs.question - # >> InputSocket(name='question', type=typing.Any, default_value=_empty, ... + # >> InputSocket(name='question', type=typing.Any, default_value=, ... ``` """ diff --git a/haystack/core/component/types.py b/haystack/core/component/types.py index f8837d8e1c6..d701a8076de 100644 --- a/haystack/core/component/types.py +++ b/haystack/core/component/types.py @@ -31,40 +31,8 @@ GreedyVariadic: TypeAlias = Annotated[Iterable[T], HAYSTACK_GREEDY_VARIADIC_ANNOTATION] -class _Empty: - """ - Type of the `_empty` sentinel, which marks an `InputSocket.default_value` as not set. - - The sentinel is also reused by the pipeline to mark a socket whose sender ran without producing a value for it. - Test for it with `isinstance(value, _Empty)`: the sentinel carries no state, so every instance means the same - thing, and unlike `==` this never runs `__eq__` on a value such as a numpy array, which does not return a bool. - The single `_empty` instance is still the one to use everywhere, so `from_dict` and `__reduce__` both hand it back - rather than building a second one. - """ - - def __repr__(self) -> str: - return "_empty" - - def __reduce__(self) -> str: - # Returning the global's name keeps `copy`, `deepcopy` and `pickle` from building a second instance, which - # would silently fail the identity comparisons. - return "_empty" - - def to_dict(self) -> dict[str, Any]: - """ - Serialize the sentinel. It carries no state, so the payload is empty. - """ - return {} - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "_Empty": # noqa: ARG003 - """ - Deserialize the sentinel back to the singleton, so that identity comparisons keep working. - """ - return _empty - - -_empty = _Empty() +class _empty: + """Custom object for marking InputSocket.default_value as not set.""" @dataclass @@ -105,7 +73,9 @@ def is_variadic(self) -> bool: @property def is_mandatory(self) -> bool: """Check if the input is mandatory.""" - return isinstance(self.default_value, _Empty) + # Compared by identity so that a default value with a custom `__eq__`, such as a numpy array, is not asked + # whether it equals the sentinel. + return self.default_value is _empty def __post_init__(self) -> None: try: diff --git a/haystack/core/pipeline/base.py b/haystack/core/pipeline/base.py index 398d59cbbe5..9e21c3ef637 100644 --- a/haystack/core/pipeline/base.py +++ b/haystack/core/pipeline/base.py @@ -16,7 +16,6 @@ from haystack import logging, tracing from haystack.core.component import Component, InputSocket, OutputSocket, component -from haystack.core.component.types import _Empty from haystack.core.errors import ( DeserializationError, PipelineComponentsBlockedError, @@ -29,6 +28,7 @@ ) from haystack.core.pipeline.component_checks import ( _NO_OUTPUT_PRODUCED, + _NoOutputProduced, all_predecessors_executed, are_all_sockets_ready, can_component_run, @@ -1224,7 +1224,9 @@ def _consume_component_inputs( greedy_inputs_to_remove = set() for socket_name, socket in component["input_sockets"].items(): socket_inputs = component_inputs.get(socket_name, []) - socket_inputs_values = [sock["value"] for sock in socket_inputs if not isinstance(sock["value"], _Empty)] + socket_inputs_values = [ + sock["value"] for sock in socket_inputs if not isinstance(sock["value"], _NoOutputProduced) + ] # if we are resuming a component, the inputs are already consumed, so we just return the first input if is_resume: @@ -1552,7 +1554,7 @@ def _write_component_outputs( # This allows us to track if a predecessor already ran but did not produce an output. value = component_outputs.get(sender_socket.name, _NO_OUTPUT_PRODUCED) - if not isinstance(value, _Empty) and conversion_strategy: + if not isinstance(value, _NoOutputProduced) and conversion_strategy: try: value = _convert_value(value=value, conversion_strategy=conversion_strategy) except Exception as e: @@ -1862,5 +1864,5 @@ def _write_to_standard_socket( current_value = inputs[receiver_name].get(receiver_socket_name) # Only overwrite if there's no existing value, or we have a new value to provide - if current_value is None or not isinstance(value, _Empty): + if current_value is None or not isinstance(value, _NoOutputProduced): inputs[receiver_name][receiver_socket_name] = [{"sender": component_name, "value": value}] diff --git a/haystack/core/pipeline/component_checks.py b/haystack/core/pipeline/component_checks.py index 95d10037191..0c022fd3652 100644 --- a/haystack/core/pipeline/component_checks.py +++ b/haystack/core/pipeline/component_checks.py @@ -4,10 +4,45 @@ from typing import Any -from haystack.core.component.types import InputSocket, _Empty, _empty +from haystack.core.component.types import InputSocket -# A socket input holding this sentinel records that its sender ran without producing a value for that socket. -_NO_OUTPUT_PRODUCED = _empty + +class _NoOutputProduced: + """ + Type of the `_NO_OUTPUT_PRODUCED` sentinel. + + A socket input holds the sentinel when its sender ran without producing a value for that socket. + + Test for it with `isinstance(value, _NoOutputProduced)`. The sentinel carries no state, so every instance means + the same thing, and unlike `==` this never runs `__eq__` on the value being checked, which for a value such as a + numpy array does not return a bool. + + The sentinel reaches a pipeline snapshot as part of the pipeline's inputs, so it has to serialize and to survive + copying as the single `_NO_OUTPUT_PRODUCED` instance. + """ + + def __repr__(self) -> str: + return "_NO_OUTPUT_PRODUCED" + + def __reduce__(self) -> str: + # Returning the global's name keeps `copy`, `deepcopy` and `pickle` from building a second instance. + return "_NO_OUTPUT_PRODUCED" + + def to_dict(self) -> dict[str, Any]: + """ + Serialize the sentinel. It carries no state, so the payload is empty. + """ + return {} + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "_NoOutputProduced": # noqa: ARG003 + """ + Deserialize the sentinel back to the singleton. + """ + return _NO_OUTPUT_PRODUCED + + +_NO_OUTPUT_PRODUCED = _NoOutputProduced() def can_component_run(component: dict, inputs: dict) -> bool: @@ -104,7 +139,7 @@ def any_socket_value_from_predecessor_received(socket_inputs: list[dict[str, Any :param socket_inputs: Inputs for the component's socket. """ # When sender is None, the input was provided from outside the pipeline. - return any(not isinstance(inp["value"], _Empty) and inp["sender"] is not None for inp in socket_inputs) + return any(not isinstance(inp["value"], _NoOutputProduced) and inp["sender"] is not None for inp in socket_inputs) def has_user_input(inputs: dict) -> bool: @@ -143,7 +178,7 @@ def any_socket_input_received(socket_inputs: list[dict]) -> bool: :param socket_inputs: Inputs for the socket. """ - return any(not isinstance(inp["value"], _Empty) for inp in socket_inputs) + return any(not isinstance(inp["value"], _NoOutputProduced) for inp in socket_inputs) def has_lazy_variadic_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict]) -> bool: @@ -155,7 +190,9 @@ def has_lazy_variadic_socket_received_all_inputs(socket: InputSocket, socket_inp """ expected_senders = set(socket.senders) actual_senders = { - sock["sender"] for sock in socket_inputs if not isinstance(sock["value"], _Empty) and sock["sender"] is not None + sock["sender"] + for sock in socket_inputs + if not isinstance(sock["value"], _NoOutputProduced) and sock["sender"] is not None } return expected_senders == actual_senders @@ -172,7 +209,11 @@ def has_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict return False # The socket is greedy variadic and at least one input was produced, it is complete. - if socket.is_variadic and socket.is_greedy and any(not isinstance(sock["value"], _Empty) for sock in socket_inputs): + if ( + socket.is_variadic + and socket.is_greedy + and any(not isinstance(sock["value"], _NoOutputProduced) for sock in socket_inputs) + ): return True # The socket is lazy variadic and all expected inputs were produced. @@ -180,7 +221,7 @@ def has_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict return True # The socket is not variadic and the only expected input is complete. - return not socket.is_variadic and not isinstance(socket_inputs[0]["value"], _Empty) + return not socket.is_variadic and not isinstance(socket_inputs[0]["value"], _NoOutputProduced) def all_predecessors_executed(component: dict, inputs: dict) -> bool: diff --git a/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml index 65550166d7c..435adfe6f29 100644 --- a/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml +++ b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml @@ -40,8 +40,3 @@ fixes: receive a bare value where it expected a list, raising errors such as ``TypeError: object of type 'int' has no len()`` from a ``BranchJoiner``. This affected snapshots taken at any visit count, including the first. - - | - The ``_empty`` sentinel, which marks an unset ``InputSocket.default_value`` and a socket whose sender produced no - output, is now recognized by type rather than by identity or equality. Testing for it no longer runs ``__eq__`` - on the value being checked, which for values such as numpy arrays does not return a bool, and the sentinel now - survives ``copy``, ``deepcopy``, ``pickle``, and serialization as the same object. diff --git a/test/core/component/test_types.py b/test/core/component/test_types.py index cdfca35c0d8..b91dfe48854 100644 --- a/test/core/component/test_types.py +++ b/test/core/component/test_types.py @@ -2,52 +2,16 @@ # # SPDX-License-Identifier: Apache-2.0 -import copy -import pickle +from haystack.core.component.types import InputSocket -import pytest -from haystack.core.component.types import InputSocket, _empty -from haystack.core.pipeline.utils import _deepcopy_with_exceptions -from haystack.utils.base_serialization import _deserialize_value_with_schema, _serialize_value_with_schema - - -class _AlwaysUnequal: # noqa: PLW1641 # __hash__ is irrelevant; this only needs a hostile __eq__ +class _IncomparableValue: # noqa: PLW1641 # __hash__ is irrelevant; this only needs a hostile __eq__ """Stands in for values such as numpy arrays, whose `__eq__` does not return a bool.""" def __eq__(self, other): raise AssertionError("__eq__ must not be called when testing for the sentinel") -class TestEmptySentinel: - """`_empty` is only ever recognized as itself, so anything that could duplicate it has to hand it back.""" - - def test_repr_names_the_sentinel(self): - assert repr(_empty) == "_empty" - - @pytest.mark.parametrize( - "duplicate", - [copy.copy, copy.deepcopy, lambda v: pickle.loads(pickle.dumps(v))], - ids=["copy", "deepcopy", "pickle"], - ) - def test_duplicating_the_sentinel_returns_the_singleton(self, duplicate): - assert duplicate(_empty) is _empty - - def test_the_sentinel_survives_a_deepcopy_of_the_pipeline_inputs(self): - """`Pipeline.run` deep-copies its inputs, which carry the sentinel, before snapshotting them.""" - inputs = {"comp": {"socket": [{"sender": "other", "value": _empty}]}} - - copied = _deepcopy_with_exceptions(inputs) - - assert copied["comp"]["socket"][0]["value"] is _empty - - def test_the_sentinel_survives_serialization(self): - """A pipeline snapshot serializes the inputs, sentinel included.""" - restored = _deserialize_value_with_schema(_serialize_value_with_schema({"value": _empty})) - - assert restored["value"] is _empty - - def test_input_socket_with_an_incomparable_default_is_not_mandatory(): - """`is_mandatory` tests for the sentinel by type, so it never compares the default it was given.""" - assert not InputSocket(name="value", type=str, default_value=_AlwaysUnequal()).is_mandatory + """`is_mandatory` tests for the sentinel by identity, so it never compares the default it was given.""" + assert not InputSocket(name="value", type=str, default_value=_IncomparableValue()).is_mandatory diff --git a/test/core/pipeline/test_breakpoint.py b/test/core/pipeline/test_breakpoint.py index 3787e567b5b..457ec573e91 100644 --- a/test/core/pipeline/test_breakpoint.py +++ b/test/core/pipeline/test_breakpoint.py @@ -13,7 +13,6 @@ from haystack.components.joiners import BranchJoiner from haystack.components.routers import ConditionalRouter from haystack.components.routers.conditional_router import Route -from haystack.core.component.types import _empty from haystack.core.errors import BreakpointException, PipelineInvalidPipelineSnapshotError from haystack.core.pipeline import Pipeline from haystack.core.pipeline.breakpoint import ( @@ -25,6 +24,7 @@ _transform_json_structure, load_pipeline_snapshot, ) +from haystack.core.pipeline.component_checks import _NO_OUTPUT_PRODUCED from haystack.dataclasses import ChatMessage from haystack.dataclasses.breakpoints import INTERNAL_INPUTS_FORMAT, Breakpoint, PipelineSnapshot, PipelineState from haystack.utils import _deserialize_value_with_schema @@ -282,7 +282,7 @@ def test_snapshot_preserves_sockets_whose_sender_produced_no_output(self): assert snapshot is not None restored = _deserialize_internal_inputs(snapshot.pipeline_state.inputs) - assert restored["collect"]["b"] == [{"sender": "router", "value": _empty}] + assert restored["collect"]["b"] == [{"sender": "router", "value": _NO_OUTPUT_PRODUCED}] assert pipeline.run(data={}, pipeline_snapshot=snapshot) == expected diff --git a/test/core/pipeline/test_component_checks.py b/test/core/pipeline/test_component_checks.py index 1b2ba07faab..00c30bb2dd0 100644 --- a/test/core/pipeline/test_component_checks.py +++ b/test/core/pipeline/test_component_checks.py @@ -2,14 +2,17 @@ # # SPDX-License-Identifier: Apache-2.0 +import copy +import pickle from typing import Any import pytest from pandas import DataFrame -from haystack.core.component.types import GreedyVariadic, InputSocket, OutputSocket, Variadic, _Empty +from haystack.core.component.types import GreedyVariadic, InputSocket, OutputSocket, Variadic, _empty from haystack.core.pipeline.component_checks import ( _NO_OUTPUT_PRODUCED, + _NoOutputProduced, all_predecessors_executed, all_socket_predecessors_executed, any_predecessors_provided_input, @@ -24,13 +27,8 @@ has_user_input, is_any_greedy_socket_ready, ) - - -class _IncomparableValue: # noqa: PLW1641 # __hash__ is irrelevant; this only needs a hostile __eq__ - """Stands in for values such as numpy arrays, whose `__eq__` does not return a bool.""" - - def __eq__(self, other): - raise AssertionError("__eq__ must not be called when testing for the sentinel") +from haystack.core.pipeline.utils import _deepcopy_with_exceptions +from haystack.utils.base_serialization import _deserialize_value_with_schema, _serialize_value_with_schema @pytest.fixture @@ -155,8 +153,8 @@ def test_component_missing_mandatory_input(self, basic_component): assert can_component_run(basic_component, inputs) is False # We added these tests because a component that returned a pandas dataframe caused the pipeline to fail. - # Previously, we compared the value of the socket using '!=' which leads to an error with dataframes. - # Instead, we use 'is not' to compare with the sentinel value. + # Comparing the socket value with '!=' errors out on dataframes, so the checks test the sentinel by type + # instead, which never compares the value at all. def test_sockets_with_ambiguous_truth_value(self, basic_component, greedy_variadic_socket, regular_socket): inputs = {"mandatory_input": [{"sender": "previous_component", "value": DataFrame.from_dict([{"value": 42}])}]} @@ -486,13 +484,42 @@ def test_any_socket_input_received_empty_list(self): def test_any_socket_input_received_with_a_separate_sentinel_instance(self): """The sentinel is recognized by type, so a second instance means no output just as the singleton does.""" - socket_inputs = [{"sender": "component1", "value": _Empty()}] + socket_inputs = [{"sender": "component1", "value": _NoOutputProduced()}] assert any_socket_input_received(socket_inputs) is False - def test_any_socket_input_received_with_an_incomparable_value(self): - """Recognizing the sentinel by type means a value that cannot be compared is still counted as received.""" - socket_inputs = [{"sender": "component1", "value": _IncomparableValue()}] - assert any_socket_input_received(socket_inputs) is True + +class TestNoOutputProducedSentinel: + """The sentinel reaches a pipeline snapshot, so anything that could duplicate it has to hand it back.""" + + def test_repr_names_the_sentinel(self): + assert repr(_NO_OUTPUT_PRODUCED) == "_NO_OUTPUT_PRODUCED" + + def test_the_sentinel_is_distinct_from_the_unset_default_marker(self): + """`_empty` marks an `InputSocket` without a default; it is a different thing and never reaches the inputs.""" + assert _NO_OUTPUT_PRODUCED is not _empty + assert not isinstance(_empty, _NoOutputProduced) + + @pytest.mark.parametrize( + "duplicate", + [copy.copy, copy.deepcopy, lambda value: pickle.loads(pickle.dumps(value))], + ids=["copy", "deepcopy", "pickle"], + ) + def test_duplicating_the_sentinel_returns_the_singleton(self, duplicate): + assert duplicate(_NO_OUTPUT_PRODUCED) is _NO_OUTPUT_PRODUCED + + def test_the_sentinel_survives_a_deepcopy_of_the_pipeline_inputs(self): + """`Pipeline.run` deep-copies its inputs, which carry the sentinel, before snapshotting them.""" + inputs = {"comp": {"socket": [{"sender": "other", "value": _NO_OUTPUT_PRODUCED}]}} + + copied = _deepcopy_with_exceptions(inputs) + + assert copied["comp"]["socket"][0]["value"] is _NO_OUTPUT_PRODUCED + + def test_the_sentinel_survives_serialization(self): + """A pipeline snapshot serializes the inputs, sentinel included.""" + restored = _deserialize_value_with_schema(_serialize_value_with_schema({"value": _NO_OUTPUT_PRODUCED})) + + assert restored["value"] is _NO_OUTPUT_PRODUCED class TestLazyVariadicSocket: From eb503725bbebcd93c9e87f7e228207035fe1b108 Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 13:10:22 +0200 Subject: [PATCH 07/13] simplify --- haystack/core/component/types.py | 3 +- haystack/core/pipeline/base.py | 8 +- haystack/core/pipeline/breakpoint.py | 2 +- haystack/core/pipeline/component_checks.py | 32 ++++---- test/components/agents/test_agent.py | 2 +- test/core/component/test_types.py | 18 +++-- test/core/pipeline/test_breakpoint.py | 6 +- test/core/pipeline/test_component_checks.py | 85 +++++++++------------ test/core/pipeline/test_pipeline_base.py | 19 ++--- 9 files changed, 81 insertions(+), 94 deletions(-) diff --git a/haystack/core/component/types.py b/haystack/core/component/types.py index d701a8076de..101ba682705 100644 --- a/haystack/core/component/types.py +++ b/haystack/core/component/types.py @@ -73,8 +73,7 @@ def is_variadic(self) -> bool: @property def is_mandatory(self) -> bool: """Check if the input is mandatory.""" - # Compared by identity so that a default value with a custom `__eq__`, such as a numpy array, is not asked - # whether it equals the sentinel. + # Identity, so a default with a custom `__eq__`, such as a DataFrame, is never compared to the sentinel. return self.default_value is _empty def __post_init__(self) -> None: diff --git a/haystack/core/pipeline/base.py b/haystack/core/pipeline/base.py index 9e21c3ef637..f2e24f96065 100644 --- a/haystack/core/pipeline/base.py +++ b/haystack/core/pipeline/base.py @@ -27,7 +27,6 @@ PipelineValidationError, ) from haystack.core.pipeline.component_checks import ( - _NO_OUTPUT_PRODUCED, _NoOutputProduced, all_predecessors_executed, are_all_sockets_ready, @@ -1549,10 +1548,10 @@ def _write_component_outputs( :param include_outputs_from: Set of component names that should always return an output from the pipeline. """ for receiver_name, sender_socket, receiver_socket, conversion_strategy in receivers: - # We either get the value that was produced by the actor or we use the _NO_OUTPUT_PRODUCED class to indicate + # We either get the value that was produced by the actor or we use a _NoOutputProduced marker to indicate # that the sender did not produce an output for this socket. # This allows us to track if a predecessor already ran but did not produce an output. - value = component_outputs.get(sender_socket.name, _NO_OUTPUT_PRODUCED) + value = component_outputs.get(sender_socket.name, _NoOutputProduced()) if not isinstance(value, _NoOutputProduced) and conversion_strategy: try: @@ -1591,7 +1590,8 @@ def _write_component_outputs( ) else: # If the receiver socket is not lazy variadic, it is greedy variadic or non-variadic. - # We overwrite with the new input if it's not _NO_OUTPUT_PRODUCED or if the current value is None. + # We overwrite with the new input if it's not a _NoOutputProduced marker, or if the current value + # is None. _write_to_standard_socket( inputs=inputs, receiver_name=receiver_name, diff --git a/haystack/core/pipeline/breakpoint.py b/haystack/core/pipeline/breakpoint.py index e3afd50c85e..16095981aa8 100644 --- a/haystack/core/pipeline/breakpoint.py +++ b/haystack/core/pipeline/breakpoint.py @@ -221,7 +221,7 @@ def _serialize_internal_inputs(inputs: dict[str, Any]) -> dict[str, Any]: The inputs a socket received are stored keyed by position rather than as a list, because a list gets a single schema derived from its first item. A socket with several senders can hold values of different types, for - example a value from one sender and the `_NO_OUTPUT_PRODUCED` sentinel from another, and those need one schema + example a value from one sender and a `_NoOutputProduced` marker from another, and those need one schema each to survive the round trip. :param inputs: The pipeline's internal inputs state. diff --git a/haystack/core/pipeline/component_checks.py b/haystack/core/pipeline/component_checks.py index 0c022fd3652..d4b32a76315 100644 --- a/haystack/core/pipeline/component_checks.py +++ b/haystack/core/pipeline/component_checks.py @@ -9,40 +9,36 @@ class _NoOutputProduced: """ - Type of the `_NO_OUTPUT_PRODUCED` sentinel. + Marker a socket input holds when its sender ran without producing a value for that socket. - A socket input holds the sentinel when its sender ran without producing a value for that socket. + Test for it with `isinstance(value, _NoOutputProduced)`, never with `==`: comparing runs `__eq__` on the value + being checked, which for a value such as a DataFrame does not return a bool. It carries no state, so instances + are interchangeable and every one of them means the same thing. - Test for it with `isinstance(value, _NoOutputProduced)`. The sentinel carries no state, so every instance means - the same thing, and unlike `==` this never runs `__eq__` on the value being checked, which for a value such as a - numpy array does not return a bool. - - The sentinel reaches a pipeline snapshot as part of the pipeline's inputs, so it has to serialize and to survive - copying as the single `_NO_OUTPUT_PRODUCED` instance. + It is serializable because it reaches a pipeline snapshot as part of the pipeline's inputs. """ def __repr__(self) -> str: - return "_NO_OUTPUT_PRODUCED" + return "_NoOutputProduced()" + + def __eq__(self, other: object) -> bool: + return isinstance(other, _NoOutputProduced) - def __reduce__(self) -> str: - # Returning the global's name keeps `copy`, `deepcopy` and `pickle` from building a second instance. - return "_NO_OUTPUT_PRODUCED" + def __hash__(self) -> int: + return hash(_NoOutputProduced) def to_dict(self) -> dict[str, Any]: """ - Serialize the sentinel. It carries no state, so the payload is empty. + Serialize the marker. It carries no state, so the payload is empty. """ return {} @classmethod def from_dict(cls, data: dict[str, Any]) -> "_NoOutputProduced": # noqa: ARG003 """ - Deserialize the sentinel back to the singleton. + Deserialize the marker. """ - return _NO_OUTPUT_PRODUCED - - -_NO_OUTPUT_PRODUCED = _NoOutputProduced() + return cls() def can_component_run(component: dict, inputs: dict) -> bool: diff --git a/test/components/agents/test_agent.py b/test/components/agents/test_agent.py index afca9ca070a..aaacacc29e3 100644 --- a/test/components/agents/test_agent.py +++ b/test/components/agents/test_agent.py @@ -2259,7 +2259,7 @@ class FilesProcessor: @component.output_types(processed_files=list[str]) def run(self, files: list[str]) -> dict: if not files: - return {} # _NO_OUTPUT_PRODUCED → blocks AttachmentsBuilder + return {} # _NoOutputProduced → blocks AttachmentsBuilder return {"processed_files": files} @component diff --git a/test/core/component/test_types.py b/test/core/component/test_types.py index b91dfe48854..d0fa5dc5026 100644 --- a/test/core/component/test_types.py +++ b/test/core/component/test_types.py @@ -2,16 +2,18 @@ # # SPDX-License-Identifier: Apache-2.0 -from haystack.core.component.types import InputSocket +from pandas import DataFrame +from haystack.core.component.types import InputSocket -class _IncomparableValue: # noqa: PLW1641 # __hash__ is irrelevant; this only needs a hostile __eq__ - """Stands in for values such as numpy arrays, whose `__eq__` does not return a bool.""" - def __eq__(self, other): - raise AssertionError("__eq__ must not be called when testing for the sentinel") +def test_input_socket_with_a_default_that_cannot_be_compared_is_not_mandatory(): + """ + `is_mandatory` tests for the sentinel by identity, so it never compares the default it was given. + Comparing with '==' returns a DataFrame rather than a bool here, which makes `is_mandatory` unusable as a + condition. + """ + socket = InputSocket(name="value", type=DataFrame, default_value=DataFrame.from_dict([{"value": 42}])) -def test_input_socket_with_an_incomparable_default_is_not_mandatory(): - """`is_mandatory` tests for the sentinel by identity, so it never compares the default it was given.""" - assert not InputSocket(name="value", type=str, default_value=_IncomparableValue()).is_mandatory + assert socket.is_mandatory is False diff --git a/test/core/pipeline/test_breakpoint.py b/test/core/pipeline/test_breakpoint.py index 457ec573e91..4121e5e84d8 100644 --- a/test/core/pipeline/test_breakpoint.py +++ b/test/core/pipeline/test_breakpoint.py @@ -24,7 +24,7 @@ _transform_json_structure, load_pipeline_snapshot, ) -from haystack.core.pipeline.component_checks import _NO_OUTPUT_PRODUCED +from haystack.core.pipeline.component_checks import _NoOutputProduced from haystack.dataclasses import ChatMessage from haystack.dataclasses.breakpoints import INTERNAL_INPUTS_FORMAT, Breakpoint, PipelineSnapshot, PipelineState from haystack.utils import _deserialize_value_with_schema @@ -263,7 +263,7 @@ def test_snapshot_preserves_the_sender_of_each_input(self): assert restored["comp2"] == {"input_value": [{"sender": "comp1", "value": "test_processed"}]} def test_snapshot_preserves_sockets_whose_sender_produced_no_output(self): - """A router that leaves a branch inactive puts the `_NO_OUTPUT_PRODUCED` sentinel in the pipeline inputs.""" + """A router that leaves a branch inactive puts the `_NoOutputProduced()` sentinel in the pipeline inputs.""" routes: list[Route] = [ {"condition": "{{ n > 5 }}", "output": "{{ 'big' }}", "output_name": "big", "output_type": str}, {"condition": "{{ n <= 5 }}", "output": "{{ 'small' }}", "output_name": "small", "output_type": str}, @@ -282,7 +282,7 @@ def test_snapshot_preserves_sockets_whose_sender_produced_no_output(self): assert snapshot is not None restored = _deserialize_internal_inputs(snapshot.pipeline_state.inputs) - assert restored["collect"]["b"] == [{"sender": "router", "value": _NO_OUTPUT_PRODUCED}] + assert restored["collect"]["b"] == [{"sender": "router", "value": _NoOutputProduced()}] assert pipeline.run(data={}, pipeline_snapshot=snapshot) == expected diff --git a/test/core/pipeline/test_component_checks.py b/test/core/pipeline/test_component_checks.py index 00c30bb2dd0..08bea0f48bc 100644 --- a/test/core/pipeline/test_component_checks.py +++ b/test/core/pipeline/test_component_checks.py @@ -2,8 +2,6 @@ # # SPDX-License-Identifier: Apache-2.0 -import copy -import pickle from typing import Any import pytest @@ -11,7 +9,6 @@ from haystack.core.component.types import GreedyVariadic, InputSocket, OutputSocket, Variadic, _empty from haystack.core.pipeline.component_checks import ( - _NO_OUTPUT_PRODUCED, _NoOutputProduced, all_predecessors_executed, all_socket_predecessors_executed, @@ -327,10 +324,10 @@ def test_any_predecessors_provided_input_no_predecessor(self, component_with_mul def test_any_predecessors_provided_input_with_no_output(self, component_with_multiple_sockets): """ - Ensures that _NO_OUTPUT_PRODUCED from a predecessor is ignored in the predecessor detection. + Ensures that _NoOutputProduced() from a predecessor is ignored in the predecessor detection. """ inputs = { - "socket1": [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}], + "socket1": [{"sender": "component1", "value": _NoOutputProduced()}], "socket2": [{"sender": None, "value": "test"}], } assert any_predecessors_provided_input(component_with_multiple_sockets, inputs) is False @@ -350,12 +347,12 @@ class TestSocketValueFromPredecessor: "socket_inputs, expected_result", [ pytest.param([{"sender": "component1", "value": 42}], True, id="valid_input"), - pytest.param([{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}], False, id="no_output"), + pytest.param([{"sender": "component1", "value": _NoOutputProduced()}], False, id="no_output"), pytest.param([{"sender": None, "value": 42}], False, id="user_input"), pytest.param( [ {"sender": None, "value": 42}, - {"sender": "component1", "value": _NO_OUTPUT_PRODUCED}, + {"sender": "component1", "value": _NoOutputProduced()}, {"sender": "component2", "value": 100}, ], True, @@ -392,10 +389,10 @@ def test_has_user_input_empty_inputs(self): def test_has_user_input_with_no_output(self): """ - Even if the input value is _NO_OUTPUT_PRODUCED, if sender=None + Even if the input value is _NoOutputProduced(), if sender=None it still counts as user input being provided. """ - inputs = {"socket1": [{"sender": None, "value": _NO_OUTPUT_PRODUCED}]} + inputs = {"socket1": [{"sender": None, "value": _NoOutputProduced()}]} assert has_user_input(inputs) is True @@ -464,62 +461,54 @@ def test_variadic_socket_no_execution(self, variadic_socket_with_senders): class TestSocketInputReceived: def test_any_socket_input_received_with_value(self): - """Checks that if there's a non-_NO_OUTPUT_PRODUCED value, the socket is marked as having input.""" + """Checks that if there's a non-_NoOutputProduced() value, the socket is marked as having input.""" socket_inputs = [{"sender": "component1", "value": 42}] assert any_socket_input_received(socket_inputs) is True def test_any_socket_input_received_with_no_output(self): - """If all inputs are _NO_OUTPUT_PRODUCED, the socket has no effective input.""" - socket_inputs = [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}] + """If all inputs are _NoOutputProduced(), the socket has no effective input.""" + socket_inputs = [{"sender": "component1", "value": _NoOutputProduced()}] assert any_socket_input_received(socket_inputs) is False def test_any_socket_input_received_mixed_inputs(self): """A single valid input among many is enough to consider the socket as having input.""" - socket_inputs = [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}, {"sender": "component2", "value": 42}] + socket_inputs = [{"sender": "component1", "value": _NoOutputProduced()}, {"sender": "component2", "value": 42}] assert any_socket_input_received(socket_inputs) is True def test_any_socket_input_received_empty_list(self): """Empty list: no input received.""" assert any_socket_input_received([]) is False - def test_any_socket_input_received_with_a_separate_sentinel_instance(self): - """The sentinel is recognized by type, so a second instance means no output just as the singleton does.""" - socket_inputs = [{"sender": "component1", "value": _NoOutputProduced()}] - assert any_socket_input_received(socket_inputs) is False +class TestNoOutputProduced: + """The marker reaches a pipeline snapshot, so it has to survive being copied and serialized.""" -class TestNoOutputProducedSentinel: - """The sentinel reaches a pipeline snapshot, so anything that could duplicate it has to hand it back.""" + def test_repr_names_the_marker(self): + assert repr(_NoOutputProduced()) == "_NoOutputProduced()" - def test_repr_names_the_sentinel(self): - assert repr(_NO_OUTPUT_PRODUCED) == "_NO_OUTPUT_PRODUCED" + def test_markers_are_interchangeable(self): + """The marker carries no state, so any two of them are the same value.""" + assert _NoOutputProduced() == _NoOutputProduced() + assert len({_NoOutputProduced(), _NoOutputProduced()}) == 1 - def test_the_sentinel_is_distinct_from_the_unset_default_marker(self): + def test_the_marker_is_not_the_unset_default_marker(self): """`_empty` marks an `InputSocket` without a default; it is a different thing and never reaches the inputs.""" - assert _NO_OUTPUT_PRODUCED is not _empty assert not isinstance(_empty, _NoOutputProduced) + assert _NoOutputProduced() != _empty - @pytest.mark.parametrize( - "duplicate", - [copy.copy, copy.deepcopy, lambda value: pickle.loads(pickle.dumps(value))], - ids=["copy", "deepcopy", "pickle"], - ) - def test_duplicating_the_sentinel_returns_the_singleton(self, duplicate): - assert duplicate(_NO_OUTPUT_PRODUCED) is _NO_OUTPUT_PRODUCED - - def test_the_sentinel_survives_a_deepcopy_of_the_pipeline_inputs(self): - """`Pipeline.run` deep-copies its inputs, which carry the sentinel, before snapshotting them.""" - inputs = {"comp": {"socket": [{"sender": "other", "value": _NO_OUTPUT_PRODUCED}]}} + def test_the_marker_survives_a_deepcopy_of_the_pipeline_inputs(self): + """`Pipeline.run` deep-copies its inputs, which carry the marker, before snapshotting them.""" + inputs = {"comp": {"socket": [{"sender": "other", "value": _NoOutputProduced()}]}} copied = _deepcopy_with_exceptions(inputs) - assert copied["comp"]["socket"][0]["value"] is _NO_OUTPUT_PRODUCED + assert isinstance(copied["comp"]["socket"][0]["value"], _NoOutputProduced) - def test_the_sentinel_survives_serialization(self): + def test_the_marker_survives_serialization(self): """A pipeline snapshot serializes the inputs, sentinel included.""" - restored = _deserialize_value_with_schema(_serialize_value_with_schema({"value": _NO_OUTPUT_PRODUCED})) + restored = _deserialize_value_with_schema(_serialize_value_with_schema({"value": _NoOutputProduced()})) - assert restored["value"] is _NO_OUTPUT_PRODUCED + assert isinstance(restored["value"], _NoOutputProduced) class TestLazyVariadicSocket: @@ -534,8 +523,8 @@ def test_lazy_variadic_partial_inputs(self, variadic_socket_with_senders): assert has_lazy_variadic_socket_received_all_inputs(variadic_socket_with_senders, socket_inputs) is False def test_lazy_variadic_with_no_output(self, variadic_socket_with_senders): - """_NO_OUTPUT_PRODUCED from a sender doesn't count as valid input, so it's not fully received.""" - socket_inputs = [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}, {"sender": "component2", "value": 42}] + """_NoOutputProduced() from a sender doesn't count as valid input, so it's not fully received.""" + socket_inputs = [{"sender": "component1", "value": _NoOutputProduced()}, {"sender": "component2", "value": 42}] assert has_lazy_variadic_socket_received_all_inputs(variadic_socket_with_senders, socket_inputs) is False def test_lazy_variadic_with_user_input(self, variadic_socket_with_senders): @@ -579,8 +568,8 @@ def test_regular_socket_complete(self, regular_socket): assert has_socket_received_all_inputs(regular_socket, inputs) is True def test_regular_socket_incomplete(self, regular_socket): - """_NO_OUTPUT_PRODUCED means the socket is not complete.""" - inputs = [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}] + """_NoOutputProduced() means the socket is not complete.""" + inputs = [{"sender": "component1", "value": _NoOutputProduced()}] assert has_socket_received_all_inputs(regular_socket, inputs) is False def test_regular_socket_no_inputs(self, regular_socket): @@ -598,8 +587,8 @@ def test_lazy_variadic_socket_partial_inputs(self, lazy_variadic_socket): assert has_socket_received_all_inputs(lazy_variadic_socket, inputs) is False def test_lazy_variadic_socket_with_no_output(self, lazy_variadic_socket): - """A sender that produces _NO_OUTPUT_PRODUCED does not fulfill the lazy socket requirement.""" - inputs = [{"sender": "component1", "value": 42}, {"sender": "component2", "value": _NO_OUTPUT_PRODUCED}] + """A sender that produces _NoOutputProduced() does not fulfill the lazy socket requirement.""" + inputs = [{"sender": "component1", "value": 42}, {"sender": "component2", "value": _NoOutputProduced()}] assert has_socket_received_all_inputs(lazy_variadic_socket, inputs) is False def test_greedy_variadic_socket_one_input(self, greedy_variadic_socket): @@ -613,8 +602,8 @@ def test_greedy_variadic_socket_multiple_inputs(self, greedy_variadic_socket): assert has_socket_received_all_inputs(greedy_variadic_socket, inputs) is True def test_greedy_variadic_socket_no_valid_inputs(self, greedy_variadic_socket): - """All _NO_OUTPUT_PRODUCED means the greedy socket is not complete.""" - inputs = [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}] + """All _NoOutputProduced() means the greedy socket is not complete.""" + inputs = [{"sender": "component1", "value": _NoOutputProduced()}] assert has_socket_received_all_inputs(greedy_variadic_socket, inputs) is False @@ -673,8 +662,8 @@ def test_greedy_socket_multiple_inputs_ready(self, complex_component): assert is_any_greedy_socket_ready(complex_component, inputs) is True def test_greedy_socket_not_ready(self, complex_component): - """If the only input is _NO_OUTPUT_PRODUCED, the greedy socket isn't ready.""" - inputs = {"greedy_var": [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}]} + """If the only input is _NoOutputProduced(), the greedy socket isn't ready.""" + inputs = {"greedy_var": [{"sender": "component1", "value": _NoOutputProduced()}]} assert is_any_greedy_socket_ready(complex_component, inputs) is False def test_greedy_socket_no_inputs(self, complex_component): diff --git a/test/core/pipeline/test_pipeline_base.py b/test/core/pipeline/test_pipeline_base.py index 6387df0b534..df63e1ebc67 100644 --- a/test/core/pipeline/test_pipeline_base.py +++ b/test/core/pipeline/test_pipeline_base.py @@ -21,7 +21,8 @@ PipelineRuntimeError, ) from haystack.core.pipeline import Pipeline -from haystack.core.pipeline.base import _NO_OUTPUT_PRODUCED, ComponentPriority, PipelineBase +from haystack.core.pipeline.base import ComponentPriority, PipelineBase +from haystack.core.pipeline.component_checks import _NoOutputProduced from haystack.core.pipeline.utils import FIFOPriorityQueue from haystack.core.serialization import DeserializationCallbacks from haystack.core.type_utils import ConversionStrategy @@ -887,7 +888,7 @@ def test__find_receivers_from(self): { "variadic_input": [ {"sender": "component1", "value": "test"}, - {"sender": "component2", "value": _NO_OUTPUT_PRODUCED}, + {"sender": "component2", "value": _NoOutputProduced()}, ], "normal_input": [{"sender": "component3", "value": "test"}], }, @@ -1050,7 +1051,7 @@ def test__write_component_outputs_output_pruning( @pytest.mark.parametrize( "output_value", - [42, None, _NO_OUTPUT_PRODUCED, "string_value", 3.14], + [42, None, _NoOutputProduced(), "string_value", 3.14], ids=["int", "none", "no-output", "string", "float"], ) def test__write_component_outputs_different_output_values( @@ -1071,9 +1072,9 @@ def test__write_component_outputs_different_output_values( assert inputs["receiver1"]["input1"] == [{"sender": "sender1", "value": output_value}] def test__write_component_outputs_dont_overwrite_with_no_output(self, regular_output_socket, regular_input_socket): - """Test that existing inputs are not overwritten with _NO_OUTPUT_PRODUCED""" + """Test that existing inputs are not overwritten with _NoOutputProduced()""" receivers = [("receiver1", regular_output_socket, regular_input_socket, None)] - component_outputs = {"output1": _NO_OUTPUT_PRODUCED} + component_outputs = {"output1": _NoOutputProduced()} inputs = {"receiver1": {"input1": [{"sender": "sender1", "value": "keep"}]}} PipelineBase()._write_component_outputs( component_name="sender1", @@ -1328,17 +1329,17 @@ def test_fill_queue(self, mock_get_metadata, mock_calc_priority): {"regular": 42, "greedy": [33], "lazy": [55, 66]}, {"regular": [{"sender": None, "value": 24}]}, # Only non-greedy user input remains ), - # Filtering _NO_OUTPUT_PRODUCED + # Filtering _NoOutputProduced() ( {"input1": InputSocket("input1", int)}, { "input1": [ - {"sender": "comp1", "value": _NO_OUTPUT_PRODUCED}, + {"sender": "comp1", "value": _NoOutputProduced()}, {"sender": "comp2", "value": 42}, - {"sender": "comp2", "value": _NO_OUTPUT_PRODUCED}, + {"sender": "comp2", "value": _NoOutputProduced()}, ] }, - {"input1": 42}, # Should skip _NO_OUTPUT_PRODUCED values + {"input1": 42}, # Should skip _NoOutputProduced() values {}, # All inputs consumed ), ], From 63bedbd32ea5449623f745df44b45c072a28f537 Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 13:12:35 +0200 Subject: [PATCH 08/13] simplify --- haystack/core/pipeline/component_checks.py | 3 --- test/core/pipeline/test_component_checks.py | 3 --- 2 files changed, 6 deletions(-) diff --git a/haystack/core/pipeline/component_checks.py b/haystack/core/pipeline/component_checks.py index d4b32a76315..3825aafc360 100644 --- a/haystack/core/pipeline/component_checks.py +++ b/haystack/core/pipeline/component_checks.py @@ -18,9 +18,6 @@ class _NoOutputProduced: It is serializable because it reaches a pipeline snapshot as part of the pipeline's inputs. """ - def __repr__(self) -> str: - return "_NoOutputProduced()" - def __eq__(self, other: object) -> bool: return isinstance(other, _NoOutputProduced) diff --git a/test/core/pipeline/test_component_checks.py b/test/core/pipeline/test_component_checks.py index 08bea0f48bc..928dd3d1a8b 100644 --- a/test/core/pipeline/test_component_checks.py +++ b/test/core/pipeline/test_component_checks.py @@ -483,9 +483,6 @@ def test_any_socket_input_received_empty_list(self): class TestNoOutputProduced: """The marker reaches a pipeline snapshot, so it has to survive being copied and serialized.""" - def test_repr_names_the_marker(self): - assert repr(_NoOutputProduced()) == "_NoOutputProduced()" - def test_markers_are_interchangeable(self): """The marker carries no state, so any two of them are the same value.""" assert _NoOutputProduced() == _NoOutputProduced() From ca3e3f2fb9bced57d12e0209708d1e58c8f40fc5 Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 13:31:06 +0200 Subject: [PATCH 09/13] simplify --- haystack/core/pipeline/pipeline.py | 23 ++++++++++----------- test/core/component/test_types.py | 7 +------ test/core/pipeline/test_component_checks.py | 20 +++--------------- 3 files changed, 15 insertions(+), 35 deletions(-) diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py index f1ec4880a62..83f80ba2a93 100644 --- a/haystack/core/pipeline/pipeline.py +++ b/haystack/core/pipeline/pipeline.py @@ -366,6 +366,8 @@ def run( # noqa: PLR0915, PLR0912, C901 # We track component visits to decide if a component can run. component_visits = dict.fromkeys(ordered_component_names, 0) + inputs = self._convert_to_internal_format(pipeline_inputs=data) + else: # Validate the pipeline snapshot against the current pipeline graph _validate_pipeline_snapshot_against_pipeline(pipeline_snapshot, self.graph) @@ -374,7 +376,15 @@ def run( # noqa: PLR0915, PLR0912, C901 component_visits = pipeline_snapshot.pipeline_state.component_visits ordered_component_names = pipeline_snapshot.ordered_component_names data = _deserialize_value_with_schema(pipeline_snapshot.original_input_data) - if pipeline_snapshot.pipeline_state.inputs_format != INTERNAL_INPUTS_FORMAT: + + if pipeline_snapshot.pipeline_state.inputs_format == INTERNAL_INPUTS_FORMAT: + inputs = _deserialize_internal_inputs(pipeline_snapshot.pipeline_state.inputs) + else: + # A legacy snapshot lost the sender of each input, so the only thing we can do is treat them as if + # they came from outside the pipeline. The paused component then has to be let through once. + inputs = self._convert_to_internal_format( + pipeline_inputs=_deserialize_value_with_schema(pipeline_snapshot.pipeline_state.inputs) + ) legacy_resume_component = pipeline_snapshot.break_point.component_name # include_outputs_from from the snapshot when resuming @@ -397,17 +407,6 @@ def run( # noqa: PLR0915, PLR0912, C901 "haystack.pipeline.execution_mode": "sync", }, ) as span: - if pipeline_snapshot is None: - inputs = self._convert_to_internal_format(pipeline_inputs=data) - elif legacy_resume_component is not None: - # Legacy snapshots stored flattened values, so the only thing we can do is treat them as if they - # came from outside the pipeline. - inputs = self._convert_to_internal_format( - pipeline_inputs=_deserialize_value_with_schema(pipeline_snapshot.pipeline_state.inputs) - ) - else: - inputs = _deserialize_internal_inputs(pipeline_snapshot.pipeline_state.inputs) - priority_queue = self._fill_queue(ordered_component_names, inputs, component_visits) # check if pipeline is blocked before execution diff --git a/test/core/component/test_types.py b/test/core/component/test_types.py index d0fa5dc5026..24f30fc23bd 100644 --- a/test/core/component/test_types.py +++ b/test/core/component/test_types.py @@ -8,12 +8,7 @@ def test_input_socket_with_a_default_that_cannot_be_compared_is_not_mandatory(): - """ - `is_mandatory` tests for the sentinel by identity, so it never compares the default it was given. - - Comparing with '==' returns a DataFrame rather than a bool here, which makes `is_mandatory` unusable as a - condition. - """ + """`is_mandatory` uses `is`, because comparing a DataFrame with `==` returns a DataFrame, not True or False.""" socket = InputSocket(name="value", type=DataFrame, default_value=DataFrame.from_dict([{"value": 42}])) assert socket.is_mandatory is False diff --git a/test/core/pipeline/test_component_checks.py b/test/core/pipeline/test_component_checks.py index 928dd3d1a8b..70ac5564e3b 100644 --- a/test/core/pipeline/test_component_checks.py +++ b/test/core/pipeline/test_component_checks.py @@ -7,7 +7,7 @@ import pytest from pandas import DataFrame -from haystack.core.component.types import GreedyVariadic, InputSocket, OutputSocket, Variadic, _empty +from haystack.core.component.types import GreedyVariadic, InputSocket, OutputSocket, Variadic from haystack.core.pipeline.component_checks import ( _NoOutputProduced, all_predecessors_executed, @@ -24,7 +24,6 @@ has_user_input, is_any_greedy_socket_ready, ) -from haystack.core.pipeline.utils import _deepcopy_with_exceptions from haystack.utils.base_serialization import _deserialize_value_with_schema, _serialize_value_with_schema @@ -481,28 +480,15 @@ def test_any_socket_input_received_empty_list(self): class TestNoOutputProduced: - """The marker reaches a pipeline snapshot, so it has to survive being copied and serialized.""" + """The marker reaches a pipeline snapshot, so it has to survive serialization.""" def test_markers_are_interchangeable(self): """The marker carries no state, so any two of them are the same value.""" assert _NoOutputProduced() == _NoOutputProduced() assert len({_NoOutputProduced(), _NoOutputProduced()}) == 1 - def test_the_marker_is_not_the_unset_default_marker(self): - """`_empty` marks an `InputSocket` without a default; it is a different thing and never reaches the inputs.""" - assert not isinstance(_empty, _NoOutputProduced) - assert _NoOutputProduced() != _empty - - def test_the_marker_survives_a_deepcopy_of_the_pipeline_inputs(self): - """`Pipeline.run` deep-copies its inputs, which carry the marker, before snapshotting them.""" - inputs = {"comp": {"socket": [{"sender": "other", "value": _NoOutputProduced()}]}} - - copied = _deepcopy_with_exceptions(inputs) - - assert isinstance(copied["comp"]["socket"][0]["value"], _NoOutputProduced) - def test_the_marker_survives_serialization(self): - """A pipeline snapshot serializes the inputs, sentinel included.""" + """A pipeline snapshot serializes the inputs, marker included.""" restored = _deserialize_value_with_schema(_serialize_value_with_schema({"value": _NoOutputProduced()})) assert isinstance(restored["value"], _NoOutputProduced) From 38c8d557e8330cb3d8b520ac2740526c50e41cb4 Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 13:57:41 +0200 Subject: [PATCH 10/13] update existing loop breakpoint test to test break points on second loop --- .../test_pipeline_breakpoints_loops.py | 123 ++++++++++++------ test/core/pipeline/test_breakpoint.py | 18 +-- 2 files changed, 83 insertions(+), 58 deletions(-) diff --git a/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py b/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py index 99b9f5d3091..844c14902bf 100644 --- a/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py +++ b/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py @@ -45,6 +45,53 @@ class CitiesData(BaseModel): cities: list[City] +VALID_CITIES_JSON = json.dumps( + { + "cities": [ + {"name": "Berlin", "country": "Germany", "population": 3850809}, + {"name": "Paris", "country": "France", "population": 2161000}, + {"name": "Lisbon", "country": "Portugal", "population": 504718}, + ] + } +) + + +def build_validation_loop_pipeline(llm: MockChatGenerator) -> Pipeline: + """Build a pipeline that sends invalid replies back to the prompt builder for another attempt.""" + prompt_template = [ + ChatMessage.from_user( + """ + Create a JSON object from the information present in this passage: {{passage}}. + Only use information that is present in the passage. Follow this JSON schema, but only return the + actual instances without any additional schema definition: + {{schema}} + Make sure your response is a dict and not a list. + {% if invalid_replies and error_message %} + You already created the following output in a previous attempt: {{invalid_replies}} + However, this doesn't comply with the format requirements from above and triggered this + Python exception: {{error_message}} + Correct the output and try again. Just return the corrected output without any extra explanations. + {% endif %} + """ + ) + ] + + pipeline = Pipeline(max_runs_per_component=5) + pipeline.add_component( + instance=ChatPromptBuilder(template=prompt_template, required_variables=["passage", "schema"]), + name="prompt_builder", + ) + pipeline.add_component(instance=llm, name="llm") + pipeline.add_component(instance=OutputValidator(pydantic_model=CitiesData), name="output_validator") + + pipeline.connect("prompt_builder.prompt", "llm.messages") + pipeline.connect("llm.replies", "output_validator") + pipeline.connect("output_validator.invalid_replies", "prompt_builder.invalid_replies") + pipeline.connect("output_validator.error_message", "prompt_builder.error_message") + + return pipeline + + class TestPipelineBreakpointsLoops: """ This class contains tests for pipelines with validation loops and breakpoints. @@ -52,49 +99,13 @@ class TestPipelineBreakpointsLoops: @pytest.fixture def validation_loop_pipeline(self): - """Create a pipeline with validation loops for testing.""" - prompt_template = [ - ChatMessage.from_user( - """ - Create a JSON object from the information present in this passage: {{passage}}. - Only use information that is present in the passage. Follow this JSON schema, but only return the - actual instances without any additional schema definition: - {{schema}} - Make sure your response is a dict and not a list. - {% if invalid_replies and error_message %} - You already created the following output in a previous attempt: {{invalid_replies}} - However, this doesn't comply with the format requirements from above and triggered this - Python exception: {{error_message}} - Correct the output and try again. Just return the corrected output without any extra explanations. - {% endif %} - """ - ) - ] + """The first reply already validates, so every component runs exactly once.""" + return build_validation_loop_pipeline(MockChatGenerator(VALID_CITIES_JSON)) - response_json = json.dumps( - { - "cities": [ - {"name": "Berlin", "country": "Germany", "population": 3850809}, - {"name": "Paris", "country": "France", "population": 2161000}, - {"name": "Lisbon", "country": "Portugal", "population": 504718}, - ] - } - ) - - pipeline = Pipeline(max_runs_per_component=5) - pipeline.add_component( - instance=ChatPromptBuilder(template=prompt_template, required_variables=["passage", "schema"]), - name="prompt_builder", - ) - pipeline.add_component(instance=MockChatGenerator(response_json), name="llm") - pipeline.add_component(instance=OutputValidator(pydantic_model=CitiesData), name="output_validator") - - pipeline.connect("prompt_builder.prompt", "llm.messages") - pipeline.connect("llm.replies", "output_validator") - pipeline.connect("output_validator.invalid_replies", "prompt_builder.invalid_replies") - pipeline.connect("output_validator.error_message", "prompt_builder.error_message") - - return pipeline + @pytest.fixture + def retrying_validation_loop_pipeline(self): + """The first reply fails validation, so the loop closes and every component runs twice.""" + return build_validation_loop_pipeline(MockChatGenerator(["this is not JSON", VALID_CITIES_JSON])) BREAKPOINT_COMPONENTS = ["prompt_builder", "llm", "output_validator"] @@ -134,3 +145,31 @@ 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" + + @pytest.mark.parametrize("component", BREAKPOINT_COMPONENTS, ids=BREAKPOINT_COMPONENTS) + @pytest.mark.integration + def test_pipeline_breakpoints_second_visit_of_validation_loop( + self, retrying_validation_loop_pipeline, output_directory, component, load_and_resume_pipeline_snapshot + ): + """ + Test that a pipeline paused on a component's second visit of a loop resumes and runs to completion. + + `pytest.raises` also asserts that the loop really closes: if it did not, the component would never reach a + second visit and the breakpoint would not trigger. + """ + data = {"prompt_builder": {"passage": "Berlin, Paris, Lisbon...", "schema": "CitiesData schema"}} + + break_point = Breakpoint(component_name=component, visit_count=1, snapshot_file_path=str(output_directory)) + + with pytest.raises(BreakpointException): + retrying_validation_loop_pipeline.run(data, break_point=break_point) + + result = load_and_resume_pipeline_snapshot( + pipeline=retrying_validation_loop_pipeline, + output_directory=output_directory, + component_name=break_point.component_name, + data=data, + ) + + cities_data = CitiesData.model_validate(json.loads(result["output_validator"]["valid_replies"][0].text)) + assert [city.name for city in cities_data.cities] == ["Berlin", "Paris", "Lisbon"] diff --git a/test/core/pipeline/test_breakpoint.py b/test/core/pipeline/test_breakpoint.py index 4121e5e84d8..64a7c51a41e 100644 --- a/test/core/pipeline/test_breakpoint.py +++ b/test/core/pipeline/test_breakpoint.py @@ -237,8 +237,6 @@ def test_break_point_matching_pipeline_snapshot_break_point_raises(self): @pytest.mark.parametrize("visit_count", [0, 1, 2, 3]) def test_break_point_in_loop_resumes_on_any_visit(self, visit_count): """A component paused on a later visit of a loop must still resume and finish the loop.""" - expected = _looping_pipeline().run({"joiner": {"value": 0}}) - with pytest.raises(BreakpointException) as exc_info: _looping_pipeline().run( {"joiner": {"value": 0}}, break_point=Breakpoint(component_name="joiner", visit_count=visit_count) @@ -247,20 +245,8 @@ def test_break_point_in_loop_resumes_on_any_visit(self, visit_count): assert snapshot is not None assert snapshot.pipeline_state.component_visits["joiner"] == visit_count - assert _looping_pipeline().run(data={}, pipeline_snapshot=snapshot) == expected - - def test_snapshot_preserves_the_sender_of_each_input(self): - pipeline = _three_component_pipeline() - - with pytest.raises(BreakpointException) as exc_info: - pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) - snapshot = exc_info.value.pipeline_snapshot - assert snapshot is not None - - assert snapshot.pipeline_state.inputs_format == INTERNAL_INPUTS_FORMAT - restored = _deserialize_internal_inputs(snapshot.pipeline_state.inputs) - # comp2's input came from comp1, not from outside the pipeline - assert restored["comp2"] == {"input_value": [{"sender": "comp1", "value": "test_processed"}]} + # The loop runs to `_CountUpTo(limit=5)` regardless of where it was paused. + assert _looping_pipeline().run(data={}, pipeline_snapshot=snapshot) == {"counter": {"done": "finished at 5"}} def test_snapshot_preserves_sockets_whose_sender_produced_no_output(self): """A router that leaves a branch inactive puts the `_NoOutputProduced()` sentinel in the pipeline inputs.""" From c9a9bb264414801e8fd9305768ca7dfd0a5aafbb Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 14:07:40 +0200 Subject: [PATCH 11/13] Add new test --- ...shot-resume-in-loops-0ab593c343026abb.yaml | 3 +- test/core/pipeline/test_breakpoint.py | 29 +++++++++++++++++++ 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml index 435adfe6f29..8c42a8d7685 100644 --- a/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml +++ b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml @@ -31,8 +31,7 @@ fixes: stored only the values of the pipeline's inputs and dropped the information about which component had sent each one, so on resume every restored input looked like it came from outside the pipeline, and such an input can only trigger a component on its first visit. Snapshots now record the sender of each input. Snapshots created by - earlier versions of Haystack can still be resumed on a first visit, as before; re-create them to resume from a - later visit of a loop. + earlier versions of Haystack behave as before, so re-create them to resume anywhere in a looping pipeline. - | Fixed a resumed ``Pipeline`` passing malformed inputs to the component the snapshot was taken on, whenever that component ran more than once after the resume, for example inside a loop. Every visit after the first reused the diff --git a/test/core/pipeline/test_breakpoint.py b/test/core/pipeline/test_breakpoint.py index 64a7c51a41e..66a709ded6f 100644 --- a/test/core/pipeline/test_breakpoint.py +++ b/test/core/pipeline/test_breakpoint.py @@ -272,6 +272,8 @@ def test_snapshot_preserves_sockets_whose_sender_produced_no_output(self): assert pipeline.run(data={}, pipeline_snapshot=snapshot) == expected + +class TestResumeFromLegacyPipelineSnapshot: def test_resume_from_legacy_snapshot_without_sender_information(self): """Snapshots taken before the sender was recorded store flattened values and must still resume.""" pipeline = _three_component_pipeline() @@ -292,6 +294,33 @@ def test_resume_from_legacy_snapshot_without_sender_information(self): result = pipeline.run(data={}, pipeline_snapshot=legacy_snapshot) assert result["comp3"]["result"] == "test_processed_processed_processed" + def test_resume_from_legacy_snapshot_into_a_loop(self): + """ + A legacy snapshot needs its special input handling for the visit it was paused on, and only that visit. + + The loop brings the paused component ordinary inputs again afterwards, which it has to consume the ordinary + way. Keeping the special handling re-reads the restored input on every visit, so the loop never advances. + + The snapshot is written out literally rather than derived from a current one, because that is what a snapshot + left over from an older Haystack looks like: a greedy socket stored the value it had already consumed. + """ + legacy_snapshot = PipelineSnapshot( + pipeline_state=PipelineState( + inputs=_serialize_value_with_schema({"joiner": {"value": [0]}, "counter": {}}), + component_visits={"joiner": 0, "counter": 0}, + pipeline_outputs=_serialize_value_with_schema({}), + inputs_format=None, + ), + break_point=Breakpoint(component_name="joiner", visit_count=0), + original_input_data=_serialize_value_with_schema({"joiner": {"value": 0}}), + ordered_component_names=["counter", "joiner"], + include_outputs_from=set(), + ) + + # The joiner is visited five more times after the resume. + result = _looping_pipeline().run(data={}, pipeline_snapshot=legacy_snapshot) + assert result == {"counter": {"done": "finished at 5"}} + class TestCreatePipelineSnapshot: def test_create_pipeline_snapshot_all_fields(self): From 11e2ab6b5d26a5de95ce30b4679839733318ecdf Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Mon, 27 Jul 2026 14:16:34 +0200 Subject: [PATCH 12/13] update reno --- ...ne-snapshot-resume-in-loops-0ab593c343026abb.yaml | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml index 8c42a8d7685..956dba579a4 100644 --- a/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml +++ b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml @@ -13,15 +13,21 @@ upgrade: to display or post-process a snapshot. Resuming a pipeline with ``Pipeline.run(pipeline_snapshot=...)`` is not affected, and neither is code that only passes snapshots around or persists them. - To adapt, read through the position and unwrap the sender: + To adapt, read the input through its position and take its ``value``: .. code-block:: python - from haystack.utils import _deserialize_value_with_schema + inputs = snapshot.pipeline_state.inputs["serialized_data"] - inputs = _deserialize_value_with_schema(snapshot.pipeline_state.inputs) + # before + value = inputs["my_component"]["my_socket"] + + # now value = inputs["my_component"]["my_socket"]["0"]["value"] + A socket that received inputs from several senders now has one entry per position, ``"0"``, ``"1"`` and so on, + each recording the ``sender`` that produced it. + Snapshots written by earlier versions of Haystack have ``inputs_format`` set to ``None`` and keep the flattened shape, so branch on that field if you need to handle both. fixes: From e37c7f3e26dc404a3003d655336604d073174427 Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Wed, 29 Jul 2026 07:55:47 +0200 Subject: [PATCH 13/13] improve test --- test/core/pipeline/test_breakpoint.py | 64 ++++++++++++++++----------- 1 file changed, 37 insertions(+), 27 deletions(-) diff --git a/test/core/pipeline/test_breakpoint.py b/test/core/pipeline/test_breakpoint.py index 66a709ded6f..09fdd6bd3fe 100644 --- a/test/core/pipeline/test_breakpoint.py +++ b/test/core/pipeline/test_breakpoint.py @@ -109,38 +109,48 @@ def run(self, input_value: str) -> dict[str, str]: # breakpoint on comp2 break_point = Breakpoint(component_name="comp2", visit_count=0, snapshot_file_path=str(tmp_path)) - try: - # run with include_outputs_from to capture intermediate outputs + # run with include_outputs_from to capture intermediate outputs + with pytest.raises(BreakpointException) as exc_info: pipeline.run(data={"comp1": {"input_value": "test"}}, include_outputs_from={"comp1"}, break_point=break_point) - except BreakpointException as e: - # breakpoint should be triggered - assert e.component == "comp2" - # verify snapshot file contains the intermediate outputs - snapshot_files = list(tmp_path.glob("comp2_*.json")) - assert len(snapshot_files) == 1, f"Expected exactly one snapshot file, found {len(snapshot_files)}" + # breakpoint should be triggered + assert exc_info.value.component == "comp2" - snapshot_file = snapshot_files[0] - loaded_snapshot = load_pipeline_snapshot(snapshot_file) + # verify snapshot file contains the intermediate outputs + snapshot_files = list(tmp_path.glob("comp2_*.json")) + assert len(snapshot_files) == 1, f"Expected exactly one snapshot file, found {len(snapshot_files)}" - # verify the snapshot contains the intermediate outputs from comp1 - assert loaded_snapshot.pipeline_state.pipeline_outputs == ( - { - "serialization_schema": { - "type": "object", - "properties": {"comp1": {"type": "object", "properties": {"result": {"type": "string"}}}}, - }, - "serialized_data": {"comp1": {"result": "processed_test"}}, - } - ) + snapshot_file = snapshot_files[0] + loaded_snapshot = load_pipeline_snapshot(snapshot_file) + + # verify the snapshot contains the intermediate outputs from comp1 + assert loaded_snapshot.pipeline_state.pipeline_outputs == ( + { + "serialization_schema": { + "type": "object", + "properties": {"comp1": {"type": "object", "properties": {"result": {"type": "string"}}}}, + }, + "serialized_data": {"comp1": {"result": "processed_test"}}, + } + ) + + # verify the saved inputs record which component sent each one, keyed by the position it arrived in. + # The accompanying schema is asserted in TestCreatePipelineSnapshot. + assert loaded_snapshot.pipeline_state.inputs_format == INTERNAL_INPUTS_FORMAT + assert loaded_snapshot.pipeline_state.inputs["serialized_data"] == { + # comp1 was given its input from outside the pipeline + "comp1": {"input_value": {"0": {"sender": None, "value": "test"}}}, + # comp2 was given its input by comp1, and had not consumed it yet when the breakpoint hit + "comp2": {"input_value": {"0": {"sender": "comp1", "value": "processed_test"}}}, + } - # verify the whole pipeline state contains the expected data - assert loaded_snapshot.pipeline_state.component_visits["comp1"] == 1 - assert loaded_snapshot.pipeline_state.component_visits["comp2"] == 0 - assert "comp1" in loaded_snapshot.include_outputs_from - assert isinstance(loaded_snapshot.break_point, Breakpoint) - assert loaded_snapshot.break_point.component_name == "comp2" - assert loaded_snapshot.break_point.visit_count == 0 + # verify the whole pipeline state contains the expected data + assert loaded_snapshot.pipeline_state.component_visits["comp1"] == 1 + assert loaded_snapshot.pipeline_state.component_visits["comp2"] == 0 + assert "comp1" in loaded_snapshot.include_outputs_from + assert isinstance(loaded_snapshot.break_point, Breakpoint) + assert loaded_snapshot.break_point.component_name == "comp2" + assert loaded_snapshot.break_point.visit_count == 0 @component