fix(delta-channel): support non-list reducers (dict) and fix MISSING handling

Use typ() instead of [] throughout DeltaChannel so reducers over dict
(and other non-list types) work correctly. fromCheckpoint(MISSING) now
leaves value as typ() from __init__ instead of overwriting with MISSING.
copy() uses value.copy() to handle dicts. update() initialises base from
typ() when value is MISSING. Add four tests covering the deepagents-style
dict-merge / file-deletion reducer pattern.

libs/langgraph/langgraph/channels/delta.py

@@ -101,7 +101,7 @@ class DeltaChannel(Generic[Value], BaseChannel[list[Value], Value, DeltaValue]):
     def copy(self) -> Self:
         new = DeltaChannel(self.operator, self.typ, snapshot_every=self.snapshot_every)
         new.key = self.key
-        new.value = self.value[:]
+        new.value = self.value if self.value is MISSING else self.value.copy()
         new._pending = self._pending[:]
         new._base_version = self._base_version
         new._last_checkpoint_id = self._last_checkpoint_id
@@ -113,9 +113,11 @@ class DeltaChannel(Generic[Value], BaseChannel[list[Value], Value, DeltaValue]):
         new = DeltaChannel(self.operator, self.typ, snapshot_every=self.snapshot_every)
         new.key = self.key
         if checkpoint is MISSING:
-            new.value = []
+            pass
         elif isinstance(checkpoint, DeltaChainValue):
-            accumulated: list[Value] = list(checkpoint.base) if checkpoint.base else []
+            accumulated: list[Value] = (
+                checkpoint.base if checkpoint.base is not None else new.typ()
+            )
             for step_writes in checkpoint.deltas:
                 for write in step_writes:
                     accumulated = new.operator(accumulated, write)
@@ -159,13 +161,14 @@ class DeltaChannel(Generic[Value], BaseChannel[list[Value], Value, DeltaValue]):
                     )
                     raise InvalidUpdateError(msg)
                 self.value = (
-                    list(overwrite_value) if overwrite_value is not None else []
+                    list(overwrite_value) if overwrite_value is not None else self.typ()
                 )
                 self._pending = list(self.value)
                 self._overwritten = True
                 seen_overwrite = True
             elif not seen_overwrite:
-                self.value = self.operator(self.value, value)
+                base = self.typ() if self.value is MISSING else self.value
+                self.value = self.operator(base, value)
                 self._pending.append(value)
         return True
 

libs/langgraph/tests/test_channels.py

@@ -492,6 +492,107 @@ def test_delta_channel_assembly_fast_path_returns_delta_value() -> None:
     assert result[2].content == "three"
 
 
+def test_delta_channel_dict_reducer_fresh_channel() -> None:
+    """DeltaChannel with a dict reducer starts as empty dict on MISSING checkpoint."""
+    from langgraph.channels.delta import DeltaChannel
+
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
+
+    ch = DeltaChannel(merge_dicts, dict).from_checkpoint(MISSING)
+    # Should be available (not raise EmptyChannelError) and start empty
+    assert ch.is_available()
+    assert ch.get() == {}
+
+
+def test_delta_channel_dict_reducer_basic_updates() -> None:
+    """DeltaChannel with a dict reducer accumulates key/value pairs across steps."""
+    from langgraph.checkpoint.base import DeltaValue
+
+    from langgraph.channels.delta import DeltaChannel
+
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
+
+    ch = DeltaChannel(merge_dicts, dict).from_checkpoint(MISSING)
+    ch.after_checkpoint(None)
+
+    ch.update([{"a": 1}])
+    d1 = ch.checkpoint()
+    assert isinstance(d1, DeltaValue)
+    assert d1.delta == [{"a": 1}]
+    ch.after_checkpoint("v1", checkpoint_id="cid1")
+
+    ch.update([{"b": 2}])
+    d2 = ch.checkpoint()
+    assert d2.delta == [{"b": 2}]
+    assert d2.prev_checkpoint_id == "cid1"
+    ch.after_checkpoint("v2")
+
+    assert ch.get() == {"a": 1, "b": 2}
+
+
+def test_delta_channel_dict_reducer_chain_reconstruction() -> None:
+    """DeltaChainValue replays correctly through a dict merge reducer."""
+    from langgraph.checkpoint.base import DeltaChainValue
+
+    from langgraph.channels.delta import DeltaChannel
+
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
+
+    spec = DeltaChannel(merge_dicts, dict)
+    chain = DeltaChainValue(
+        base={"a": 1},
+        deltas=[[{"b": 2}], [{"c": 3}]],
+    )
+    ch = spec.from_checkpoint(chain)
+    assert ch.get() == {"a": 1, "b": 2, "c": 3}
+    assert ch._steps_since_snapshot == 2
+
+
+def test_delta_channel_dict_reducer_with_deletions() -> None:
+    """Dict reducer that treats None values as deletions works end-to-end (deepagents pattern)."""
+    from langgraph.checkpoint.base import DeltaChainValue
+
+    from langgraph.channels.delta import DeltaChannel
+
+    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}
+        result = {**left}
+        for k, v in right.items():
+            if v is None:
+                result.pop(k, None)
+            else:
+                result[k] = v
+        return result
+
+    ch = DeltaChannel(merge_files, dict).from_checkpoint(MISSING)
+    ch.after_checkpoint(None)
+
+    ch.update([{"file1.py": "content1", "file2.py": "content2"}])
+    ch.after_checkpoint("v1", checkpoint_id="cid1")
+
+    # Delete file1, add file3
+    ch.update([{"file1.py": None, "file3.py": "content3"}])
+    ch.after_checkpoint("v2", checkpoint_id="cid2")
+
+    assert ch.get() == {"file2.py": "content2", "file3.py": "content3"}
+
+    # Confirm chain reconstruction produces the same result
+    chain = DeltaChainValue(
+        base={},
+        deltas=[
+            [{"file1.py": "content1", "file2.py": "content2"}],
+            [{"file1.py": None, "file3.py": "content3"}],
+        ],
+    )
+    spec = DeltaChannel(merge_files, dict)
+    ch2 = spec.from_checkpoint(chain)
+    assert ch2.get() == {"file2.py": "content2", "file3.py": "content3"}
+
+
 def test_delta_channel_assembly_broken_chain_logs_warning() -> None:
     """If a prev_checkpoint_id points to a missing checkpoint, log a warning and use partial chain."""
     from unittest.mock import MagicMock