deepset-ai--haystack/test/core/pipeline/test_breakpoint.py

# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

import json
import logging

import pytest

from haystack import component
from haystack.core.errors import BreakpointException, PipelineInvalidPipelineSnapshotError
from haystack.core.pipeline import Pipeline
from haystack.core.pipeline.breakpoint import (
    HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED,
    _create_pipeline_snapshot,
    _is_snapshot_save_enabled,
    _save_pipeline_snapshot,
    _transform_json_structure,
    load_pipeline_snapshot,
)
from haystack.dataclasses import ChatMessage
from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot, PipelineState
from haystack.utils import _deserialize_value_with_schema

_EMPTY_OBJECT_PAYLOAD = {"serialization_schema": {"type": "object", "properties": {}}, "serialized_data": {}}


def test_transform_json_structure_unwraps_sender_value():
    data = {
        "key1": [{"sender": None, "value": "some value"}],
        "key2": [{"sender": "comp1", "value": 42}],
        "key3": "direct value",
    }

    result = _transform_json_structure(data)

    assert result == {"key1": "some value", "key2": 42, "key3": "direct value"}


def test_transform_json_structure_handles_nested_structures():
    data = {
        "key1": [{"sender": None, "value": "value1"}],
        "key2": {"nested": [{"sender": "comp1", "value": "value2"}], "direct": "value3"},
        "key3": [[{"sender": None, "value": "value4"}], [{"sender": "comp2", "value": "value5"}]],
    }

    result = _transform_json_structure(data)

    assert result == {"key1": "value1", "key2": {"nested": "value2", "direct": "value3"}, "key3": ["value4", "value5"]}


def test_load_pipeline_snapshot_loads_valid_snapshot(tmp_path):
    pipeline_snapshot = {
        "break_point": {"component_name": "comp1", "visit_count": 0},
        "pipeline_state": {"inputs": {}, "component_visits": {"comp1": 0, "comp2": 0}, "pipeline_outputs": {}},
        "original_input_data": {},
        "ordered_component_names": ["comp1", "comp2"],
        "include_outputs_from": ["comp1", "comp2"],
    }
    pipeline_snapshot_file = tmp_path / "state.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_load_state_handles_invalid_state(tmp_path):
    pipeline_snapshot = {
        "break_point": {"component_name": "comp1", "visit_count": 0},
        "pipeline_state": {"inputs": {}, "component_visits": {"comp1": 0, "comp2": 0}, "pipeline_outputs": {}},
        "original_input_data": {},
        "include_outputs_from": ["comp1", "comp2"],
        "ordered_component_names": ["comp1", "comp3"],  # inconsistent with component_visits
    }

    pipeline_snapshot_file = tmp_path / "invalid_pipeline_snapshot.json"
    with open(pipeline_snapshot_file, "w") as f:
        json.dump(pipeline_snapshot, f)

    with pytest.raises(ValueError, match="Invalid pipeline snapshot from"):
        load_pipeline_snapshot(pipeline_snapshot_file)


def test_breakpoint_saves_intermediate_outputs(tmp_path, monkeypatch):
    monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, "true")

    @component
    class SimpleComponent:
        @component.output_types(result=str)
        def run(self, input_value: str) -> dict[str, str]:
            return {"result": f"processed_{input_value}"}

    pipeline = Pipeline()
    comp1 = SimpleComponent()
    comp2 = SimpleComponent()
    pipeline.add_component("comp1", comp1)
    pipeline.add_component("comp2", comp2)
    pipeline.connect("comp1", "comp2")

    # 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
        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)}"

        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 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
class _AppendingComponent:
    @component.output_types(result=str)
    def run(self, input_value: str) -> dict[str, str]:
        return {"result": f"{input_value}_processed"}


def _three_component_pipeline() -> Pipeline:
    pipeline = Pipeline()
    pipeline.add_component("comp1", _AppendingComponent())
    pipeline.add_component("comp2", _AppendingComponent())
    pipeline.add_component("comp3", _AppendingComponent())
    pipeline.connect("comp1", "comp2")
    pipeline.connect("comp2", "comp3")
    return pipeline


def test_break_point_with_pipeline_snapshot_steps_through_pipeline():
    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}

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


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

    # 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():
    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(PipelineInvalidPipelineSnapshotError, match="different component or visit count"):
        pipeline.run(data={}, pipeline_snapshot=snapshot, break_point=Breakpoint(component_name="comp2", visit_count=0))


