Related Issues

• fixes Resuming a pipeline snapshot after the first loop visit fails with PipelineComponentsBlockedError #12145

Proposed Changes:

Fixes

• 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 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 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 sentinel marking a socket whose sender produced no output for it is now its own type, _NoOutputProduced, rather than an alias of the _empty marker for an unset InputSocket.default_value. The two record unrelated things, and we needed to add serde support to _NoOutputProduced.

Breaking Change

• 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 code that only passes snapshots around or persists them.

To adapt, read the input through its position and take its value:

inputs = snapshot.pipeline_state.inputs["serialized_data"]

# 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.

How did you test it?

New tests and updated existing ones.

Notes for the reviewer

This has a breaking change associated with it in addition to being a fix. The inputs format has changed so anyone who relied on inspecting on the old format will get errors. More details in the reno.

Checklist

• I have read the contributors guidelines and the code of conduct.
• I have updated the related issue with new insights and changes.
• I have added unit tests and updated the docstrings.
• I've used one of the conventional commit types for my PR title: fix:, feat:, build:, chore:, ci:, docs:, style:, refactor:, perf:, test: and added ! in case the PR includes breaking changes.
• I have documented my code.
• I have added a release note file, following the contributors guidelines.
• I have run pre-commit hooks and fixed any issue.