From a0053bb6169b33b1d39c70e919ecbe8f9de43b4d Mon Sep 17 00:00:00 2001 From: Elior Nataf Lackritz Date: Wed, 7 Oct 2026 19:03:46 -0400 Subject: [PATCH 1/2] fix(sdk-py): end a thread stream's run only on a root lifecycle event (#9228) The thread stream's lifecycle watcher ended the run on any `completed` or `failed` lifecycle event, including the one a subgraph sends when it finishes. So `thread.output` could read the thread state before the parent stored the step that ran the subgraph, and a run that failed after a subgraph completed was reported as completed. The async and sync watchers now end the run only on a root event, with the `_is_root_terminal_lifecycle` check the fanout already uses. The JS SDK checks the root namespace here too. This is the `sdk-py integration` flake where the final `items` comes back as `['streamed', 'tool', 'asked']` without `'sub'`: the example graph's last node runs a subgraph. ## Tests New async and sync tests send a subgraph `completed` and then a root `failed`. Without the fix the run ends as completed. `make lint` and `make test` in `libs/sdk-py` pass. --- libs/sdk-py/langgraph_sdk/_async/stream.py | 2 +- libs/sdk-py/langgraph_sdk/_sync/stream.py | 8 ++++++-- .../tests/streaming/test_lifecycle_watcher.py | 20 +++++++++++++++++++ .../streaming/test_sync_thread_stream.py | 18 +++++++++++++++++ 4 files changed, 45 insertions(+), 3 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/_async/stream.py b/libs/sdk-py/langgraph_sdk/_async/stream.py index c26158c11..fc4253192 100644 --- a/libs/sdk-py/langgraph_sdk/_async/stream.py +++ b/libs/sdk-py/langgraph_sdk/_async/stream.py @@ -1972,7 +1972,7 @@ class AsyncThreadStream: # Mark that we have observed an active run so thread.output # knows a run exists (handles reattach without run.start). self._run_seen = True - elif phase in ("completed", "failed"): + elif _is_root_terminal_lifecycle(event): # Why: interrupts describe current-run state. Clear on terminal # lifecycle so a subsequent run.respond() can't fire against a # stale prior-run interrupt_id. Acquire `_interrupts_lock` so diff --git a/libs/sdk-py/langgraph_sdk/_sync/stream.py b/libs/sdk-py/langgraph_sdk/_sync/stream.py index f89dc8224..7b2c6ba03 100644 --- a/libs/sdk-py/langgraph_sdk/_sync/stream.py +++ b/libs/sdk-py/langgraph_sdk/_sync/stream.py @@ -35,7 +35,11 @@ from langgraph_sdk.stream.decoders import ( validate_interleave_channels, ) from langgraph_sdk.stream.subscription import compute_union_filter, infer_channel -from langgraph_sdk.stream.sync_controller import SyncStreamController, _SyncSubscription +from langgraph_sdk.stream.sync_controller import ( + SyncStreamController, + _is_root_terminal_lifecycle, + _SyncSubscription, +) from langgraph_sdk.stream.transport import ( SyncEventStreamHandle, SyncProtocolSseTransport, @@ -1614,7 +1618,7 @@ class SyncThreadStream: phase = data.get("event") if isinstance(data, dict) else None if phase in ("started", "running"): self._run_seen = True - elif phase in ("completed", "failed"): + elif _is_root_terminal_lifecycle(event): self.interrupted = False self.interrupts = [] run_done = self._run_done diff --git a/libs/sdk-py/tests/streaming/test_lifecycle_watcher.py b/libs/sdk-py/tests/streaming/test_lifecycle_watcher.py index 2d759361a..34a4712b0 100644 --- a/libs/sdk-py/tests/streaming/test_lifecycle_watcher.py +++ b/libs/sdk-py/tests/streaming/test_lifecycle_watcher.py @@ -15,6 +15,7 @@ from langgraph_sdk.stream.transport import EventStreamHandle, ProtocolSseTranspo from streaming._events import ( input_requested_event, lifecycle_completed_event, + lifecycle_errored_event, lifecycle_event, ) from streaming._fake_server import FakeServer, _StreamScript @@ -115,6 +116,25 @@ async def test_terminal_lifecycle_clears_interrupts(): assert thread.interrupts == [] +async def test_subgraph_completed_event_does_not_end_run(): + fake = FakeServer() + fake.script( + [ + lifecycle_completed_event(seq=0, namespace=["child:1"]), + lifecycle_errored_event(seq=1, error="root failed"), + ] + ) + asgi = httpx.ASGITransport(app=fake.app) + async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw: + threads = ThreadsClient(HttpClient(raw)) + async with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + run_done = thread._run_done + assert run_done is not None + terminal = await asyncio.wait_for(run_done, timeout=2.0) + assert terminal.status == "errored", "a subgraph's completed event ended the run" + assert "root failed" in str(terminal.error) + + async def test_lifecycle_error_captured_for_output(): """Lifecycle error terminal state is captured in _run_done with error set.""" fake = FakeServer() diff --git a/libs/sdk-py/tests/streaming/test_sync_thread_stream.py b/libs/sdk-py/tests/streaming/test_sync_thread_stream.py index 5eecd74b8..c183cb95a 100644 --- a/libs/sdk-py/tests/streaming/test_sync_thread_stream.py +++ b/libs/sdk-py/tests/streaming/test_sync_thread_stream.py @@ -27,6 +27,7 @@ from streaming._events import ( checkpoints_event, custom_event, lifecycle_completed_event, + lifecycle_errored_event, lifecycle_event, lifecycle_started_event, message_finish_event, @@ -475,6 +476,23 @@ def test_sync_lifecycle_watcher_reconnects_with_since_after_transport_drop(): assert fake.stream_request_bodies[1]["since"] == 1 +def test_sync_subgraph_completed_event_does_not_end_run(): + fake = SyncFakeServer() + fake.script( + [ + lifecycle_completed_event(seq=1, namespace=["child:1"]), + lifecycle_errored_event(seq=2, error="root failed"), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="existing", assistant_id="agent") as thread: + terminal = thread._wait_for_run_done() + + assert terminal.status == "errored", "a subgraph's completed event ended the run" + assert "root failed" in str(terminal.error) + + def test_sync_threads_stream_accepts_websocket_transport_option(): with httpx.Client(base_url="http://test") as raw: threads = SyncThreadsClient(SyncHttpClient(raw)) From 40a2e6d845054cc0cc17a6a169ca6e7394e5231c Mon Sep 17 00:00:00 2001 From: Elior Nataf Lackritz Date: Wed, 7 Oct 2026 19:17:02 -0400 Subject: [PATCH 2/2] fix(langgraph): never store exit-mode delta writes under the null task id (#9229) In exit mode, a run stores its DeltaChannel writes on the checkpoint it started from, under synthetic task ids that put the superstep first. For a `Command`'s writes in a run whose first superstep is 0, that id came out as `NULL_TASK_ID` itself. Readers take writes under `NULL_TASK_ID` for the checkpoint's own pending writes, so after `invoke(Command(update=...), durability="exit")`: - on a new thread, `get_state` on the empty first checkpoint showed the update in the DeltaChannel but not in a plain channel; - on a thread whose first checkpoint came from `update_state(..., as_node="__input__")`, `get_state` on that checkpoint showed the update the same way, and a replay or fork from it applied the update to the DeltaChannel only. `exit_delta_task_id` now never returns `NULL_TASK_ID`. Threads already saved this way stay as they are. ## Tests - `test_command_update_on_an_input_checkpoint_matches_a_plain_channel` compares every checkpoint in the history, and a replay from the input checkpoint, with a plain channel, on every checkpointer and durability. The exit cases fail without the fix. - `test_exit_command_update_on_a_new_thread_matches_a_plain_channel` compares every checkpoint in a new thread's history with a plain channel, on every checkpointer, and fails without the fix. It runs in exit mode only: in sync and async the two channels already differ on `main`, because of a separate bug that applies a new thread's `Command` update twice. `make format`, `make lint` and the langgraph suite pass. --- .../langgraph/langgraph/pregel/_checkpoint.py | 9 ++++- .../tests/test_delta_channel_exit_mode.py | 39 +++++++++++++++++++ 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/_checkpoint.py b/libs/langgraph/langgraph/pregel/_checkpoint.py index 9f42e6d11..b2cddc19d 100644 --- a/libs/langgraph/langgraph/pregel/_checkpoint.py +++ b/libs/langgraph/langgraph/pregel/_checkpoint.py @@ -26,6 +26,7 @@ from langgraph._internal._constants import ( CONFIG_KEY_CHECKPOINT_ID, NS_END, NS_SEP, + NULL_TASK_ID, PUSH, SNAPSHOT_BUMPS, ) @@ -64,10 +65,14 @@ def exit_delta_task_id(step: int, task_id: str) -> str: Embeds the superstep in the first UUID group so `ORDER BY task_id, idx` preserves chronological order while remaining a valid RFC UUID (required by - Postgres `checkpoint_writes.task_id uuid` columns). + Postgres `checkpoint_writes.task_id uuid` columns). Never `NULL_TASK_ID`: + readers apply writes under it as the anchor checkpoint's own pending writes. """ parts = str(uuid.UUID(task_id)).split("-") - return f"{step:08d}-{parts[1]}-{parts[2]}-{parts[3]}-{parts[4]}" + synthetic = f"{step:08d}-{parts[1]}-{parts[2]}-{parts[3]}-{parts[4]}" + if synthetic == NULL_TASK_ID: + return f"{step:08d}-0000-0000-0000-000000000001" + return synthetic def exit_delta_late_task_id(step: int, task_id: str) -> str: diff --git a/libs/langgraph/tests/test_delta_channel_exit_mode.py b/libs/langgraph/tests/test_delta_channel_exit_mode.py index 50a206fad..27aa2e255 100644 --- a/libs/langgraph/tests/test_delta_channel_exit_mode.py +++ b/libs/langgraph/tests/test_delta_channel_exit_mode.py @@ -17,6 +17,7 @@ from langgraph.checkpoint.memory import InMemorySaver from langgraph.checkpoint.serde.types import _DeltaSnapshot from typing_extensions import TypedDict +from langgraph._internal._constants import NULL_TASK_ID from langgraph.channels.delta import DeltaChannel from langgraph.graph import START, StateGraph from langgraph.graph.message import _messages_delta_reducer @@ -38,6 +39,7 @@ def test_exit_delta_task_id_is_valid_uuid_and_ordered() -> None: assert id1.split("-")[0] == "00000001" assert id7.split("-")[0] == "00000007" assert id1.endswith("-0270-bf16-1ef8-fb321bef9f3d") + assert exit_delta_task_id(0, NULL_TASK_ID) != NULL_TASK_ID with pytest.raises(ValueError): uuid.UUID(f"00000001-{tid}") @@ -463,6 +465,43 @@ def test_resume_with_a_command_update_replays_its_write_once( assert state.values["log"] == state.values["plain"] == ["in", "cmd", "ask", "done"] +def test_command_update_on_an_input_checkpoint_matches_a_plain_channel( + sync_checkpointer: BaseCheckpointSaver, durability: Durability +) -> None: + builder = StateGraph(_ResumeState) + builder.add_node("node", lambda state: _both("node")) + builder.add_edge(START, "node") + graph = builder.compile(checkpointer=sync_checkpointer) + config = {"configurable": {"thread_id": "t"}} + graph.update_state(config, _both("in"), as_node="__input__") + + graph.invoke(Command(update=_both("cmd")), config, durability=durability) + + history = list(graph.get_state_history(config)) + assert [s.values.get("log", []) for s in history] == [ + s.values.get("plain", []) for s in history + ] + replayed = graph.invoke(None, history[-1].config, durability=durability) + assert replayed["log"] == replayed["plain"] + + +def test_exit_command_update_on_a_new_thread_matches_a_plain_channel( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + builder = StateGraph(_ResumeState) + builder.add_node("node", lambda state: _both("node")) + builder.add_edge(START, "node") + graph = builder.compile(checkpointer=sync_checkpointer) + config = {"configurable": {"thread_id": "t"}} + + graph.invoke(Command(update=_both("cmd")), config, durability="exit") + + history = list(graph.get_state_history(config)) + assert [s.values.get("log", []) for s in history] == [ + s.values.get("plain", []) for s in history + ] + + class _FlagState(_ResumeState, total=False): extra: Annotated[list, DeltaChannel(_append)] flag: bool