class TestCreatePipelineSnapshot:
    def test_create_pipeline_snapshot_all_fields(self):
        break_point = Breakpoint(component_name="comp2")
        ordered_component_names = ["comp1", "comp2"]
        include_outputs_from = {"comp1"}

        snapshot = _create_pipeline_snapshot(
            inputs={"comp1": {"input_value": [{"sender": None, "value": "test"}]}, "comp2": {}},
            component_inputs={"input_value": "processed_test"},
            break_point=break_point,
            component_visits={"comp1": 1, "comp2": 0},
            original_input_data={"comp1": {"input_value": "test"}},
            ordered_component_names=ordered_component_names,
            include_outputs_from=include_outputs_from,
            pipeline_outputs={"comp1": {"result": "processed_test"}},
        )

        assert snapshot.original_input_data == {
            "serialization_schema": {
                "type": "object",
                "properties": {"comp1": {"type": "object", "properties": {"input_value": {"type": "string"}}}},
            },
            "serialized_data": {"comp1": {"input_value": "test"}},
        }
        assert snapshot.ordered_component_names == ordered_component_names
        assert snapshot.break_point == break_point
        assert snapshot.include_outputs_from == include_outputs_from
        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"}}},
                    },
                },
                "serialized_data": {"comp1": {"input_value": "test"}, "comp2": {"input_value": "processed_test"}},
            },
            component_visits={"comp1": 1, "comp2": 0},
            pipeline_outputs={
                "serialization_schema": {
                    "type": "object",
                    "properties": {"comp1": {"type": "object", "properties": {"result": {"type": "string"}}}},
                },
                "serialized_data": {"comp1": {"result": "processed_test"}},
            },
        )

    def test_create_pipeline_snapshot_with_dataclasses_in_pipeline_outputs(self):
        snapshot = _create_pipeline_snapshot(
            inputs={},
            component_inputs={},
            break_point=Breakpoint(component_name="comp2"),
            component_visits={"comp1": 1, "comp2": 0},
            original_input_data={},
            ordered_component_names=["comp1", "comp2"],
            include_outputs_from={"comp1"},
            pipeline_outputs={"comp1": {"result": ChatMessage.from_user("hello")}},
        )

        assert snapshot.pipeline_state == PipelineState(
            inputs={
                "serialization_schema": {
                    "type": "object",
                    "properties": {"comp2": {"type": "object", "properties": {}}},
                },
                "serialized_data": {"comp2": {}},
            },
            component_visits={"comp1": 1, "comp2": 0},
            pipeline_outputs={
                "serialization_schema": {
                    "type": "object",
                    "properties": {
                        "comp1": {
                            "type": "object",
                            "properties": {"result": {"type": "haystack.dataclasses.chat_message.ChatMessage"}},
                        }
                    },
                },
                "serialized_data": {
                    "comp1": {"result": {"role": "user", "meta": {}, "name": None, "content": [{"text": "hello"}]}}
                },
            },
        )

    def test_create_pipeline_snapshot_non_serializable_inputs(self, caplog):
        class NonSerializable:
            def to_dict(self):
                raise TypeError("Cannot serialize")

        with caplog.at_level(logging.WARNING):
            _create_pipeline_snapshot(
                inputs={"comp1": {"input_value": [{"sender": None, "value": NonSerializable()}]}, "comp2": {}},
                component_inputs={},
                break_point=Breakpoint(component_name="comp2"),
                component_visits={"comp1": 1, "comp2": 0},
                original_input_data={"comp1": {"input_value": NonSerializable()}},
                ordered_component_names=["comp1", "comp2"],
                include_outputs_from={"comp1"},
                pipeline_outputs={},
            )

        assert any("Failed to serialize the inputs of the current pipeline state" in msg for msg in caplog.messages)
        assert any("Failed to serialize original input data for `pipeline.run`." in msg for msg in caplog.messages)

    def test_create_pipeline_snapshot_non_serializable_inputs_snapshot_is_resumable(self, caplog):
        """
        Guards against the same non-resumable snapshot regression fixed at the agent level: when
        top-level pipeline inputs/outputs contain non-serializable values, the snapshot fields
        must still round-trip through ``_deserialize_value_with_schema`` instead of failing with
        ``DeserializationError: ... Got: {}``. Serializable sibling components must stay intact.
        """

        class NonSerializable:
            def to_dict(self):
                raise TypeError("Cannot serialize")

        with caplog.at_level(logging.WARNING):
            snapshot = _create_pipeline_snapshot(
                inputs={
                    "comp1": {"input_value": [{"sender": None, "value": NonSerializable()}]},
                    "comp2": {"input_value": [{"sender": None, "value": "keep me"}]},
                },
                component_inputs={},
                break_point=Breakpoint(component_name="comp3"),
                component_visits={"comp1": 1, "comp2": 1, "comp3": 0},
                original_input_data={"comp1": {"input_value": NonSerializable()}},
                ordered_component_names=["comp1", "comp2", "comp3"],
                include_outputs_from=set(),
                pipeline_outputs={"comp1": {"result": NonSerializable()}},
            )

        # No DeserializationError on any of the three pipeline-level payloads.
        deserialized_inputs = _deserialize_value_with_schema(snapshot.pipeline_state.inputs)
        deserialized_original_input_data = _deserialize_value_with_schema(snapshot.original_input_data)
        deserialized_outputs = _deserialize_value_with_schema(snapshot.pipeline_state.pipeline_outputs)

        # 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["comp3"] == {}
        # original_input_data and pipeline_outputs degrade to empty-but-valid payloads.
        assert deserialized_original_input_data == {}
        assert deserialized_outputs == {}
        assert any("Failed to serialize the inputs of the current pipeline state" in msg for msg in caplog.messages)
        assert any("Failed to serialize outputs of the current pipeline state" in msg for msg in caplog.messages)


