fix: preserve full root messages inbox

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -102,6 +102,11 @@ def create_checkpoint(
                 continue
             ch = channels[k]
             if k in channels_to_snapshot:
+                # Callers force a full snapshot blob here: exit mode when a
+                # delta channel reaches its snapshot cadence, and update_state
+                # on a fresh thread (no ancestor to replay writes from). The
+                # manual version-bump below only applies to the exit-mode case.
+                #
                 # In exit mode, the snapshot decision is deferred to exit
                 # time (intermediate steps have do_checkpoint=False). The
                 # channel's count may have reached snapshot_frequency over

libs/langgraph/langgraph/pregel/main.py

@@ -1997,33 +1997,11 @@ class Pregel(
                     ),
                 )
             # save task writes
-            has_delta_writes = any(
-                isinstance(channels.get(c), DeltaChannel)
-                for task in run_tasks
-                for c, _ in task.writes
-            )
-            should_put_writes = saved is not None or has_delta_writes
-
-            if saved is None and has_delta_writes:
-                # If there is no previous checkpoint, we need to create a stub checkpoint
-                # so the first delta writes has a parent to anchor under.
-                # This is the model of DeltaChannel.
-                stub = empty_checkpoint()
-                checkpoint_config = checkpointer.put(
-                    patch_configurable(
-                        checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
-                    ),
-                    stub,
-                    {"source": "update", "step": -1, "parents": {}},
-                    {},
-                )
-
             for task_id, task in zip(run_task_ids, run_tasks):
                 # channel writes are saved to current checkpoint
                 channel_writes = [w for w in task.writes if w[0] != PUSH]
-                if should_put_writes and channel_writes:
+                if saved and channel_writes:
                     checkpointer.put_writes(checkpoint_config, channel_writes, task_id)
