Python: Fix state snapshot to use deepcopy so nested mutations are detected in durable workflow activities

REPRODUCTION_REPORT.md

@@ -0,0 +1,35 @@
+# Reproduction Report — Issue #4500
+**Python: [Bug]: State updates not propagated to the next activity with durable function and declarative workflow**
+
+*Generated: 2026-03-06 04:28:49 UTC*
+
+## Result
+
+| Field | Value |
+|-------|-------|
+| Status | `reproduced` |
+| Failure observed | `yes` |
+| Evidence strength | `high` |
+| Confidence | `high` |
+| Failing test | `/Users/evmattso/git/agent-framework-2/.worktrees/agent/fix-4500-1/python/packages/azurefunctions/tests/test_shallow_copy_bug.py` |
+| Gate passed | Yes |
+
+## Files Examined
+
+python/packages/azurefunctions/agent_framework_azurefunctions/_app.py, python/packages/core/agent_framework/_workflows/_state.py, python/packages/azurefunctions/agent_framework_azurefunctions/_serialization.py, python/packages/azurefunctions/tests/test_app.py
+
+## Tests Run
+
+python/packages/azurefunctions/tests/test_shallow_copy_bug.py
+
+## Finding
+
+The bug is confirmed: `dict()` at line 303 of `_app.py` creates a shallow copy that shares nested object references with the state. When an executor mutates nested values in place, both `original_snapshot` and `current_state` reflect the same mutation, producing an empty diff and causing state updates to not propagate to downstream activities. The test suite demonstrates this with nested dicts, lists, and contrasts with `copy.deepcopy()` which correctly preserves the original snapshot.
+
+## Full Analysis
+
+Now I can see the bug clearly. Line 303 uses `dict()` (shallow copy). Let me find the `State` class and existing tests.Now I understand the bug. Let me look at `deserialize_value` and the `export_state` method to understand the full flow, then write a test.Now I have a clear understanding of the bug. The issue is at line 303: `original_snapshot = dict(deserialized_state)` creates a shallow copy. When the activity mutates a nested value (like a dict or list) in place, the mutation is reflected in both `deserialized_state` and `original_snapshot`, making the diff at line 334 see no changes.
+
+Let me write a focused test to reproduce this.Now I clearly see the issue. The `original_snapshot` at line 303 shares references to nested objects with `deserialized_state`, which is then imported into `shared_state` at line 304. When the executor mutates nested state values, the mutations are visible in `original_snapshot` too, making the diff at line 334 produce no updates.
+
+Let me write a minimal test to demonstrate this.Now let me write a minimal, focused test that demonstrates the shallow copy bug.All 5 tests pass, confirming the bug. The key tests (`test_shallow_copy_loses_nested_mutations`, `test_shallow_copy_loses_new_key_in_nested_dict`, `test_shallow_copy_loses_list_mutations`) pass by asserting `updates == {}` — proving that the shallow copy causes nested mutations to be invisible in the diff. The `test_deep_copy_preserves_nested_mutations` test shows that `copy.deepcopy()` correctly detects changes.

python/packages/azurefunctions/agent_framework_azurefunctions/_app.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import json
 import logging
 import re
@@ -300,7 +301,7 @@ async def run() -> dict[str, Any]:
 
                 # Deserialize shared state values to reconstruct dataclasses/Pydantic models
                 deserialized_state = {k: deserialize_value(v) for k, v in (shared_state_snapshot or {}).items()}
-                original_snapshot = dict(deserialized_state)
+                original_snapshot = copy.deepcopy(deserialized_state)
                 shared_state.import_state(deserialized_state)
 
                 if is_hitl_response:

python/packages/azurefunctions/tests/test_app.py

@@ -4,6 +4,7 @@
 
 # pyright: reportPrivateUsage=false
 
+import copy
 import json
 from collections.abc import Awaitable, Callable
 from typing import Any, TypeVar
@@ -1441,5 +1442,140 @@ def test_build_status_url_handles_trailing_slash(self) -> None:
         assert "instance-456" in url
 
 