def test_save_pipeline_snapshot_raises_on_failure(tmp_path, caplog, monkeypatch):
    monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, "true")

    snapshot = _create_pipeline_snapshot(
        inputs={},
        component_inputs={},
        break_point=Breakpoint(component_name="comp2", snapshot_file_path=str(tmp_path)),
        component_visits={"comp1": 1, "comp2": 0},
        original_input_data={},
        ordered_component_names=["comp1", "comp2"],
        include_outputs_from={"comp1"},
        # We use a non-serializable type (bytes) directly in pipeline outputs to trigger the error
        pipeline_outputs={"comp1": {"result": b"test"}},
    )

    with pytest.raises(TypeError):
        _save_pipeline_snapshot(snapshot)

    with caplog.at_level(logging.ERROR):
        _save_pipeline_snapshot(snapshot, raise_on_failure=False)
        assert any("Failed to save pipeline snapshot to" in msg for msg in caplog.messages)


class TestSnapshotCallback:
    def test_save_pipeline_snapshot_with_callback_no_file_created(self, tmp_path):
        captured_snapshots = []

        def custom_callback(snapshot: PipelineSnapshot) -> str:
            captured_snapshots.append(snapshot)
            return "custom_path_or_id"

        snapshot = _create_pipeline_snapshot(
            inputs={},
            component_inputs={},
            break_point=Breakpoint(component_name="comp2", snapshot_file_path=str(tmp_path)),
            component_visits={"comp1": 1, "comp2": 0},
            original_input_data={},
            ordered_component_names=["comp1", "comp2"],
            include_outputs_from=set(),
            pipeline_outputs={},
        )

        result = _save_pipeline_snapshot(snapshot, snapshot_callback=custom_callback)

        # Verify callback was invoked and returned expected value
        assert result == "custom_path_or_id"
        assert len(captured_snapshots) == 1
        assert captured_snapshots[0] == snapshot

        # Verify NO file was created on disk (callback bypasses file saving)
        assert list(tmp_path.glob("*.json")) == []

    def test_save_pipeline_snapshot_callback_returns_none_no_file_created(self, tmp_path):
        captured_snapshots = []

        def custom_callback(snapshot: PipelineSnapshot) -> None:
            captured_snapshots.append(snapshot)

        snapshot = _create_pipeline_snapshot(
            inputs={},
            component_inputs={},
            break_point=Breakpoint(component_name="comp2", snapshot_file_path=str(tmp_path)),
            component_visits={"comp1": 1, "comp2": 0},
            original_input_data={},
            ordered_component_names=["comp1", "comp2"],
            include_outputs_from=set(),
            pipeline_outputs={},
        )

        result = _save_pipeline_snapshot(snapshot, snapshot_callback=custom_callback)

        assert result is None
        assert len(captured_snapshots) == 1

        # Verify NO file was created on disk even when snapshot_file_path is set
        assert list(tmp_path.glob("*.json")) == []

    def test_save_pipeline_snapshot_without_callback_creates_file(self, tmp_path, monkeypatch):
        monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, "true")

        snapshot = _create_pipeline_snapshot(
            inputs={},
            component_inputs={},
            break_point=Breakpoint(component_name="comp2", snapshot_file_path=str(tmp_path)),
            component_visits={"comp1": 1, "comp2": 0},
            original_input_data={},
            ordered_component_names=["comp1", "comp2"],
            include_outputs_from=set(),
            pipeline_outputs={},
        )

        result = _save_pipeline_snapshot(snapshot)

        # Verify file WAS created on disk
        snapshot_files = list(tmp_path.glob("comp2_*.json"))

        # A file should be created when no callback is provided
        assert len(snapshot_files) == 1
        assert result == str(snapshot_files[0])

        # Verify file contains valid snapshot data
        loaded = load_pipeline_snapshot(snapshot_files[0])
        assert isinstance(loaded.break_point, Breakpoint)
        assert loaded.break_point.component_name == "comp2"

    def test_save_pipeline_snapshot_callback_raises_exception_no_file_created(self, tmp_path, caplog):
        def failing_callback(snapshot: PipelineSnapshot) -> str:
            raise RuntimeError("Database connection failed")

        snapshot = _create_pipeline_snapshot(
            inputs={},
            component_inputs={},
            break_point=Breakpoint(component_name="comp2", snapshot_file_path=str(tmp_path)),
            component_visits={"comp1": 1, "comp2": 0},
            original_input_data={},
            ordered_component_names=["comp1", "comp2"],
            include_outputs_from=set(),
            pipeline_outputs={},
        )

        # Test with raise_on_failure=True (default)
        with pytest.raises(RuntimeError, match="Database connection failed"):
            _save_pipeline_snapshot(snapshot, snapshot_callback=failing_callback)

        # Verify NO file was created even after exception
        assert list(tmp_path.glob("*.json")) == []

        # Test with raise_on_failure=False
        with caplog.at_level(logging.ERROR):
            result = _save_pipeline_snapshot(snapshot, raise_on_failure=False, snapshot_callback=failing_callback)
            assert result is None
            assert any("Failed to handle pipeline snapshot with custom callback" in msg for msg in caplog.messages)

        # Still no file should exist
        assert list(tmp_path.glob("*.json")) == []

    def test_pipeline_run_with_snapshot_callback(self, tmp_path):
        captured_snapshots = []

        def custom_callback(snapshot: PipelineSnapshot) -> str:
            captured_snapshots.append(snapshot)
            return "custom_snapshot_id"

        @component
        class SimpleComponent:
            @component.output_types(result=str)
            def run(self, input_value: str) -> dict[str, str]:
                return {"result": f"processed_{input_value}"}

        pipeline = Pipeline()
        comp1 = SimpleComponent()
        comp2 = SimpleComponent()
        pipeline.add_component("comp1", comp1)
        pipeline.add_component("comp2", comp2)
        pipeline.connect("comp1", "comp2")

        # breakpoint on comp2
        break_point = Breakpoint(component_name="comp2", visit_count=0, snapshot_file_path=str(tmp_path))

        with pytest.raises(BreakpointException) as exc_info:
            pipeline.run(
                data={"comp1": {"input_value": "test"}}, break_point=break_point, snapshot_callback=custom_callback
            )

        # Verify callback was called
        assert len(captured_snapshots) == 1
        assert isinstance(captured_snapshots[0].break_point, Breakpoint)
        assert captured_snapshots[0].break_point.component_name == "comp2"
        # Verify the file path in exception is from callback
        assert exc_info.value.pipeline_snapshot_file_path == "custom_snapshot_id"
        # Verify no file was saved to disk
        assert list(tmp_path.glob("*.json")) == []

    def test_pipeline_run_without_snapshot_callback_saves_file(self, tmp_path, monkeypatch):
        monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, "true")

        @component
        class SimpleComponent:
            @component.output_types(result=str)
            def run(self, input_value: str) -> dict[str, str]:
                return {"result": f"processed_{input_value}"}

        pipeline = Pipeline()
        comp1 = SimpleComponent()
        comp2 = SimpleComponent()
        pipeline.add_component("comp1", comp1)
        pipeline.add_component("comp2", comp2)
        pipeline.connect("comp1", "comp2")

        # breakpoint on comp2
        break_point = Breakpoint(component_name="comp2", visit_count=0, snapshot_file_path=str(tmp_path))

        with pytest.raises(BreakpointException):
            pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=break_point)

        # Verify file was saved to disk
        snapshot_files = list(tmp_path.glob("comp2_*.json"))
        assert len(snapshot_files) == 1