-
             # apply to checkpoint and save
             apply_writes(
                 checkpoint,
@@ -2032,7 +2010,21 @@ class Pregel(
                 checkpointer.get_next_version,
                 self.trigger_to_nodes,
             )
-            checkpoint = create_checkpoint(checkpoint, channels, step + 1)
+            # On a fresh thread there is no ancestor to replay DeltaChannel
+            # writes from, so force a self-contained snapshot in the first
+            # checkpoint instead of relying on ancestor write-replay.
+            delta_snapshot = (
+                {
+                    k
+                    for k, ch in channels.items()
+                    if isinstance(ch, DeltaChannel) and ch.is_available()
+                }
+                if saved is None
+                else None
+            )
+            checkpoint = create_checkpoint(
+                checkpoint, channels, step + 1, channels_to_snapshot=delta_snapshot
+            )
             next_config = checkpointer.put(
                 checkpoint_config,
                 checkpoint,
@@ -2464,31 +2456,10 @@ class Pregel(
                     ),
                 )
             # save task writes
-            has_delta_writes = any(
-                isinstance(channels.get(c), DeltaChannel)
-                for task in run_tasks
-                for c, _ in task.writes
-            )
-            should_put_writes = saved is not None or has_delta_writes
-
-            if saved is None and has_delta_writes:
-                # If there is no previous checkpoint, we need to create a stub checkpoint
-                # so the first delta writes has a parent to anchor under.
-                # This is the model of DeltaChannel.
-                stub = empty_checkpoint()
-                checkpoint_config = await checkpointer.aput(
-                    patch_configurable(
-                        checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
-                    ),
-                    stub,
-                    {"source": "update", "step": -1, "parents": {}},
-                    {},
-                )
-
             for task_id, task in zip(run_task_ids, run_tasks):
                 # channel writes are saved to current checkpoint
                 channel_writes = [w for w in task.writes if w[0] != PUSH]
-                if should_put_writes and channel_writes:
+                if saved and channel_writes:
                     await checkpointer.aput_writes(
                         checkpoint_config, channel_writes, task_id
                     )
@@ -2500,7 +2471,21 @@ class Pregel(
                 checkpointer.get_next_version,
                 self.trigger_to_nodes,
             )
-            checkpoint = create_checkpoint(checkpoint, channels, step + 1)
+            # On a fresh thread there is no ancestor to replay DeltaChannel
+            # writes from, so force a self-contained snapshot in the first
+            # checkpoint instead of relying on ancestor write-replay.
+            delta_snapshot = (
+                {
+                    k
+                    for k, ch in channels.items()
+                    if isinstance(ch, DeltaChannel) and ch.is_available()
+                }
+                if saved is None
+                else None
+            )
+            checkpoint = create_checkpoint(
+                checkpoint, channels, step + 1, channels_to_snapshot=delta_snapshot
+            )
             # save checkpoint, after applying writes
             next_config = await checkpointer.aput(
                 checkpoint_config,

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.2.7"
+version = "1.2.8"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"

libs/langgraph/tests/test_delta_channel_update_state.py

@@ -2,10 +2,13 @@
 
 Originally a regression suite for deepagents#3774 — `update_state` on a *fresh*
 thread silently dropped the first write to a `DeltaChannel`-backed channel
-because channel writes were only persisted when a previous checkpoint existed.
-Fixed by lazily persisting an empty stub checkpoint on a fresh thread so the
-first write has a parent to anchor under (mirrors the exit-mode lazy-stub
-pattern in `_loop._put_exit_delta_writes`).
+because channel writes were only persisted when a previous checkpoint existed
+and no snapshot was written either, so the checkpoint reconstructed to empty.
+
+Fixed by forcing a self-contained `_DeltaSnapshot` blob into the first
+checkpoint on a fresh thread (`saved is None`), so the value is stored inline
+and no ancestor write-replay is required. This keeps the read/replay path
+untouched.
 
 Coverage:
 
@@ -13,8 +16,8 @@ Coverage:
 * non-fresh thread: `update_state` after `invoke`, after another `update_state`,
   and `bulk_update_state` with multiple per-superstep updates
 * update-by-id end-to-end via `update_state` (DeltaChannel reducer semantics)
-* state-history chain shape on a fresh thread (lazy stub + update checkpoint
-  with correct parent linking)
+* state-history chain shape on a fresh thread (single self-contained update
+  checkpoint with the snapshot inline and no parent)
 """
 
 from typing import Annotated, Any
@@ -94,7 +97,7 @@ async def test_aupdate_state_fresh_thread_delta_channel() -> None:
 
 def test_update_state_after_invoke_delta_channel() -> None:
     """The non-fresh-thread path was already working before the fix; pin it
-    down so the lazy-stub change for fresh threads doesn't regress it."""
+    down so the forced-snapshot change for fresh threads doesn't regress it."""
     saver = InMemorySaver()
     graph = _build_graph(saver)
     config = {"configurable": {"thread_id": "after-invoke-sync"}}
@@ -133,9 +136,9 @@ async def test_aupdate_state_after_invoke_delta_channel() -> None:
 
 
 def test_consecutive_update_states_delta_channel() -> None:
-    """First update_state lazily persists a stub; the second sees a real
-    parent (`saved is not None`) and takes the original write path. Both
-    messages must round-trip in chronological order."""
+    """First update_state forces a self-contained snapshot seed; the second
+    sees a real parent (`saved is not None`) and anchors its writes under that
+    seed. Both messages must round-trip in chronological order."""
     saver = InMemorySaver()
     graph = _build_graph(saver)
     config = {"configurable": {"thread_id": "consecutive-sync"}}
@@ -252,14 +255,14 @@ def test_bulk_update_state_multi_task_per_superstep_delta_channel() -> None:
 
 
 # ---------------------------------------------------------------------------
-# Public-API observation of the lazy-stub mechanism
+# Public-API observation of the forced-snapshot mechanism
 # ---------------------------------------------------------------------------
 
 
 def test_state_history_chain_after_fresh_update_state_delta_channel() -> None:
-    """A fresh-thread `update_state` should produce two checkpoints visible
-    via `get_state_history`: a stub (step=-1, no parent) and the update
-    (step=0, parent=stub). Both attributed `source='update'`."""
+    """A fresh-thread `update_state` should produce a single self-contained
+    checkpoint visible via `get_state_history`: step=0, `source='update'`,
+    no parent, with the DeltaChannel value snapshotted inline."""
     saver = InMemorySaver()
     graph = _build_graph(saver)
     config = {"configurable": {"thread_id": "history-chain"}}
@@ -270,25 +273,12 @@ def test_state_history_chain_after_fresh_update_state_delta_channel() -> None:
         as_node="model",
     )
 
-    # Newest first per `get_state_history` ordering.
     history = list(graph.get_state_history(config))
-    assert len(history) == 2
-
-    update_snapshot, stub_snapshot = history
+    assert len(history) == 1
 
+    (update_snapshot,) = history
     assert update_snapshot.metadata is not None
     assert update_snapshot.metadata["source"] == "update"
     assert update_snapshot.metadata["step"] == 0
+    assert update_snapshot.parent_config is None
     assert [m.content for m in update_snapshot.values["messages"]] == ["hello"]
-
-    assert stub_snapshot.metadata is not None
-    assert stub_snapshot.metadata["source"] == "update"
-    assert stub_snapshot.metadata["step"] == -1
-    assert stub_snapshot.parent_config is None
-
-    # The update checkpoint's parent is the stub.
-    assert update_snapshot.parent_config is not None
-    assert (
-        update_snapshot.parent_config["configurable"]["checkpoint_id"]
-        == stub_snapshot.config["configurable"]["checkpoint_id"]
-    )

libs/langgraph/uv.lock

@@ -1438,7 +1438,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.7"
+version = "1.2.8"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/uv.lock

@@ -285,7 +285,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.7"
+version = "1.2.8"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/langgraph_sdk/_async/stream.py

@@ -911,6 +911,34 @@ class _SubgraphsProjection:
     def __aiter__(self) -> AsyncIterator[ScopedStreamHandle]:
         return self._subgraphs_iter()
 
+    @staticmethod
+    def _put_root_message(root_inbox: asyncio.Queue[Event | None], item: Event) -> None:
+        if root_inbox.maxsize > 0 and root_inbox.qsize() >= root_inbox.maxsize - 1:
+            raise RuntimeError(
+                "Root messages inbox exceeded max_queue_size while buffering "
+                "root-scope messages. Iterate thread.messages concurrently "
+                "or increase max_queue_size."
+            )
+        try:
+            root_inbox.put_nowait(item)
+        except asyncio.QueueFull as exc:
+            raise RuntimeError(
+                "Root messages inbox exceeded max_queue_size while buffering "
+                "root-scope messages. Iterate thread.messages concurrently "
+                "or increase max_queue_size."
+            ) from exc
+
+    @staticmethod
+    def _signal_root_inbox_closed(root_inbox: asyncio.Queue[Event | None]) -> None:
+        try:
+            root_inbox.put_nowait(None)
+        except asyncio.QueueFull as exc:
+            raise RuntimeError(
+                "Root messages inbox exceeded max_queue_size while closing "
+                "root-scope messages. Iterate thread.messages concurrently "
+                "or increase max_queue_size."
+            ) from exc
+
     async def _subgraphs_iter(self) -> AsyncGenerator[ScopedStreamHandle, None]:
         if self._thread._transport is None:
             raise RuntimeError("AsyncThreadStream not entered - use `async with`.")
@@ -943,7 +971,7 @@ class _SubgraphsProjection:
                     and item.get("method") == "messages"
                     and tuple(_event_namespace(params_field)) == self._scope
                 ):
