fix(delta-channel): unwrap NotRequired[X] for dict/set reducers

Annotated[NotRequired[dict[...]], DeltaChannel(reducer)] (the shape used
by deepagents' filesystem middleware) fell through type inference to
`list`, so the first operator call blew up with
"'list' object is not a mapping". `_is_field_channel` now unwraps a
parameterized Required[X]/NotRequired[X] before stripping extras, which
lets dict/set/mapping outer types reach the abc normalization block.

Also type-annotates the `new` locals in DeltaChannel.copy() and
from_checkpoint() so mypy can infer them through the abstract return
type.

Adds tests covering: dict Overwrite in update and in writes replay,
snapshot_write with a dict reducer, dict backwards-compat checkpoints,
NotRequired type inference, and a filesystem-shaped end-to-end graph.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

libs/langgraph/langgraph/channels/delta.py

@@ -86,19 +86,19 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
         return self.typ
 
     def copy(self) -> Self:
-        new = DeltaChannel(self.operator, snapshot_every=self.snapshot_every)
+        new: DeltaChannel[Value] = DeltaChannel(
+            self.operator, snapshot_every=self.snapshot_every
+        )
         new.typ = self.typ
         new.key = self.key
-        new.value = (
-            self.value
-            if self.value is MISSING
-            else _copy.copy(self.value)
-        )
+        new.value = self.value if self.value is MISSING else _copy.copy(self.value)
         new._writes_since_snapshot = self._writes_since_snapshot
         return new
 
     def from_checkpoint(self, checkpoint: Any) -> Self:
-        new = DeltaChannel(self.operator, snapshot_every=self.snapshot_every)
+        new: DeltaChannel[Value] = DeltaChannel(
+            self.operator, snapshot_every=self.snapshot_every
+        )
         new.typ = self.typ
         new.key = self.key
         if checkpoint is MISSING:

libs/langgraph/langgraph/graph/state.py

@@ -1671,7 +1671,15 @@ def _is_field_channel(typ: type[Any]) -> BaseChannel | None:
         for item in meta:
             if isinstance(item, BaseChannel):
                 if isinstance(item, DeltaChannel) and hasattr(typ, "__origin__"):
-                    outer = _strip_extras(typ.__origin__)
+                    origin = typ.__origin__
+                    # Unwrap parameterized Required[X]/NotRequired[X] to X
+                    # (e.g. Annotated[NotRequired[dict[...]], ...]).
+                    if hasattr(origin, "__origin__") and origin.__origin__ in (
+                        Required,
+                        NotRequired,
+                    ):
+                        origin = origin.__args__[0]
+                    outer = _strip_extras(origin)
                     if outer in (
                         collections.abc.Sequence,
                         collections.abc.MutableSequence,

libs/langgraph/tests/test_channels.py

@@ -438,6 +438,7 @@ def test_delta_channel_dict_reducer_overwrite_in_update() -> None:
 def test_delta_channel_dict_reducer_overwrite_in_writes_replay() -> None:
     """Overwrite(dict) embedded in DeltaChannelWrites must reconstruct as dict."""
     from langgraph.checkpoint.base import DeltaChannelWrites
+
     from langgraph.types import Overwrite
 
     def merge_dicts(left: dict, right: dict) -> dict:
@@ -479,6 +480,36 @@ def test_delta_channel_dict_reducer_snapshot_write_preserves_shape() -> None:
     assert isinstance(w.value, dict)
 
 
+def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None:
+    """DeltaChannel infers dict type through `Annotated[NotRequired[dict[...]], ch]`.
+
+    This is the shape the deepagents filesystem middleware uses for its
+    `files` field; without unwrapping NotRequired we'd fall through to `list`
+    and blow up on the first dict operator call.
+    """
+    from typing import Annotated
+
+    from typing_extensions import NotRequired
+
+    from langgraph.channels.delta import DeltaChannel
+    from langgraph.graph.state import _get_channel
+
+    def merge_dicts(left: dict | None, right: dict) -> dict:
+        if left is None:
+            return dict(right)
+        return {**left, **right}
+
+    annotation = Annotated[
+        NotRequired[dict[str, int]],
+        DeltaChannel(merge_dicts),
+    ]
+    ch = _get_channel("files", annotation).from_checkpoint(MISSING)
+    assert ch.get() == {}
+    ch.update([{"a": 1}])
+    ch.update([{"b": 2}])
+    assert ch.get() == {"a": 1, "b": 2}
+
+
 def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
     """End-to-end: graph with dict-reducer (filesystem-style) channel wrapped in DeltaChannel.
 
@@ -487,12 +518,13 @@ def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
     """
     from typing import Annotated
 
-    from langgraph.channels.delta import DeltaChannel
     from langgraph.checkpoint.base import DELTA_SENTINEL, DeltaChannelWrites
     from langgraph.checkpoint.memory import InMemorySaver
-    from langgraph.graph import START, StateGraph
     from typing_extensions import TypedDict
 
+    from langgraph.channels.delta import DeltaChannel
+    from langgraph.graph import START, StateGraph
+
     def merge_files(left: dict | None, right: dict) -> dict:
         if left is None:
             return {k: v for k, v in right.items() if v is not None}