fix(langgraph): treat DeltaChannel Overwrite as a hard reset in update()

DeltaChannel.update() kept writes ordered before an Overwrite within the
same super-step, while replay_writes() dropped them, so reconstructing a
thread from a checkpoint produced different state than the live run. Align
update() with replay_writes() (drop everything up to and including the
overwrite) so reload reproduces live state.

libs/langgraph/langgraph/channels/delta.py

@@ -177,7 +177,12 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
                 if overwrite_value is not None
                 else self.typ()
             )
-            remaining = [v for i, v in enumerate(values) if i != overwrite_idx]
+            # Treat Overwrite as a hard reset: drop everything up to and
+            # including the overwrite, keeping only writes that follow it. This
+            # mirrors replay_writes so reconstruction from a checkpoint
+            # reproduces the live state even when a plain write precedes the
+            # Overwrite in the same super-step.
+            remaining = list(values[overwrite_idx + 1 :])
             self.value = self.reducer(base, remaining) if remaining else base
             return True
         base = self.typ() if self.value is MISSING else self.value

libs/langgraph/langgraph/pregel/_remote_run_stream.py

@@ -11,6 +11,7 @@ from langchain_core.runnables import RunnableConfig
 from langgraph_sdk._async.stream import AsyncThreadStream
 from langgraph_sdk._sync.stream import SyncThreadStream
 from langgraph_sdk.client import LangGraphClient, SyncLangGraphClient
+from langgraph_sdk.stream.decoders import DataDecoder
 
 from langgraph.types import Command
 