-                    root_inbox.put_nowait(item)
+                    self._put_root_message(root_inbox, item)
                 for handle in decoder.feed(item):
                     yield handle
         finally:
@@ -961,7 +989,7 @@ class _SubgraphsProjection:
                     handle._finish(terminal_status)
             self._thread._unregister_subscription(sub.id)
             if root_inbox is not None:
-                root_inbox.put_nowait(None)
+                self._signal_root_inbox_closed(root_inbox)
 
 
 class ToolCallHandle:
@@ -1335,7 +1363,8 @@ class AsyncThreadStream:
         that arrive at namespace `[]` before `thread.messages` has subscribed.
         """
         if self._root_messages_inbox is None:
-            self._root_messages_inbox = asyncio.Queue()
+            maxsize = self._max_queue_size + 1 if self._max_queue_size > 0 else 0
+            self._root_messages_inbox = asyncio.Queue(maxsize=maxsize)
         return self._root_messages_inbox
 
     def _register_active_message_stream(self, stream: AsyncChatModelStream) -> None:

libs/sdk-py/langgraph_sdk/_sync/stream.py

@@ -954,6 +954,34 @@ class _SyncSubgraphsProjection:
     def __iter__(self) -> Iterator[SyncScopedStreamHandle]:
         return self._subgraphs_iter()
 
+    @staticmethod
+    def _put_root_message(root_inbox: queue.Queue[Event | None], item: Event) -> None:
+        if root_inbox.maxsize > 0 and root_inbox.qsize() >= root_inbox.maxsize - 1:
+            raise RuntimeError(
+                "Root messages inbox exceeded max_queue_size while buffering "
+                "root-scope messages. Iterate thread.messages concurrently "
+                "or increase max_queue_size."
+            )
+        try:
+            root_inbox.put_nowait(item)
+        except queue.Full as exc:
+            raise RuntimeError(
+                "Root messages inbox exceeded max_queue_size while buffering "
+                "root-scope messages. Iterate thread.messages concurrently "
+                "or increase max_queue_size."
+            ) from exc
+
+    @staticmethod
+    def _signal_root_inbox_closed(root_inbox: queue.Queue[Event | None]) -> None:
+        try:
+            root_inbox.put_nowait(None)
+        except queue.Full as exc:
+            raise RuntimeError(
+                "Root messages inbox exceeded max_queue_size while closing "
+                "root-scope messages. Iterate thread.messages concurrently "
+                "or increase max_queue_size."
+            ) from exc
+
     def _subgraphs_iter(self) -> Iterator[SyncScopedStreamHandle]:
         if self._thread._transport is None:
             raise RuntimeError("SyncThreadStream not entered — use `with`.")
@@ -986,7 +1014,7 @@ class _SyncSubgraphsProjection:
                     and item.get("method") == "messages"
                     and tuple(_event_namespace(params_field)) == self._scope
                 ):
-                    root_inbox.put_nowait(item)
+                    self._put_root_message(root_inbox, item)
                 for handle in decoder.feed(cast(dict[str, Any], item)):
                     yield handle
         finally:
@@ -1007,7 +1035,7 @@ class _SyncSubgraphsProjection:
                     handle._finish(terminal_status)
             self._thread._unregister_subscription(sub.id)
             if root_inbox is not None:
-                root_inbox.put_nowait(None)
+                self._signal_root_inbox_closed(root_inbox)
 
 
 class _SyncExtensionsProjection:
@@ -1222,7 +1250,7 @@ class SyncThreadStream:
 
     def _activate_root_messages_inbox(self) -> queue.Queue[Event | None]:
         if self._root_messages_inbox is None:
-            self._root_messages_inbox = queue.Queue()
+            self._root_messages_inbox = queue.Queue(maxsize=1025)
         return self._root_messages_inbox
 
     def _register_active_message_stream(self, stream: ChatModelStream) -> None:

libs/sdk-py/tests/streaming/test_scoped_handles.py

@@ -2,7 +2,13 @@
 
 from __future__ import annotations
 
+import asyncio
+from typing import cast
+from unittest.mock import MagicMock
+
 import httpx
+import pytest
+from langchain_protocol import Event
 
 from langgraph_sdk._async.http import HttpClient
 from langgraph_sdk._async.threads import ThreadsClient
@@ -447,6 +453,56 @@ def test_scoped_handle_inboxes_bounded_by_max_queue_size():
     assert handle._tasks_inbox.maxsize == 16
 
 
+def test_root_messages_inbox_bounded_by_max_queue_size():
+    """Root messages inbox must reserve one terminal sentinel slot."""
+    from langgraph_sdk._async.stream import AsyncThreadStream
+
+    thread = AsyncThreadStream(
+        http=MagicMock(),
+        thread_id="t-1",
+        assistant_id="agent",
+        max_queue_size=16,
+    )
+
+    inbox = thread._activate_root_messages_inbox()
+
+    assert inbox.maxsize == 17
+
+
+def test_subgraphs_root_message_overflow_raises_runtime_error():
+    """Overflowing the root messages inbox must fail explicitly."""
+    from langgraph_sdk._async.stream import _SubgraphsProjection
+
+    inbox: asyncio.Queue[Event | None] = asyncio.Queue(maxsize=2)
+    inbox.put_nowait(cast(Event, message_start_event(seq=1, message_id="msg-1")))
+
+    with pytest.raises(RuntimeError, match="Root messages inbox exceeded"):
+        _SubgraphsProjection._put_root_message(
+            inbox,
+            cast(
+                Event,
+                message_text_delta_event(seq=2, text="overflow", message_id="msg-1"),
+            ),
+        )
+
+
+def test_subgraphs_root_message_close_preserves_full_inbox():
+    """Closing a full root messages inbox must not evict buffered events."""
+    from langgraph_sdk._async.stream import _SubgraphsProjection
+
+    inbox: asyncio.Queue[Event | None] = asyncio.Queue(maxsize=3)
+    first = cast(Event, message_start_event(seq=1, message_id="msg-1"))
+    second = cast(Event, message_finish_event(seq=2, message_id="msg-1"))
+    inbox.put_nowait(first)
+    inbox.put_nowait(second)
+
+    _SubgraphsProjection._signal_root_inbox_closed(inbox)
+
+    assert inbox.get_nowait() is first
+    assert inbox.get_nowait() is second
+    assert inbox.get_nowait() is None
+
+
 async def test_child_handle_inherits_max_queue_size_from_parent():
     """Grandchild ScopedStreamHandles created by _HandleSubgraphsProjection
     inherit the parent's max_queue_size so all queues are consistently bounded."""

