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

* Use deepcopy for state snapshot to detect nested mutations (#4500)

Replace dict() shallow copy with copy.deepcopy() when snapshotting
workflow state before activity execution. The shallow copy shared
references to nested objects (dicts, lists), so in-place mutations by
executors were reflected in both the snapshot and live state, producing
an empty diff and preventing state updates from propagating to
downstream activities.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

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

Fixes #4500

* Address PR review: remove report, extract testable helpers (#4500)

- Delete REPRODUCTION_REPORT.md (debugging artifact with local paths
  and raw LLM output)
- Extract _create_state_snapshot() and _compute_state_updates() as
  module-level helpers in _app.py so tests exercise the production
  code path
- Update TestStateSnapshotDiff to import and use production helpers
  instead of reimplementing snapshot/diff logic locally

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Apply pre-commit auto-fixes

* Add regression tests proving shallow copy bug and deep copy isolation (#4500)

Add two additional tests to TestStateSnapshotDiff:
- test_shallow_copy_would_miss_nested_mutations: reproduces the original
  bug by demonstrating that dict() (shallow copy) misses nested mutations
- test_create_state_snapshot_isolates_nested_objects: verifies the
  production _create_state_snapshot helper creates a true deep copy

These tests ensure a regression back to shallow copy would be caught.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Add integration test exercising full activity code path (#4500)

Address PR review comment: add test_executor_activity_detects_nested_state_mutations
that captures the actual executor_activity function from _setup_executor_activity
and verifies it detects in-place nested mutations. This test would fail if
_app.py line 314 regressed from _create_state_snapshot() back to dict().

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Address review feedback for #4518: review comment fixes

* Address PR review feedback for state snapshot diff

- Inline _compute_state_updates logic at call site to reuse precomputed
  original_keys/current_keys sets, avoiding redundant set allocations
- Fix test docstring to describe behavioral regression instead of
  hard-coding a specific line number
- Use SOURCE_ORCHESTRATOR constant in integration test instead of
  literal string

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Apply pre-commit auto-fixes

* fix: remove unused _compute_state_updates from _app.py (#4518)

The function was inlined per review comment, making the module-level
helper unused and triggering a pyright reportUnusedFunction error.
Move the helper into the test file where it is still needed for unit
testing the diffing logic.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

python/packages/azurefunctions/agent_framework_azurefunctions/_app.py

@@ -14,6 +14,7 @@ import logging
 import re
 import uuid
 from collections.abc import Callable, Mapping
+from copy import deepcopy
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, TypeVar, cast
@@ -58,6 +59,11 @@ EntityHandler = Callable[[df.DurableEntityContext], None]
 HandlerT = TypeVar("HandlerT", bound=Callable[..., Any])
 
 
+def _create_state_snapshot(state: dict[str, Any]) -> dict[str, Any]:
+    """Create a deep copy of the deserialized state for later diffing."""
+    return deepcopy(state)
+
+
 @dataclass
 class AgentMetadata:
     """Metadata for a registered agent.
@@ -306,7 +312,7 @@ class AgentFunctionApp(DFAppBase):
                 deserialized_state: dict[str, Any] = {
                     str(k): deserialize_value(v) for k, v in shared_state_snapshot.items()
                 }
-                original_snapshot: dict[str, Any] = dict(deserialized_state)
+                original_snapshot = _create_state_snapshot(deserialized_state)
                 shared_state.import_state(deserialized_state)
 
                 if is_hitl_response:
@@ -339,9 +345,10 @@ class AgentFunctionApp(DFAppBase):
                 deletes: set[str] = original_keys - current_keys
 
                 # Updates = keys in current that are new or have different values
-                updates = {
-                    k: v for k, v in current_state.items() if k not in original_snapshot or original_snapshot[k] != v
-                }
+                updates: dict[str, Any] = {}
+                for key in current_keys:
+                    if key not in original_keys or current_state[key] != original_snapshot.get(key):
+                        updates[key] = current_state[key]
 
                 # Drain messages and events from runner context
                 sent_messages = await runner_context.drain_messages()