fix: isolate raw tool payloads in session snapshots

docs/contrib.md

@@ -13,6 +13,7 @@ The helpers under `acp.contrib` package recurring patterns we saw in integration
 
 - `SessionAccumulator.apply(notification)` reconciles `tool_call` and `tool_call_update` payloads, even if the start event arrives late.
 - `snapshot()` returns an immutable `SessionSnapshot` Pydantic model containing the plan, current mode, commands, and ordered message history.
+- Raw tool input/output payloads are copied when applying notifications and creating snapshots, so nested payload mutations do not change the accumulated state.
 - `subscribe(callback)` lets you push snapshots into stores or UI components whenever state changes.
 - Automatic reset on session-change (toggle with `auto_reset_on_session_change`).
 

src/acp/contrib/session_state.py

@@ -2,6 +2,7 @@
 
 from collections.abc import Callable, Sequence
 from contextlib import suppress
+from copy import deepcopy
 from typing import Any
 
 from pydantic import BaseModel, ConfigDict
@@ -62,8 +63,8 @@ def apply_start(self, update: ToolCallStart) -> None:
         self.status = update.status
         self.content = _copy_model_list(update.content)
         self.locations = _copy_model_list(update.locations)
-        self.raw_input = update.raw_input
-        self.raw_output = update.raw_output
+        self.raw_input = deepcopy(update.raw_input)
+        self.raw_output = deepcopy(update.raw_output)
 
     def apply_progress(self, update: ToolCallProgress) -> None:
         if update.title is not None:
@@ -77,9 +78,9 @@ def apply_progress(self, update: ToolCallProgress) -> None:
         if update.locations is not None:
             self.locations = _copy_model_list(update.locations)
         if update.raw_input is not None:
-            self.raw_input = update.raw_input
+            self.raw_input = deepcopy(update.raw_input)
         if update.raw_output is not None:
-            self.raw_output = update.raw_output
+            self.raw_output = deepcopy(update.raw_output)
 
     def snapshot(self) -> ToolCallView:
         return ToolCallView(
@@ -89,8 +90,8 @@ def snapshot(self) -> ToolCallView:
             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),
         )
 
 

tests/contrib/test_contrib_session_state.py

@@ -22,6 +22,66 @@ def notification(session_id: str, update):
     return SessionNotification(session_id=session_id, update=update)
 
 
+@pytest.fixture(params=["start", "progress"])
+def raw_tool_notification(request: pytest.FixtureRequest) -> SessionNotification:
+    raw_input = {"options": {"paths": ["input.txt"]}}
+    raw_output = {"result": {"lines": ["original"]}}
+    update: ToolCallStart | ToolCallProgress
+    if request.param == "start":
+        update = ToolCallStart(
+            session_update="tool_call",
+            tool_call_id="call-1",
+            title="Read file",
+            raw_input=raw_input,
+            raw_output=raw_output,
+        )
+    else:
+        update = ToolCallProgress(
+            session_update="tool_call_update",
+            tool_call_id="call-1",
+            raw_input=raw_input,
+            raw_output=raw_output,
+        )
+    return notification("s", update)
+
+
+def test_raw_notification_mutation_does_not_change_accumulated_state(
+    raw_tool_notification: SessionNotification,
+) -> None:
+    acc = SessionAccumulator()
+    first = acc.apply(raw_tool_notification)
+    update = raw_tool_notification.update
+    assert isinstance(update, ToolCallStart | ToolCallProgress)
+    assert update.raw_input is not None
+    assert update.raw_output is not None
+
+    update.raw_input["options"]["paths"].append("external.txt")
+    update.raw_output["result"]["lines"].append("external")
+
+    for snapshot in (first, acc.snapshot()):
+        assert snapshot.tool_calls["call-1"].raw_input == {"options": {"paths": ["input.txt"]}}
+        assert snapshot.tool_calls["call-1"].raw_output == {"result": {"lines": ["original"]}}
+
+
+def test_raw_snapshot_mutation_does_not_change_other_snapshots_or_state(
+    raw_tool_notification: SessionNotification,
+) -> None:
+    acc = SessionAccumulator()
+    first = acc.apply(raw_tool_notification)
+    second = acc.snapshot()
+
+    first.tool_calls["call-1"].raw_input["options"]["paths"].append("external.txt")
+    first.tool_calls["call-1"].raw_output["result"]["lines"].append("external")
+
+    for snapshot in (second, acc.snapshot()):
+        assert snapshot.tool_calls["call-1"].raw_input == {"options": {"paths": ["input.txt"]}}
+        assert snapshot.tool_calls["call-1"].raw_output == {"result": {"lines": ["original"]}}
+    update = raw_tool_notification.update
+    assert isinstance(update, ToolCallStart | ToolCallProgress)
+    assert update.raw_input == {"options": {"paths": ["input.txt"]}}
+    assert update.raw_output == {"result": {"lines": ["original"]}}
+
+
 def test_session_accumulator_merges_tool_calls():
     acc = SessionAccumulator()
     start = ToolCallStart(