From 43ecd9dc2ded9e0b125798b67bdb30cef73ceea7 Mon Sep 17 00:00:00 2001 From: Sydney Runkle Date: Tue, 21 Apr 2026 13:04:40 -0400 Subject: [PATCH] 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 | 13 ++- libs/langgraph/tests/test_channels.py | 101 +++++++++++++++++++++ 2 files changed, 109 insertions(+), 5 deletions(-) diff --git a/libs/langgraph/langgraph/channels/delta.py b/libs/langgraph/langgraph/channels/delta.py index d8548c2b5..7c17c6ad6 100644 --- a/libs/langgraph/langgraph/channels/delta.py +++ b/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 diff --git a/libs/langgraph/tests/test_channels.py b/libs/langgraph/tests/test_channels.py index 231f491d8..3cc328c11 100644 --- a/libs/langgraph/tests/test_channels.py +++ b/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