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.
This commit is contained in:
Elior Nataf Lackritz
2026-10-07 19:17:02 -04:00
committed by GitHub
parent a0053bb616
commit 40a2e6d845
2 changed files with 46 additions and 2 deletions
@@ -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:
@@ -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