+class TestStateSnapshotDiff:
+    """Test suite for state snapshot diffing in activity execution.
+
+    The activity executor snapshots state before execution and diffs against the
+    post-execution state to determine which keys were updated. The snapshot must
+    be independent of the live state so that in-place mutations to nested objects
+    (dicts, lists) are correctly detected as changes.
+    """
+
+    @staticmethod
+    def _compute_updates(original_snapshot: dict[str, Any], current_state: dict[str, Any]) -> dict[str, Any]:
+        """Compute state updates the same way as _app.py run()."""
+        return {k: v for k, v in current_state.items() if k not in original_snapshot or original_snapshot[k] != v}
+
+    def test_nested_dict_mutation_detected_in_diff(self) -> None:
+        """Test that mutating values inside a nested dict appears in the diff."""
+        from agent_framework._workflows._state import State
+
+        deserialized_state: dict[str, Any] = {
+            "Local.config": {"code": "", "enabled": False},
+            "simple_key": "simple_value",
+        }
+
+        original_snapshot = copy.deepcopy(deserialized_state)
+
+        shared_state = State()
+        shared_state.import_state(deserialized_state)
+
+        config = shared_state.get("Local.config")
+        config["code"] = "SOMECODEXXX"
+        config["enabled"] = True
+
+        shared_state.commit()
+        current_state = shared_state.export_state()
+
+        updates = self._compute_updates(original_snapshot, current_state)
+
+        assert "Local.config" in updates
+        assert updates["Local.config"]["code"] == "SOMECODEXXX"
+        assert updates["Local.config"]["enabled"] is True
+
+    def test_new_key_in_nested_dict_detected_in_diff(self) -> None:
+        """Test that adding a key to a nested dict appears in the diff."""
+        from agent_framework._workflows._state import State
+
+        deserialized_state: dict[str, Any] = {
+            "Local.data": {"existing": "value"},
+        }
+
+        original_snapshot = copy.deepcopy(deserialized_state)
+
+        shared_state = State()
+        shared_state.import_state(deserialized_state)
+
+        data = shared_state.get("Local.data")
+        data["code"] = "NEW_CODE"
+
+        shared_state.commit()
+        current_state = shared_state.export_state()
+
+        updates = self._compute_updates(original_snapshot, current_state)
+
+        assert "Local.data" in updates
+        assert updates["Local.data"]["code"] == "NEW_CODE"
+
+    def test_nested_list_mutation_detected_in_diff(self) -> None:
+        """Test that appending to a nested list appears in the diff."""
+        from agent_framework._workflows._state import State
+
+        deserialized_state: dict[str, Any] = {
+            "Local.items": [1, 2, 3],
+        }
+
+        original_snapshot = copy.deepcopy(deserialized_state)
+
+        shared_state = State()
+        shared_state.import_state(deserialized_state)
+
+        items = shared_state.get("Local.items")
+        items.append(4)
+
+        shared_state.commit()
+        current_state = shared_state.export_state()
+
+        updates = self._compute_updates(original_snapshot, current_state)
+
+        assert "Local.items" in updates
+        assert updates["Local.items"] == [1, 2, 3, 4]
+
+    def test_new_top_level_key_detected_in_diff(self) -> None:
+        """Test that setting a new top-level key appears in the diff."""
+        from agent_framework._workflows._state import State
+
+        deserialized_state: dict[str, Any] = {
+            "existing": "value",
+        }
+
+        original_snapshot = copy.deepcopy(deserialized_state)
+
+        shared_state = State()
+        shared_state.import_state(deserialized_state)
+
+        shared_state.set("Local.code", "SOMECODEXXX")
+
+        shared_state.commit()
+        current_state = shared_state.export_state()
+
+        updates = self._compute_updates(original_snapshot, current_state)
+
+        assert "Local.code" in updates
+        assert updates["Local.code"] == "SOMECODEXXX"
+
+    def test_unchanged_nested_state_produces_empty_diff(self) -> None:
+        """Test that unmodified nested state produces no updates."""
+        from agent_framework._workflows._state import State
+
+        deserialized_state: dict[str, Any] = {
+            "Local.config": {"code": "existing", "enabled": True},
+            "simple_key": "simple_value",
+        }
+
+        original_snapshot = copy.deepcopy(deserialized_state)
+
+        shared_state = State()
+        shared_state.import_state(deserialized_state)
+
+        # No mutations performed
+        shared_state.commit()
+        current_state = shared_state.export_state()
+
+        updates = self._compute_updates(original_snapshot, current_state)
+
+        assert updates == {}
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v", "--tb=short"])