fix: produce() snapshots mutable values when recording patches

patchdiff/produce.py

@@ -7,7 +7,7 @@
 
 from __future__ import annotations
 
-import copy
+from copy import deepcopy
 from typing import Any, Callable, Dict, List, Set, Tuple, Union
 
 from .pointer import Pointer
@@ -57,7 +57,7 @@ def record_add(
                          If not provided, uses the same path. This is needed
                          for sets where add uses "/-" but remove needs "/value".
         """
-        self.patches.append({"op": "add", "path": path, "value": value})
+        self.patches.append({"op": "add", "path": path, "value": deepcopy(value)})
         self.reverse_patches.insert(
             0, {"op": "remove", "path": reverse_path if reverse_path else path}
         )
@@ -81,17 +81,19 @@ def record_remove(
             {
                 "op": "add",
                 "path": reverse_path if reverse_path else path,
-                "value": old_value,
+                "value": deepcopy(old_value),
             },
         )
 
     def record_replace(self, path: Pointer, old_value: Any, new_value: Any) -> None:
         """Record a replace operation, but only if the value actually changed."""
         if old_value == new_value:
             return  # Skip no-op replacements
-        self.patches.append({"op": "replace", "path": path, "value": new_value})
+        self.patches.append(
+            {"op": "replace", "path": path, "value": deepcopy(new_value)}
+        )
         self.reverse_patches.insert(
-            0, {"op": "replace", "path": path, "value": old_value}
+            0, {"op": "replace", "path": path, "value": deepcopy(old_value)}
         )
 
 
@@ -752,7 +754,7 @@ def produce(
             base = observ_to_raw(base)
 
         # Create a deep copy of the base object
-        draft = copy.deepcopy(base)
+        draft = deepcopy(base)
 
     # Create a patch recorder
     recorder = PatchRecorder()

tests/test_produce_core.py

@@ -869,3 +869,51 @@ def test_set_proxy_api_completeness(self):
             f"SetProxy is missing methods from set: {sorted(unhandled)}. "
             f"Either implement them on SetProxy or add to _SET_SKIPPED."
         )
+
+
+class TestPatchSnapshotting:
+    """Patches must snapshot mutable values so later proxy mutations don't
+    silently rewrite earlier patches' stored values."""
+
+    def test_dict_value_mutated_after_assignment_is_snapshotted(self):
+        """Recipe assigns a list, then mutates it. Replay must match result."""
+
+        def recipe(draft):
+            draft["a"] = [1, 2, 3]
+            draft["a"].append(4)
+
+        result, patches, reverse = produce({}, recipe)
+        assert result == {"a": [1, 2, 3, 4]}
+        assert apply({}, patches) == result
+        assert apply(result, reverse) == {}
+
+    def test_list_element_mutated_after_append_is_snapshotted(self):
+        """Recipe appends a dict, then mutates it. Replay must match result."""
+
+        def recipe(draft):
+            draft.append({"x": 1})
+            draft[-1]["x"] = 2
+
+        result, patches, reverse = produce([], recipe)
+        assert result == [{"x": 2}]
+        assert apply([], patches) == result
+        assert apply(result, reverse) == []
+
+    def test_nested_dict_in_list_in_dict_mutation_is_snapshotted(self):
+        """Arbitrarily nested mutation after the parent was added."""
+
+        def recipe(draft):
+            draft["users"] = [{"name": "Alice", "tags": ["py"]}]
+            draft["users"][0]["tags"].append("js")
+            draft["users"].append({"name": "Bob", "tags": ["go"]})
+            draft["users"][1]["tags"][0] = "rust"
+
+        result, patches, reverse = produce({}, recipe)
+        assert result == {
+            "users": [
+                {"name": "Alice", "tags": ["py", "js"]},
+                {"name": "Bob", "tags": ["rust"]},
+            ]
+        }
+        assert apply({}, patches) == result
+        assert apply(result, reverse) == {}