class TestSnapshotSaveEnabled:
    def test_is_snapshot_save_enabled_default(self, monkeypatch):
        monkeypatch.delenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, raising=False)
        assert _is_snapshot_save_enabled() is False

    @pytest.mark.parametrize("value", ["true", "TRUE", "True", "1"])
    def test_is_snapshot_save_enabled_truthy_values(self, monkeypatch, value):
        monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, value)
        assert _is_snapshot_save_enabled() is True

    @pytest.mark.parametrize("value", ["false", "FALSE", "False", "0"])
    def test_is_snapshot_save_enabled_falsy_values(self, monkeypatch, value):
        monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, value)
        assert _is_snapshot_save_enabled() is False

    def test_save_pipeline_snapshot_disabled_via_env_var(self, tmp_path, monkeypatch):
        monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, "false")

        snapshot = _create_pipeline_snapshot(
            inputs={},
            component_inputs={},
            break_point=Breakpoint(component_name="comp2", snapshot_file_path=str(tmp_path)),
            component_visits={"comp1": 1, "comp2": 0},
            original_input_data={},
            ordered_component_names=["comp1", "comp2"],
            include_outputs_from=set(),
            pipeline_outputs={},
        )

        result = _save_pipeline_snapshot(snapshot)

        # Verify no file was created
        assert result is None
        assert list(tmp_path.glob("*.json")) == []

    def test_save_pipeline_snapshot_enabled_via_env_var(self, tmp_path, monkeypatch):
        monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, "true")

        snapshot = _create_pipeline_snapshot(
            inputs={},
            component_inputs={},
            break_point=Breakpoint(component_name="comp2", snapshot_file_path=str(tmp_path)),
            component_visits={"comp1": 1, "comp2": 0},
            original_input_data={},
            ordered_component_names=["comp1", "comp2"],
            include_outputs_from=set(),
            pipeline_outputs={},
        )

        result = _save_pipeline_snapshot(snapshot)

        # Verify file was created
        snapshot_files = list(tmp_path.glob("comp2_*.json"))
        assert len(snapshot_files) == 1
        assert result == str(snapshot_files[0])

    def test_callback_still_invoked_when_env_var_disables_saving(self, tmp_path, monkeypatch):
        """
        This is more a behaviour documentation test: we want to ensure that when the snapshot_callback is provided,
        the file-saving behaviour is always bypassed (the callback is invoked instead).
        """
        monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, "false")

        captured_snapshots = []

        def custom_callback(snapshot: PipelineSnapshot) -> str:
            captured_snapshots.append(snapshot)
            return "custom_result"

        snapshot = _create_pipeline_snapshot(
            inputs={},
            component_inputs={},
            break_point=Breakpoint(component_name="comp2", snapshot_file_path=str(tmp_path)),
            component_visits={"comp1": 1, "comp2": 0},
            original_input_data={},
            ordered_component_names=["comp1", "comp2"],
            include_outputs_from=set(),
            pipeline_outputs={},
        )

        result = _save_pipeline_snapshot(snapshot, snapshot_callback=custom_callback)

        # Callback should still be invoked
        assert result == "custom_result"
        assert len(captured_snapshots) == 1
        # No file should be created (callback handles it)
        assert list(tmp_path.glob("*.json")) == []

    def test_pipeline_run_with_env_var_disabled(self, tmp_path, monkeypatch):
        """Test that pipeline.run respects the env var when breakpoint is triggered."""
        monkeypatch.setenv(HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, "false")

        @component
        class SimpleComponent:
            @component.output_types(result=str)
            def run(self, input_value: str) -> dict[str, str]:
                return {"result": f"processed_{input_value}"}

        pipeline = Pipeline()
        pipeline.add_component("comp1", SimpleComponent())
        pipeline.add_component("comp2", SimpleComponent())
        pipeline.connect("comp1", "comp2")

        break_point = Breakpoint(component_name="comp2", visit_count=0, snapshot_file_path=str(tmp_path))

        with pytest.raises(BreakpointException) as exc_info:
            pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=break_point)

        # Verify no file was saved
        assert exc_info.value.pipeline_snapshot_file_path is None
        assert list(tmp_path.glob("*.json")) == []

        # Verify snapshot object is still available for programmatic access
        assert exc_info.value.pipeline_snapshot is not None
        assert isinstance(exc_info.value.pipeline_snapshot.break_point, Breakpoint)
        assert exc_info.value.pipeline_snapshot.break_point.component_name == "comp2"