Keep agent snapshots resumable when runtime-only inputs cannot be serialized

The fallback added for agent snapshot serialization errors preserved the
original runtime failure, but it could also replace the entire
chat_generator or tool_invoker payload with an empty dict. That made the
saved snapshot impossible to resume even when only a runtime-only field
such as a streaming callback was non-serializable.

This change narrows the fallback behavior: Haystack now retries those
component inputs field-by-field and omits only the fields that cannot be
serialized, preserving resumable fields like messages, state, and tools.
A regression test covers resuming from a tool-invoker snapshot created
with a non-serializable runtime callback.

Constraint: Must preserve the original deepset-ai#11108 goal of not masking the real runtime error
Rejected: Keep saving `{}` and document snapshots as non-resumable | breaks the existing resume contract more than necessary
Confidence: high
Scope-risk: narrow
Reversibility: clean
Directive: If agent snapshot fallback behavior changes again, verify both error preservation and snapshot resumability
Tested: hatch -e test run pytest test/components/agents/test_agent_breakpoints.py -k 'resume_from_tool_invoker' -q
Tested: hatch -e test run pytest test/core/pipeline/test_breakpoint.py -k 'create_agent_snapshot' -q
Tested: hatch run fmt-check haystack/core/pipeline/breakpoint.py test/components/agents/test_agent_breakpoints.py
Not-tested: Full unit suite and integration suite
Related: deepset-ai#11126

Author: shaun0927

haystack/core/pipeline/breakpoint.py

