diff --git a/src/acp/contrib/tool_calls.py b/src/acp/contrib/tool_calls.py index 6fa5a7f5..3b5c05c9 100644 --- a/src/acp/contrib/tool_calls.py +++ b/src/acp/contrib/tool_calls.py @@ -2,6 +2,7 @@ import uuid from collections.abc import Callable, Sequence +from copy import deepcopy from typing import Any, cast from pydantic import BaseModel, ConfigDict @@ -86,8 +87,8 @@ def __init__( self.status = status self.content = _copy_model_list(content) self.locations = _copy_model_list(locations) - self.raw_input = raw_input - self.raw_output = raw_output + self.raw_input = deepcopy(raw_input) + self.raw_output = deepcopy(raw_output) self._stream_buffer: str | None = None def to_view(self) -> TrackedToolCallView: diff --git a/tests/contrib/test_contrib_tool_calls.py b/tests/contrib/test_contrib_tool_calls.py index 71ffefc2..80e6f05c 100644 --- a/tests/contrib/test_contrib_tool_calls.py +++ b/tests/contrib/test_contrib_tool_calls.py @@ -1,5 +1,7 @@ from __future__ import annotations +import pytest + from acp.contrib.tool_calls import ToolCallTracker from acp.schema import ContentToolCallContent, TextContentBlock, ToolCallProgress @@ -33,3 +35,41 @@ def test_tool_call_tracker_streaming_text_updates_content(): assert isinstance(second_content, ContentToolCallContent) assert isinstance(second_content.content, TextContentBlock) assert second_content.content.text == "hello, world" + + +@pytest.mark.parametrize("field", ["raw_input", "raw_output"]) +def test_tracker_start_snapshots_incoming_raw_payloads(field: str) -> None: + tracker = ToolCallTracker() + payload = {"nested": ["original"]} + notification = tracker.start( + "external", + title="Run", + raw_input=payload if field == "raw_input" else None, + raw_output=payload if field == "raw_output" else None, + ) + + payload["nested"].append("caller mutation") + + assert getattr(tracker.view("external"), field) == {"nested": ["original"]} + assert getattr(notification, field) == {"nested": ["original"]} + + +@pytest.mark.parametrize("payload", [None, False, 0, "text", ["item"]]) +def test_tracker_preserves_json_raw_payload_values(payload: object) -> None: + tracker = ToolCallTracker() + tracker.start("external", title="Run", raw_input=payload, raw_output=payload) + + assert tracker.view("external").raw_input == payload + assert tracker.tool_call_model("external").raw_output == payload + + +def test_tracker_progress_can_clear_raw_payloads() -> None: + tracker = ToolCallTracker() + tracker.start("external", title="Run", raw_input={"old": True}, raw_output={"old": True}) + + progress = tracker.progress("external", raw_input=None, raw_output=None) + + assert progress.raw_input is None + assert progress.raw_output is None + assert tracker.view("external").raw_input is None + assert tracker.view("external").raw_output is None