From 7912a1ce50273273e2c6cedbe107cd78f9692aa5 Mon Sep 17 00:00:00 2001 From: Elior Nataf Lackritz Date: Wed, 5 Aug 2026 14:06:33 -0400 Subject: [PATCH] fix(langgraph): hydrate subgraph delta channels with resolved saver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_prepare_state_snapshot` and `bulk_update_state` hydrated channels with `self.checkpointer`, which is `None` for a subgraph — it borrows the parent's saver through `CONFIG_KEY_CHECKPOINTER` at read time. Without a saver a `DeltaChannel` cannot walk ancestors to replay its writes, so it fell through to `from_checkpoint(MISSING)` and hydrated empty, silently. The callers already resolve the right saver a few lines above every call site. Pass it in, as a required keyword argument so no future call site can drop it. On the write path the empty value was persisted as a `_DeltaSnapshot` whenever the update reached the channel's snapshot cadence, losing history on disk. Fixes #8470 Co-authored-by: gururafiki <22777967+gururafiki@users.noreply.github.com> Co-authored-by: Yuan Gao <119447586+DavidGao520@users.noreply.github.com> --- libs/langgraph/langgraph/pregel/main.py | 44 ++- .../tests/test_delta_channel_subgraph.py | 288 ++++++++++++++++++ 2 files changed, 316 insertions(+), 16 deletions(-) create mode 100644 libs/langgraph/tests/test_delta_channel_subgraph.py diff --git a/libs/langgraph/langgraph/pregel/main.py b/libs/langgraph/langgraph/pregel/main.py index f4ca48024..c75211ebd 100644 --- a/libs/langgraph/langgraph/pregel/main.py +++ b/libs/langgraph/langgraph/pregel/main.py @@ -1146,9 +1146,26 @@ class Pregel( self, config: RunnableConfig, saved: CheckpointTuple | None, + *, + saver: BaseCheckpointSaver | None, recurse: BaseCheckpointSaver | None = None, apply_pending_writes: bool = False, ) -> StateSnapshot: + """Assemble a `StateSnapshot` from a saved checkpoint. + + Args: + config: Config identifying the checkpoint being read. + saved: The checkpoint tuple to render, or `None` for an empty snapshot. + saver: Checkpointer to read with, as resolved by the caller from + `CONFIG_KEY_CHECKPOINTER` before falling back to `self.checkpointer`. + Required rather than defaulted because `self.checkpointer` is `None` + for a subgraph — it borrows the parent's saver through the config — + and a `DeltaChannel` silently hydrates empty without one. + recurse: When set, resolve subgraph task states with this checkpointer + instead of returning a config that merely signals they exist. + apply_pending_writes: Apply this checkpoint's pending writes to the + returned values. + """ if not saved: return StateSnapshot( values={}, @@ -1169,9 +1186,7 @@ class Pregel( channels, managed = channels_from_checkpoint( self.channels, saved.checkpoint, - saver=self.checkpointer - if isinstance(self.checkpointer, BaseCheckpointSaver) - else None, + saver=saver, config=saved.config, ) # tasks for this checkpoint @@ -1269,9 +1284,12 @@ class Pregel( self, config: RunnableConfig, saved: CheckpointTuple | None, + *, + saver: BaseCheckpointSaver | None, recurse: BaseCheckpointSaver | None = None, apply_pending_writes: bool = False, ) -> StateSnapshot: + """Async version of `_prepare_state_snapshot`. See docstring there.""" if not saved: return StateSnapshot( values={}, @@ -1292,9 +1310,7 @@ class Pregel( channels, managed = await achannels_from_checkpoint( self.channels, saved.checkpoint, - saver=self.checkpointer - if isinstance(self.checkpointer, BaseCheckpointSaver) - else None, + saver=saver, config=saved.config, ) # tasks for this checkpoint @@ -1429,6 +1445,7 @@ class Pregel( return self._prepare_state_snapshot( config, saved, + saver=checkpointer, recurse=checkpointer if subgraphs else None, apply_pending_writes=CONFIG_KEY_CHECKPOINT_ID not in config[CONF], ) @@ -1473,6 +1490,7 @@ class Pregel( return await self._aprepare_state_snapshot( config, saved, + saver=checkpointer, recurse=checkpointer if subgraphs else None, apply_pending_writes=CONFIG_KEY_CHECKPOINT_ID not in config[CONF], ) @@ -1527,7 +1545,7 @@ class Pregel( checkpointer.list(config, before=before, limit=limit, filter=filter) ): yield self._prepare_state_snapshot( - checkpoint_tuple.config, checkpoint_tuple + checkpoint_tuple.config, checkpoint_tuple, saver=checkpointer ) async def aget_state_history( @@ -1584,7 +1602,7 @@ class Pregel( ) ]: yield await self._aprepare_state_snapshot( - checkpoint_tuple.config, checkpoint_tuple + checkpoint_tuple.config, checkpoint_tuple, saver=checkpointer ) def bulk_update_state( @@ -1666,10 +1684,7 @@ class Pregel( channels, managed = channels_from_checkpoint( self.channels, checkpoint, - saver=self.checkpointer - if saved is not None - and isinstance(self.checkpointer, BaseCheckpointSaver) - else None, + saver=checkpointer if saved is not None else None, config=saved.config if saved is not None else None, ) values, as_node = updates[0][:2] @@ -2132,10 +2147,7 @@ class Pregel( channels, managed = await achannels_from_checkpoint( self.channels, checkpoint, - saver=self.checkpointer - if saved is not None - and isinstance(self.checkpointer, BaseCheckpointSaver) - else None, + saver=checkpointer if saved is not None else None, config=saved.config if saved is not None else None, ) values, as_node = updates[0][:2] diff --git a/libs/langgraph/tests/test_delta_channel_subgraph.py b/libs/langgraph/tests/test_delta_channel_subgraph.py new file mode 100644 index 000000000..d50758512 --- /dev/null +++ b/libs/langgraph/tests/test_delta_channel_subgraph.py @@ -0,0 +1,288 @@ +"""Tests for reading and updating a subgraph's `DeltaChannel` state. + +Regression suite for #8470. + +A subgraph is compiled without a checkpointer of its own — the parent lends it +one through `CONFIG_KEY_CHECKPOINTER` at read time. `DeltaChannel` stores no +value in `channel_values`, so hydrating it requires that saver to walk ancestors +and replay their writes. Reading with `self.checkpointer` instead of the +caller-resolved saver leaves a subgraph with none, and the channel falls through +to `from_checkpoint(MISSING)`: an empty value, indistinguishable from a channel +that was never written and raising nothing. + +Coverage: + +* `get_state` / `aget_state` and `get_state_history` / `aget_state_history` on a + subgraph namespace, at one and two levels of nesting +* `get_state(subgraphs=True)` task states, the surface a human-in-the-loop client + reads while a subgraph sits interrupted +* `update_state` / `aupdate_state`, which persist the hydrated value as a + snapshot blob and so turn an empty hydration into permanent data loss +* root-graph controls, which own their checkpointer and were never affected +""" + +from typing import Annotated, Any + +import pytest +from langgraph.checkpoint.memory import InMemorySaver +from typing_extensions import TypedDict + +from langgraph.channels.delta import DeltaChannel +from langgraph.graph import END, START, StateGraph +from langgraph.types import interrupt + +pytestmark = pytest.mark.anyio + + +def _extend(state: list | None, writes: list[Any]) -> list: + """Batching-invariant list reducer: flattens each write batch onto the state.""" + out = list(state or []) + for write in writes: + out.extend(write if isinstance(write, list) else [write]) + return out + + +def _child_builder(*, snapshot_frequency: int) -> StateGraph: + """Two-node graph writing `["a1"]` then `["b1", "b2"]` to a `DeltaChannel`.""" + + class State(TypedDict, total=False): + msgs: Annotated[ + list, DeltaChannel(_extend, snapshot_frequency=snapshot_frequency) + ] + + builder = StateGraph(State) + builder.add_node("a", lambda state: {"msgs": ["a1"]}) + builder.add_node("b", lambda state: {"msgs": ["b1", "b2"]}) + builder.add_edge(START, "a") + builder.add_edge("a", "b") + builder.add_edge("b", END) + return builder + + +def _wrap(inner: StateGraph, *, node_name: str) -> StateGraph: + """Nest `inner` as a single node, sharing its state schema.""" + builder = StateGraph(inner.state_schema) + builder.add_node(node_name, inner.compile()) + builder.add_edge(START, node_name) + builder.add_edge(node_name, END) + return builder + + +def _build_nested_graph( + checkpointer: InMemorySaver, + *, + snapshot_frequency: int = 1000, + depth: int = 1, +) -> Any: + """Compile a graph whose `child` node is a subgraph nested `depth` levels deep.""" + graph = _child_builder(snapshot_frequency=snapshot_frequency) + for _ in range(depth): + graph = _wrap(graph, node_name="child") + return graph.compile(checkpointer=checkpointer) + + +def _namespaced(config: dict, namespace: str) -> dict: + return {"configurable": {**config["configurable"], "checkpoint_ns": namespace}} + + +def _subgraph_namespace(app: Any, config: dict, *, depth: int = 1) -> str: + """Resolve the `child` namespace `depth` levels down, as a client would. + + Each snapshot's tasks carry `{name, state: {"configurable": {"checkpoint_ns"}}}`, + and passing that namespace back is the documented way to read a subgraph's own + supersteps. + """ + namespace = "" + for level in range(depth): + scoped = _namespaced(config, namespace) if namespace else config + namespace = next( + ( + task.state["configurable"]["checkpoint_ns"] + for snapshot in app.get_state_history(scoped) + for task in snapshot.tasks + if task.name == "child" and isinstance(task.state, dict) + ), + "", + ) + assert namespace, f"no `child` subgraph task at nesting level {level}" + return namespace + + +# --------------------------------------------------------------------------- +# Reading a subgraph's state +# --------------------------------------------------------------------------- + + +def test_subgraph_get_state_replays_delta_channel() -> None: + app = _build_nested_graph(InMemorySaver()) + config = {"configurable": {"thread_id": "1"}} + app.invoke({}, config) + + namespace = _subgraph_namespace(app, config) + + assert app.get_state(config).values == {"msgs": ["a1", "b1", "b2"]} + assert app.get_state(_namespaced(config, namespace)).values == { + "msgs": ["a1", "b1", "b2"] + } + + +async def test_subgraph_aget_state_replays_delta_channel() -> None: + app = _build_nested_graph(InMemorySaver()) + config = {"configurable": {"thread_id": "1"}} + await app.ainvoke({}, config) + + namespace = _subgraph_namespace(app, config) + + assert (await app.aget_state(_namespaced(config, namespace))).values == { + "msgs": ["a1", "b1", "b2"] + } + + +def test_subgraph_get_state_history_replays_delta_channel() -> None: + app = _build_nested_graph(InMemorySaver()) + config = {"configurable": {"thread_id": "1"}} + app.invoke({}, config) + + namespace = _subgraph_namespace(app, config) + values = [ + snapshot.values + for snapshot in app.get_state_history(_namespaced(config, namespace)) + ] + + # newest first: after `b`, after `a`, then the two supersteps preceding any write + assert values == [ + {"msgs": ["a1", "b1", "b2"]}, + {"msgs": ["a1"]}, + {"msgs": []}, + {"msgs": []}, + ] + + +async def test_subgraph_aget_state_history_replays_delta_channel() -> None: + app = _build_nested_graph(InMemorySaver()) + config = {"configurable": {"thread_id": "1"}} + await app.ainvoke({}, config) + + namespace = _subgraph_namespace(app, config) + values = [ + snapshot.values + async for snapshot in app.aget_state_history(_namespaced(config, namespace)) + ] + + assert values == [ + {"msgs": ["a1", "b1", "b2"]}, + {"msgs": ["a1"]}, + {"msgs": []}, + {"msgs": []}, + ] + + +def test_doubly_nested_subgraph_get_state_replays_delta_channel() -> None: + app = _build_nested_graph(InMemorySaver(), depth=2) + config = {"configurable": {"thread_id": "1"}} + app.invoke({}, config) + + namespace = _subgraph_namespace(app, config, depth=2) + + assert app.get_state(_namespaced(config, namespace)).values == { + "msgs": ["a1", "b1", "b2"] + } + + +def test_interrupted_subgraph_task_state_replays_delta_channel() -> None: + """`get_state(subgraphs=True)` resolves task states through a separate + recursion; it reads the paused subgraph a human-in-the-loop client is shown.""" + + class State(TypedDict, total=False): + msgs: Annotated[list, DeltaChannel(_extend)] + + def pause(state: State) -> dict: + interrupt("pause") + return {"msgs": ["b1"]} + + child = StateGraph(State) + child.add_node("a", lambda state: {"msgs": ["a1", "a2"]}) + child.add_node("b", pause) + child.add_edge(START, "a") + child.add_edge("a", "b") + child.add_edge("b", END) + + app = _wrap(child, node_name="child").compile(checkpointer=InMemorySaver()) + config = {"configurable": {"thread_id": "1"}} + app.invoke({}, config) + + task_values = [ + task.state.values + for task in app.get_state(config, subgraphs=True).tasks + if task.name == "child" + ] + + assert task_values == [{"msgs": ["a1", "a2"]}] + + +# --------------------------------------------------------------------------- +# Updating a subgraph's state +# --------------------------------------------------------------------------- + + +def test_subgraph_update_state_preserves_delta_channel_history() -> None: + """`update_state` persists the hydrated channel, so an empty hydration is + written to disk rather than merely returned. + + `snapshot_frequency=2` is what makes that observable: this update reaches the + cadence and forces a `_DeltaSnapshot` blob built from the hydrated value. At + the default frequency nothing is written and the loss stays latent, so the + assertion below would hold either way. + """ + app = _build_nested_graph(InMemorySaver(), snapshot_frequency=2) + config = {"configurable": {"thread_id": "1"}} + app.invoke({}, config) + + namespace = _subgraph_namespace(app, config) + child_config = _namespaced(config, namespace) + app.update_state(child_config, {"msgs": ["manual"]}) + + assert app.get_state(child_config).values == {"msgs": ["a1", "b1", "b2", "manual"]} + + +async def test_subgraph_aupdate_state_preserves_delta_channel_history() -> None: + app = _build_nested_graph(InMemorySaver(), snapshot_frequency=2) + config = {"configurable": {"thread_id": "1"}} + await app.ainvoke({}, config) + + namespace = _subgraph_namespace(app, config) + child_config = _namespaced(config, namespace) + await app.aupdate_state(child_config, {"msgs": ["manual"]}) + + assert (await app.aget_state(child_config)).values == { + "msgs": ["a1", "b1", "b2", "manual"] + } + + +# --------------------------------------------------------------------------- +# Controls: a graph owning its checkpointer resolves the same saver as before +# --------------------------------------------------------------------------- + + +def test_root_graph_get_state_replays_delta_channel() -> None: + app = _child_builder(snapshot_frequency=1000).compile(checkpointer=InMemorySaver()) + config = {"configurable": {"thread_id": "1"}} + app.invoke({}, config) + + assert app.get_state(config).values == {"msgs": ["a1", "b1", "b2"]} + assert [snapshot.values for snapshot in app.get_state_history(config)] == [ + {"msgs": ["a1", "b1", "b2"]}, + {"msgs": ["a1"]}, + {"msgs": []}, + {"msgs": []}, + ] + + +def test_root_graph_update_state_preserves_delta_channel_history() -> None: + app = _child_builder(snapshot_frequency=2).compile(checkpointer=InMemorySaver()) + config = {"configurable": {"thread_id": "1"}} + app.invoke({}, config) + + app.update_state(config, {"msgs": ["manual"]}) + + assert app.get_state(config).values == {"msgs": ["a1", "b1", "b2", "manual"]}