@@ -375,29 +375,12 @@ def _create_agent_snapshot(
     :param agent_breakpoint: AgentBreakpoint object containing breakpoints
     :return: An AgentSnapshot containing the agent's state and component visits.
     """
-    try:
-        serialized_chat_generator = _serialize_value_with_schema(
-            _deepcopy_with_exceptions(component_inputs["chat_generator"])
-        )
-    except Exception as error:
-        logger.warning(
-            "Failed to serialize the agent's chat_generator inputs. "
-            "The inputs in the snapshot will be replaced with an empty dictionary. Error: {e}",
-            e=error,
-        )
-        serialized_chat_generator = {}
-
-    try:
-        serialized_tool_invoker = _serialize_value_with_schema(
-            _deepcopy_with_exceptions(component_inputs["tool_invoker"])
-        )
-    except Exception as error:
-        logger.warning(
-            "Failed to serialize the agent's tool_invoker inputs. "
-            "The inputs in the snapshot will be replaced with an empty dictionary. Error: {e}",
-            e=error,
-        )
-        serialized_tool_invoker = {}
+    serialized_chat_generator = _serialize_agent_component_inputs(
+        component_name="chat_generator", component_inputs=component_inputs["chat_generator"]
+    )
+    serialized_tool_invoker = _serialize_agent_component_inputs(
+        component_name="tool_invoker", component_inputs=component_inputs["tool_invoker"]
+    )
 
     return AgentSnapshot(
         component_inputs={"chat_generator": serialized_chat_generator, "tool_invoker": serialized_tool_invoker},
@@ -407,6 +390,56 @@ def _create_agent_snapshot(
     )
 
 
+def _serialize_agent_component_inputs(component_name: str, component_inputs: dict[str, Any]) -> dict[str, Any]:
+    """
+    Serialize agent component inputs while preserving resumable fields whenever possible.
+
+    If serializing the whole input mapping fails (for example due to a non-serializable callback),
+    we retry field-by-field and omit only the failing fields. This keeps snapshots resumable when
+    required fields like ``messages`` or ``state`` are still serializable.
+
+    :param component_name: Name of the agent sub-component (e.g. ``chat_generator`` or ``tool_invoker``).
+    :param component_inputs: Runtime inputs for that sub-component.
+    :returns: A serialized payload, or ``{}`` if no fields can be serialized at all.
+    """
+    try:
+        return _serialize_value_with_schema(_deepcopy_with_exceptions(component_inputs))
+    except Exception as error:
+        logger.warning(
+            "Failed to serialize the agent's {component_name} inputs. "
+            "Haystack will omit only the non-serializable fields when possible. Error: {e}",
+            component_name=component_name,
+            e=error,
+        )
+
+    serialized_properties: dict[str, Any] = {}
+    serialized_data: dict[str, Any] = {}
+
+    for field_name, value in component_inputs.items():
+        try:
+            serialized_value = _serialize_value_with_schema(_deepcopy_with_exceptions(value))
+        except Exception as field_error:
+            logger.warning(
+                "Failed to serialize the agent's {component_name}.{field_name} input. "
+                "The field will be omitted from the snapshot. Error: {e}",
+                component_name=component_name,
+                field_name=field_name,
+                e=field_error,
+            )
+            continue
+
+        serialized_properties[field_name] = serialized_value["serialization_schema"]
+        serialized_data[field_name] = serialized_value["serialized_data"]
+
+    if not serialized_properties:
+        return {}
+
+    return {
+        "serialization_schema": {"type": "object", "properties": serialized_properties},
+        "serialized_data": serialized_data,
+    }
+
+
 def _validate_tool_breakpoint_is_valid(agent_breakpoint: AgentBreakpoint, tools: "ToolsType") -> None:
     """
     Validates the AgentBreakpoint passed to the agent.

releasenotes/notes/fix-agent-snapshot-resume-after-fallback-7fd7ff9a0f8f8b87.yaml

@@ -0,0 +1,6 @@
+---
+fixes:
+  - |
+    Preserve resumable agent snapshots when some ``chat_generator`` or ``tool_invoker`` inputs are
+    non-serializable. Haystack now omits only the failing runtime-only fields (for example
+    non-serializable callbacks) instead of replacing the whole payload with an empty dictionary.

test/components/agents/test_agent_breakpoints.py

@@ -472,6 +472,40 @@ def test_resume_from_tool_invoker(self, agent, tmp_path, monkeypatch):
         assert "last_message" in result
         assert len(result["messages"]) > 0
 
+    def test_resume_from_tool_invoker_omits_non_serializable_runtime_callback(self, agent, tmp_path, monkeypatch):
+        monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, "true")
+        debug_path = str(tmp_path / "debug_snapshots")
+        tool_bp = ToolBreakpoint(component_name="tool_invoker", tool_name="weather_tool", snapshot_file_path=debug_path)
+        agent_breakpoint = AgentBreakpoint(break_point=tool_bp, agent_name="test_agent")
+
+        try:
+            agent.run(
+                messages=[ChatMessage.from_user("What's the weather in Berlin?")],
+                break_point=agent_breakpoint,
+                streaming_callback=lambda chunk: None,
+            )
+        except BreakpointException:
+            pass
+
+        snapshot_files = list(Path(debug_path).glob("test_agent_tool_invoker_*.json"))
+        assert len(snapshot_files) > 0
+        latest_snapshot_file = str(max(snapshot_files, key=os.path.getctime))
+        agent_snapshot = load_pipeline_snapshot(latest_snapshot_file).agent_snapshot
+
+        assert agent_snapshot is not None
+        assert "streaming_callback" not in agent_snapshot.component_inputs["chat_generator"]["serialized_data"]
+        assert "streaming_callback" not in agent_snapshot.component_inputs["tool_invoker"]["serialized_data"]
+        assert "state" in agent_snapshot.component_inputs["tool_invoker"]["serialized_data"]
+
+        result = agent.run(
+            messages=[ChatMessage.from_user("This is actually ignored when resuming from snapshot.")],
+            snapshot=agent_snapshot,
+        )
+
+        assert "messages" in result
+        assert "last_message" in result
+        assert len(result["messages"]) == 4
+
     def test_resume_from_tool_invoker_and_new_breakpoint(self, weather_tool, tmp_path, monkeypatch):
         monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, "true")
         agent = Agent(