libs/sdk-py/tests/streaming/test_sync_projections.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import queue
 from typing import Any, cast
 
 import httpx
@@ -398,6 +399,53 @@ def test_sync_tool_calls_run_error_fails_active_handle():
 # ---------------------------------------------------------------------------
 
 
+def test_sync_root_messages_inbox_is_bounded():
+    """Root messages inbox must reserve one terminal sentinel slot."""
+    fake = SyncFakeServer()
+    fake.script([lifecycle_completed_event(seq=1)])
+    fake.set_state({})
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            inbox = thread._activate_root_messages_inbox()
+
+    assert inbox.maxsize == 1025
+
+
+def test_sync_subgraphs_root_message_overflow_raises_runtime_error():
+    """Overflowing the root messages inbox must fail explicitly."""
+    from langgraph_sdk._sync.stream import _SyncSubgraphsProjection
+
+    inbox: queue.Queue[Event | None] = queue.Queue(maxsize=2)
+    inbox.put_nowait(cast(Event, message_start_event(seq=1, message_id="msg-1")))
+
+    with pytest.raises(RuntimeError, match="Root messages inbox exceeded"):
+        _SyncSubgraphsProjection._put_root_message(
+            inbox,
+            cast(
+                Event,
+                message_text_delta_event(seq=2, text="overflow", message_id="msg-1"),
+            ),
+        )
+
+
+def test_sync_subgraphs_root_message_close_preserves_full_inbox():
+    """Closing a full root messages inbox must not evict buffered events."""
+    from langgraph_sdk._sync.stream import _SyncSubgraphsProjection
+
+    inbox: queue.Queue[Event | None] = queue.Queue(maxsize=3)
+    first = cast(Event, message_start_event(seq=1, message_id="msg-1"))
+    second = cast(Event, message_finish_event(seq=2, message_id="msg-1"))
+    inbox.put_nowait(first)
+    inbox.put_nowait(second)
+
+    _SyncSubgraphsProjection._signal_root_inbox_closed(inbox)
+
+    assert inbox.get_nowait() is first
+    assert inbox.get_nowait() is second
+    assert inbox.get_nowait() is None
+
+
 def test_sync_drain_messages_inbox_pre_dispatches_before_yield():
     """When draining the root inbox, str(message.text) must work immediately on yield."""
     fake = SyncFakeServer()

libs/sdk-py/uv.lock

@@ -298,7 +298,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.7"
+version = "1.2.8"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },