feat!: remove backward compatibility for legacy pipeline_outputs format

Remove temporary deserialization logic for old pipeline_outputs format in
pipeline snapshots. This backward-compatibility layer was introduced in
PR deepset-ai#10096 and marked for removal in Haystack 2.23.0.

BREAKING CHANGE: Pipeline snapshots created before Haystack 2.17.0 will
no longer deserialize correctly. Users must recreate snapshots with the
current Haystack version or manually migrate to the new serialization schema.

Changes:
- Remove conditional deserialization check in Pipeline.run()
- Remove test_load_pipeline_snapshot_with_old_pipeline_outputs_format test
- Add breaking change documentation with migration instructions

The new format uses 'serialization_schema' and 'serialized_data' fields,
consistent with other pipeline state serialization.

Closes deepset-ai#10168
Ref deepset-ai#10096

Author: GunaPalanivel

haystack/core/pipeline/pipeline.py

@@ -252,12 +252,7 @@ def run(  # noqa: PLR0915, PLR0912, C901, pylint: disable=too-many-branches
             include_outputs_from = pipeline_snapshot.include_outputs_from
 
             # also intermediate_outputs from the snapshot when resuming
-            # keep the deserialization of pipeline_outputs backwards compatible with the old pipeline_outputs format
-            # TODO: remove this in haystack 2.23.0
-            if "serialization_schema" not in pipeline_snapshot.pipeline_state.pipeline_outputs.keys():
-                pipeline_outputs = pipeline_snapshot.pipeline_state.pipeline_outputs
-            else:
-                pipeline_outputs = _deserialize_value_with_schema(pipeline_snapshot.pipeline_state.pipeline_outputs)
+            pipeline_outputs = _deserialize_value_with_schema(pipeline_snapshot.pipeline_state.pipeline_outputs)
 
         cached_topological_sort = None
         # We need to access a component's receivers multiple times during a pipeline run.

releasenotes/notes/remove-legacy-pipeline-outputs-deserialization-5acccb0245b84890.yaml

@@ -0,0 +1,8 @@
+---
+upgrade:
+  - |
+    Remove backward-compatibility support for deserializing pipeline snapshots with
+    the old ``pipeline_outputs`` format. Pipeline snapshots created before Haystack 2.22.0
+    that contain ``pipeline_outputs`` without the ``serialization_schema`` and ``serialized_data``
+    structure are no longer supported. Users should recreate their pipeline snapshots
+    with the current Haystack version before upgrading to 2.23.0.

test/core/pipeline/test_breakpoint.py

@@ -154,37 +154,6 @@ def run(self, input_value: str) -> dict[str, str]:
         assert loaded_snapshot.break_point.visit_count == 0
 
 
-def test_load_pipeline_snapshot_with_old_pipeline_outputs_format(tmp_path):
-    "Test to ensure backwards compatibility with the old pipeline_outputs format"
-    # TODO: remove this test in haystack 2.23.0
-    pipeline_snapshot = {
-        "pipeline_state": {
-            "inputs": {
-                "serialization_schema": {
-                    "type": "object",
-                    "properties": {"comp2": {"type": "object", "properties": {}}},
-                },
-                "serialized_data": {"comp2": {}},
-            },
-            "component_visits": {"comp1": 1, "comp2": 0},
-            "pipeline_outputs": {"comp1": {"result": "Answer from comp1"}},
-        },
-        "break_point": {"component_name": "comp2", "visit_count": 0, "snapshot_file_path": "test_breakpoints"},
-        "agent_snapshot": None,
-        "timestamp": "2025-12-01T17:14:24.366124",
-        "original_input_data": {"serialization_schema": {"type": "object", "properties": {}}, "serialized_data": {}},
-        "ordered_component_names": ["comp1", "comp2"],
-        "include_outputs_from": ["comp1"],
-    }
-
-    pipeline_snapshot_file = tmp_path / "old_pipeline_outputs_format.json"
-    with open(pipeline_snapshot_file, "w") as f:
-        json.dump(pipeline_snapshot, f)
-
-    loaded_snapshot = load_pipeline_snapshot(pipeline_snapshot_file)
-    assert loaded_snapshot == PipelineSnapshot.from_dict(pipeline_snapshot)
-
-
 def test_trigger_tool_invoker_breakpoint(make_pipeline_snapshot_with_agent_snapshot):
     pipeline_snapshot_with_agent_breakpoint = make_pipeline_snapshot_with_agent_snapshot(
         break_point=AgentBreakpoint("agent", ToolBreakpoint(component_name="tool_invoker"))