fix(contrib): isolate tracked raw tool payloads

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:
@@ -99,8 +100,8 @@ def to_view(self) -> TrackedToolCallView:
             status=self.status,
             content=tuple(item.model_copy(deep=True) for item in self.content) if self.content else None,
             locations=tuple(loc.model_copy(deep=True) for loc in self.locations) if self.locations else None,
-            raw_input=self.raw_input,
-            raw_output=self.raw_output,
+            raw_input=deepcopy(self.raw_input),
+            raw_output=deepcopy(self.raw_output),
         )
 
     def to_tool_call_model(self) -> ToolCallUpdate:
@@ -112,8 +113,8 @@ def to_tool_call_model(self) -> ToolCallUpdate:
             status=self.status,
             content=_copy_model_list(self.content),
             locations=_copy_model_list(self.locations),
-            raw_input=self.raw_input,
-            raw_output=self.raw_output,
+            raw_input=deepcopy(self.raw_input),
+            raw_output=deepcopy(self.raw_output),
         )
 
     def to_start_model(self) -> ToolCallStart:
@@ -128,8 +129,8 @@ def to_start_model(self) -> ToolCallStart:
             status=self.status,
             content=_copy_model_list(self.content),
             locations=_copy_model_list(self.locations),
-            raw_input=self.raw_input,
-            raw_output=self.raw_output,
+            raw_input=deepcopy(self.raw_input),
+            raw_output=deepcopy(self.raw_output),
         )
 
     def update(
@@ -169,11 +170,11 @@ def update(
             )
             kwargs["locations"] = _copy_model_list(seq_locations)
         if raw_input is not UNSET:
-            self.raw_input = raw_input
-            kwargs["rawInput"] = raw_input
+            self.raw_input = deepcopy(raw_input)
+            kwargs["rawInput"] = deepcopy(self.raw_input)
         if raw_output is not UNSET:
-            self.raw_output = raw_output
-            kwargs["raw_output"] = raw_output
+            self.raw_output = deepcopy(raw_output)
+            kwargs["raw_output"] = deepcopy(self.raw_output)
         return ToolCallProgress(session_update="tool_call_update", tool_call_id=self.tool_call_id, **kwargs)
 
     def append_stream_text(
@@ -189,7 +190,11 @@ def append_stream_text(
 
 
 class ToolCallTracker:
-    """Utility for generating ACP tool call updates on the server side."""
+    """Utility for generating ACP tool call updates on the server side.
+
+    Raw input/output payloads are copied on entry and export so caller, view,
+    notification, and permission-request mutations cannot change tracked state.
+    """
 
     def __init__(self, *, id_factory: Callable[[], str] | None = None) -> None:
         self._id_factory = id_factory or (lambda: uuid.uuid4().hex)

tests/contrib/test_contrib_permissions.py

@@ -116,3 +116,22 @@ def test_default_permission_options_shape():
     options = default_permission_options()
     assert len(options) == 3
     assert {opt.option_id for opt in options} == {"approve", "approve_for_session", "reject"}
+
+
+@pytest.mark.asyncio
+async def test_permission_request_raw_payloads_are_detached_from_tracker() -> None:
+    tracker = ToolCallTracker()
+    tracker.start("external", title="Run", raw_input={"args": ["original"]}, raw_output={"result": ["original"]})
+
+    async def requester(request: RequestPermissionRequest) -> RequestPermissionResponse:
+        assert isinstance(request.tool_call.raw_input, dict)
+        assert isinstance(request.tool_call.raw_output, dict)
+        request.tool_call.raw_input["args"].append("request mutation")
+        request.tool_call.raw_output["result"].append("request mutation")
+        return RequestPermissionResponse(outcome=AllowedOutcome(option_id="approve", outcome="selected"))
+
+    broker = PermissionBroker("session", requester, tracker=tracker)
+    await broker.request_for("external")
+
+    assert tracker.view("external").raw_input == {"args": ["original"]}
+    assert tracker.view("external").raw_output == {"result": ["original"]}

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,79 @@ 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"])
+@pytest.mark.parametrize("operation", ["start", "progress"])
+def test_tracker_snapshots_incoming_raw_payloads(field: str, operation: str) -> None:
+    tracker = ToolCallTracker()
+    payload = {"nested": ["original"]}
+    if operation == "start":
+        notification = tracker.start(
+            "external",
+            title="Run",
+            raw_input=payload if field == "raw_input" else None,
+            raw_output=payload if field == "raw_output" else None,
+        )
+    else:
+        tracker.start("external", title="Run")
+        notification = tracker.progress(
+            "external",
+            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("field", ["raw_input", "raw_output"])
+@pytest.mark.parametrize("output", ["start", "progress", "view", "tool_call_model"])
+def test_tracker_raw_payload_exports_cannot_mutate_state(field: str, output: 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,
+    )
+    if output == "progress":
+        exported = tracker.progress(
+            "external",
+            raw_input=payload if field == "raw_input" else None,
+            raw_output=payload if field == "raw_output" else None,
+        )
+    elif output == "view":
+        exported = tracker.view("external")
+    elif output == "tool_call_model":
+        exported = tracker.tool_call_model("external")
+    else:
+        exported = notification
+
+    getattr(exported, field)["nested"].append("consumer mutation")
+
+    assert getattr(tracker.view("external"), 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