@@ -43,9 +44,10 @@ def _translate_command_input(input: Any) -> Any:
 class _ChannelProjection:
     """Decoded projection for a wire channel the SDK doesn't type natively.
 
-    Subscribes to `channel` and yields each event's `params["data"]` — the same
-    item shape the SDK's typed projections yield (`_ValuesProjection` etc.) and
-    that local's `UpdatesTransformer` / `CheckpointsTransformer` /
+    Subscribes to `channel` and decodes each event's `params["data"]` through the
+    SDK's `DataDecoder` — the same decoder the SDK's own plain-payload projections
+    (`values` / `updates` / `checkpoints` / `tasks`) use, which yields the item
+    shape that local's `UpdatesTransformer` / `CheckpointsTransformer` /
     `TasksTransformer` / `CustomTransformer` push, so iterating this matches the
     corresponding local projection. Iterate with `for` against a sync stream and
     `async for` against an async stream (matching the underlying SDK). Opening
@@ -56,30 +58,23 @@ class _ChannelProjection:
         self._sdk = sdk
         self._channel = channel
 
-    @staticmethod
-    def _data(event: Any) -> Any:
-        """Extract `params.data` from a protocol event, tolerating odd shapes."""
-        params = event.get("params") if isinstance(event, dict) else None
-        return params.get("data") if isinstance(params, dict) else None
-
     def __iter__(self) -> Iterator[Any]:
         # Sync lane: the sync adapter's SDK returns a sync iterator here.
+        decoder = DataDecoder(self._channel)
         events = cast(Iterator[Any], self._sdk.subscribe([self._channel]))
         for event in events:
-            data = self._data(event)
-            if data is not None:
-                yield data
+            yield from decoder.feed(event)
 
     def __aiter__(self) -> AsyncIterator[Any]:
         return self._aiter()
 
     async def _aiter(self) -> AsyncIterator[Any]:
         # Async lane: the async adapter's SDK returns an async iterator here.
+        decoder = DataDecoder(self._channel)
         events = cast(AsyncIterator[Any], self._sdk.subscribe([self._channel]))
         async for event in events:
-            data = self._data(event)
-            if data is not None:
-                yield data
+            for item in decoder.feed(event):
+                yield item
 
 
 class _ProjectionRegistry(Mapping[str, Any]):
@@ -243,7 +238,7 @@ class _RemoteGraphRunStream:
         return self._events_iter
 
     def interleave(self, *names: str) -> Iterator[tuple[str, Any]]:
-        raise NotImplementedError
+        yield from self._sdk.interleave_projections(list(names))
 
 
 class _AsyncRemoteGraphRunStream:

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.2.2"
+version = "1.2.3"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"
@@ -26,7 +26,7 @@ classifiers = [
 dependencies = [
     "langchain-core>=1.4.0,<2",
     "langgraph-checkpoint>=4.1.0,<5.0.0",
-    "langgraph-sdk>=0.4.0,<0.5.0",
+    "langgraph-sdk>=0.4.2,<0.5.0",
     "langgraph-prebuilt>=1.1.0,<1.2.0",
     "xxhash>=3.5.0",
     "pydantic>=2.7.4",

libs/langgraph/tests/test_channels.py

@@ -186,6 +186,38 @@ def test_delta_channel_overwrite() -> None:
     assert ch.get()[0].content == "new"
 
 
+def test_delta_channel_overwrite_after_plain_write_in_one_step() -> None:
+    """Regression: a plain write ordered BEFORE an Overwrite in the same
+    super-step must reconstruct identically to the live state.
+
+    Overwrite is a hard reset: update() and replay_writes() both drop
+    everything up to and including the overwrite, keeping only writes that
+    follow it. Mirrors the JS end-state (channels/delta.ts).
+    """
+
+    def list_reducer(state: list, writes: list) -> list:
+        out = list(state)
+        for w in writes:
+            out.extend(w)
+        return out
+
+    # Live: a single super-step receives [1] then Overwrite([50]).
+    live = DeltaChannel(list_reducer, list).from_checkpoint(MISSING)
+    live.update([[1], Overwrite([50])])
+    assert live.get() == [50]
+
+    # Reload: the same two writes are replayed from the checkpoint.
+    replayed = DeltaChannel(list_reducer, list).from_checkpoint(MISSING)
+    replayed.replay_writes(
+        [
+            ("t1", "messages", [1]),
+            ("t2", "messages", Overwrite([50])),
+        ]
+    )
+    # Invariant: reconstructed state must equal live state.
+    assert replayed.get() == live.get()
+
+
 def test_delta_channel_remove_message_and_replay() -> None:
     """RemoveMessage must round-trip correctly when writes are replayed."""
     spec = DeltaChannel(_messages_delta_reducer, list)

libs/langgraph/tests/test_remote_graph_v3.py

@@ -139,14 +139,15 @@ def test_projection_registry_typed_decoded_and_custom():
 
 def test_channel_projection_decodes_params_data():
     sdk = MagicMock()
-    # Two events with data + one malformed/dataless event that must be skipped.
+    # Real wire events carry `method`; the SDK `DataDecoder` yields matching
+    # events' `params.data` and skips dataless and off-channel ones.
     sdk.subscribe = MagicMock(
         return_value=iter(
             [
-                {"params": {"data": {"n": 1}}},
-                {"params": {}},  # no data -> skipped
-                {"params": {"data": {"n": 2}}},
-                {"unexpected": "shape"},  # not a params dict -> skipped
+                {"method": "checkpoints", "params": {"data": {"n": 1}}},
+                {"method": "checkpoints", "params": {}},  # no data -> skipped
+                {"method": "checkpoints", "params": {"data": {"n": 2}}},
+                {"method": "lifecycle", "params": {"data": {"n": 3}}},  # other channel
             ]
         )
     )
@@ -155,6 +156,39 @@ def test_channel_projection_decodes_params_data():
     sdk.subscribe.assert_called_once_with(["checkpoints"])
 
 
+@pytest.mark.anyio
+async def test_channel_projection_decodes_params_data_async():
+    """Async lane mirrors the sync lane: `async for` over the SDK's async
+    subscription, decoded through the same `DataDecoder`."""
+
+    class _FakeAsyncEvents:
+        def __init__(self, items):
+            self._items = list(items)
+
+        def __aiter__(self):
+            return self
+
+        async def __anext__(self):
+            if not self._items:
+                raise StopAsyncIteration
+            return self._items.pop(0)
+
+    sdk = MagicMock()
+    sdk.subscribe = MagicMock(
+        return_value=_FakeAsyncEvents(
+            [
+                {"method": "checkpoints", "params": {"data": {"n": 1}}},
+                {"method": "checkpoints", "params": {}},  # no data -> skipped
+                {"method": "checkpoints", "params": {"data": {"n": 2}}},
+                {"method": "lifecycle", "params": {"data": {"n": 3}}},  # other channel
+            ]
+        )
+    )
+    proj = _ChannelProjection(sdk, "checkpoints")
+    assert [item async for item in proj] == [{"n": 1}, {"n": 2}]
+    sdk.subscribe.assert_called_once_with(["checkpoints"])
+
+
 def test_sync_adapter_translates_command_input():
     sync_client = MagicMock()
     sdk_thread = MagicMock()
@@ -213,11 +247,14 @@ def test_abort_swallows_cancel_failure_and_still_closes():
         sdk_thread.close.assert_called_once()
 
 
-def test_sync_interleave_raises_not_implemented():
-    adapter, _, _ = _make_sync_adapter()
+def test_sync_interleave_delegates_to_interleave_projections():
+    adapter, _, sdk_thread = _make_sync_adapter()
+    pairs = [("values", {"x": 1}), ("messages", object())]
+    sdk_thread.interleave_projections.return_value = pairs
     with adapter as stream:
-        with pytest.raises(NotImplementedError):
-            list(stream.interleave("messages"))
+        result = list(stream.interleave("values", "messages"))
+    assert result == pairs
+    sdk_thread.interleave_projections.assert_called_once_with(["values", "messages"])
 
 
 def test_async_adapter_has_no_interleave():

libs/langgraph/uv.lock

@@ -1382,7 +1382,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.2"
+version = "1.2.3"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/uv.lock

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

libs/sdk-py/langgraph_sdk/__init__.py

@@ -3,6 +3,6 @@ from langgraph_sdk.client import get_client, get_sync_client
 from langgraph_sdk.encryption import Encryption
 from langgraph_sdk.encryption.types import EncryptionContext
 
-__version__ = "0.4.1"
+__version__ = "0.4.2"
 
 __all__ = ["Auth", "Encryption", "EncryptionContext", "get_client", "get_sync_client"]

libs/sdk-py/langgraph_sdk/stream/transport/http.py

@@ -20,6 +20,7 @@ import httpx
 import orjson
 from langchain_protocol import Event
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.sse import BytesLineDecoder, SSEDecoder
 from langgraph_sdk.stream.transport.base import (
     EventStreamHandle,
@@ -49,8 +50,12 @@ class ProtocolSseTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_url = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_url = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._max_queue_size = max_queue_size
         self._closed = False

libs/sdk-py/langgraph_sdk/stream/transport/sync_http.py

@@ -10,6 +10,7 @@ import httpx
 import orjson
 from langchain_protocol import Event
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.sse import BytesLineDecoder, SSEDecoder
 from langgraph_sdk.stream.transport.base import (
     SyncEventStreamHandle,
@@ -31,8 +32,12 @@ class SyncProtocolSseTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_url = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_url = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._closed = False
         self._open_responses: list[httpx.Response] = []

libs/sdk-py/langgraph_sdk/stream/transport/sync_ws.py

@@ -11,6 +11,7 @@ import orjson
 from langchain_protocol import Event
 from websockets.sync.client import connect as websocket_connect
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.stream.transport.base import (
     SyncEventStreamHandle,
     build_event_stream_body,
@@ -36,8 +37,12 @@ class SyncProtocolWebSocketTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_path = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_path = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._connect = connect
         self._ping_interval = ping_interval

libs/sdk-py/langgraph_sdk/stream/transport/ws.py

@@ -13,6 +13,7 @@ from langchain_protocol import Event
 from websockets.asyncio.client import connect as websocket_connect
 from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.stream.transport.base import (
     EventStreamHandle,
     build_event_stream_body,
@@ -39,8 +40,12 @@ class ProtocolWebSocketTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_path = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_path = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._connect = connect
         self._max_queue_size = max_queue_size

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

@@ -0,0 +1,136 @@
+"""Regression tests for #7953: v3 stream transports must percent-encode
+`thread_id` in their default paths so a value containing reserved characters
+or dot-segments stays an opaque identifier under `/threads/{thread_id}/...`
+instead of being normalized into a different resource path by the HTTP stack.
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from langgraph_sdk.stream.transport.base import build_websocket_url
+from langgraph_sdk.stream.transport.http import ProtocolSseTransport
+from langgraph_sdk.stream.transport.sync_http import SyncProtocolSseTransport
+from langgraph_sdk.stream.transport.sync_ws import SyncProtocolWebSocketTransport
+from langgraph_sdk.stream.transport.ws import ProtocolWebSocketTransport
+
+# A thread_id that escapes the /threads/ namespace if interpolated raw: an HTTP
+# client collapses `/threads/../assistants/abc/...` to `/assistants/abc/...`.
+TRAVERSAL_THREAD_ID = "../assistants/abc"
+ENCODED_COMMANDS_PATH = "/threads/..%2Fassistants%2Fabc/commands"
+ENCODED_STREAM_PATH = "/threads/..%2Fassistants%2Fabc/stream/events"
+
+
+@pytest.mark.anyio
+async def test_async_sse_default_paths_encode_thread_id():
+    transport = ProtocolSseTransport(
+        client=httpx.AsyncClient(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_url == ENCODED_STREAM_PATH
+
+
+def test_sync_sse_default_paths_encode_thread_id():
+    transport = SyncProtocolSseTransport(
+        client=httpx.Client(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_url == ENCODED_STREAM_PATH
+
+
+@pytest.mark.anyio
+async def test_async_ws_default_paths_encode_thread_id():
+    transport = ProtocolWebSocketTransport(
+        client=httpx.AsyncClient(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_path == ENCODED_STREAM_PATH
+
+
+def test_sync_ws_default_paths_encode_thread_id():
+    transport = SyncProtocolWebSocketTransport(
+        client=httpx.Client(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_path == ENCODED_STREAM_PATH
+
+
+async def test_async_sse_wire_path_stays_under_threads_namespace():
+    """The path that actually goes on the wire must not be normalized away."""
+    seen: list[str] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request.url.raw_path.decode("ascii"))
+        return httpx.Response(202)
+
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(handler),
+        base_url="https://example.com",
+        trust_env=False,
+    ) as client:
+        transport = ProtocolSseTransport(client=client, thread_id=TRAVERSAL_THREAD_ID)
+        await transport.send_command({"id": 1, "method": "noop", "params": {}})
+
+    assert seen[0] == ENCODED_COMMANDS_PATH
+
+
+def test_sync_sse_wire_path_stays_under_threads_namespace():
+    seen: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request.url.raw_path.decode("ascii"))
+        return httpx.Response(202)
+
+    with httpx.Client(
+        transport=httpx.MockTransport(handler),
+        base_url="https://example.com",
+        trust_env=False,
+    ) as client:
+        transport = SyncProtocolSseTransport(
+            client=client, thread_id=TRAVERSAL_THREAD_ID
+        )
+        transport.send_command({"id": 1, "method": "noop", "params": {}})
+
+    assert seen[0] == ENCODED_COMMANDS_PATH
+
+
+@pytest.mark.anyio
+async def test_async_ws_url_stays_under_threads_namespace():
+    transport = ProtocolWebSocketTransport(
+        client=httpx.AsyncClient(base_url="https://example.com/api"),
+        thread_id=TRAVERSAL_THREAD_ID,
+    )
+    url = build_websocket_url(transport._client.base_url, transport._stream_path)
+    assert url == "wss://example.com/api/threads/..%2Fassistants%2Fabc/stream/events"
+
+
+def test_sync_ws_url_stays_under_threads_namespace():
+    transport = SyncProtocolWebSocketTransport(
+        client=httpx.Client(base_url="https://example.com/api"),
+        thread_id=TRAVERSAL_THREAD_ID,
+    )
+    url = build_websocket_url(transport._client.base_url, transport._stream_path)
+    assert url == "wss://example.com/api/threads/..%2Fassistants%2Fabc/stream/events"
+
+
+@pytest.mark.anyio
+async def test_explicit_path_overrides_are_left_untouched():
+    """Callers passing explicit paths opt out of default encoding entirely."""
+    sse = ProtocolSseTransport(
+        client=httpx.AsyncClient(),
+        thread_id=TRAVERSAL_THREAD_ID,
+        commands_path="/custom/commands",
+        stream_path="/custom/events",
+    )
+    assert sse._commands_url == "/custom/commands"
+    assert sse._stream_url == "/custom/events"
+
+    ws = ProtocolWebSocketTransport(
+        client=httpx.AsyncClient(),
+        thread_id=TRAVERSAL_THREAD_ID,
+        commands_path="/custom/commands",
+        stream_path="/custom/events",
+    )
+    assert ws._commands_url == "/custom/commands"
+    assert ws._stream_path == "/custom/events"

libs/sdk-py/uv.lock

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