diff --git a/libs/langgraph/langgraph/pregel/_checkpoint.py b/libs/langgraph/langgraph/pregel/_checkpoint.py index 081bc69bf..cb28563c9 100644 --- a/libs/langgraph/langgraph/pregel/_checkpoint.py +++ b/libs/langgraph/langgraph/pregel/_checkpoint.py @@ -1,5 +1,6 @@ from __future__ import annotations +import uuid from collections.abc import Callable, Mapping from datetime import datetime, timezone from typing import Any, cast @@ -34,6 +35,17 @@ def empty_checkpoint() -> Checkpoint: ) +def exit_delta_task_id(step: int, task_id: str) -> str: + """Synthetic task id for exit-mode DeltaChannel writes. + + 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). + """ + parts = str(uuid.UUID(task_id)).split("-") + return f"{step:08d}-{parts[1]}-{parts[2]}-{parts[3]}-{parts[4]}" + + def delta_channels_to_snapshot( channels: Mapping[str, BaseChannel], counters_since_delta_snapshot: Mapping[str, tuple[int, int]], diff --git a/libs/langgraph/langgraph/pregel/_loop.py b/libs/langgraph/langgraph/pregel/_loop.py index 587f5905a..69f5c97e1 100644 --- a/libs/langgraph/langgraph/pregel/_loop.py +++ b/libs/langgraph/langgraph/pregel/_loop.py @@ -102,6 +102,7 @@ from langgraph.pregel._checkpoint import ( create_checkpoint, delta_channels_to_snapshot, empty_checkpoint, + exit_delta_task_id, ) from langgraph.pregel._executor import ( AsyncBackgroundExecutor, @@ -1272,7 +1273,7 @@ class PregelLoop: }, ) for (step, tid), entries in grouped.items(): - synth_tid = f"{step:08d}-{tid}" + synth_tid = exit_delta_task_id(step, tid) if self.checkpointer_put_writes_accepts_task_path: fut = self.submit( self.checkpointer_put_writes, diff --git a/libs/langgraph/tests/test_delta_channel_exit_mode.py b/libs/langgraph/tests/test_delta_channel_exit_mode.py index ae53dbbf7..3d18ce708 100644 --- a/libs/langgraph/tests/test_delta_channel_exit_mode.py +++ b/libs/langgraph/tests/test_delta_channel_exit_mode.py @@ -6,6 +6,7 @@ channel), lazy stub creation when no parent exists, and proper read-path reconstruction via ancestor walks. """ +import uuid from typing import Annotated, Any import pytest @@ -17,10 +18,28 @@ from typing_extensions import TypedDict from langgraph.channels.delta import DeltaChannel from langgraph.graph import START, StateGraph from langgraph.graph.message import _messages_delta_reducer +from langgraph.pregel._checkpoint import exit_delta_task_id pytestmark = pytest.mark.anyio +def test_exit_delta_task_id_is_valid_uuid_and_ordered() -> None: + """Exit-mode synthetic task ids must parse as UUID and sort by superstep.""" + tid = "4f7226e4-0270-bf16-1ef8-fb321bef9f3d" + id1 = exit_delta_task_id(1, tid) + id7 = exit_delta_task_id(7, tid) + + uuid.UUID(id1) + uuid.UUID(id7) + assert id1 < id7 + assert id1.split("-")[0] == "00000001" + assert id7.split("-")[0] == "00000007" + assert id1.endswith("-0270-bf16-1ef8-fb321bef9f3d") + + with pytest.raises(ValueError): + uuid.UUID(f"00000001-{tid}") + + def _build_graph( checkpointer: InMemorySaver, *,