fix(checkpoint): apply list(before=...) to LangGraph's UUIDv6 checkpoint ids

langgraph/checkpoint/redis/__init__.py

@@ -24,7 +24,6 @@
 from redisvl.query import FilterQuery
 from redisvl.query.filter import Num, Tag
 from redisvl.redis.connection import RedisConnectionFactory
-from ulid import ULID
 
 from langgraph.checkpoint.redis.aio import AsyncRedisSaver
 from langgraph.checkpoint.redis.ashallow import AsyncShallowRedisSaver
@@ -44,6 +43,7 @@
 from langgraph.checkpoint.redis.shallow import ShallowRedisSaver
 from langgraph.checkpoint.redis.util import (
     EMPTY_ID_SENTINEL,
+    checkpoint_id_timestamp,
     from_storage_safe_id,
     from_storage_safe_str,
     to_storage_safe_id,
@@ -276,14 +276,11 @@ def list(
         if before:
             before_checkpoint_id = get_checkpoint_id(before)
             if before_checkpoint_id:
-                try:
-                    before_ulid = ULID.from_str(before_checkpoint_id)
-                    before_ts = before_ulid.timestamp
+                before_ts = checkpoint_id_timestamp(before_checkpoint_id)
+                # An id that encodes no time (e.g. a uuid4) cannot be placed, so no filter.
+                if before_ts is not None:
                     # Use numeric range query: checkpoint_ts < before_ts
                     filter_expression.append(Num("checkpoint_ts") < before_ts)
-                except Exception:
-                    # If not a valid ULID, ignore the before filter
-                    pass
 
         # Combine all filter expressions
         combined_filter = filter_expression[0] if filter_expression else "*"
@@ -524,16 +521,12 @@ def put(
             }
         }
 
-        # Extract timestamp from checkpoint_id (ULID)
+        # Extract timestamp from checkpoint_id (ULID or LangGraph's UUIDv6)
         checkpoint_ts = None
         if checkpoint_id:
-            try:
-                from ulid import ULID
-
-                ulid_obj = ULID.from_str(checkpoint_id)
-                checkpoint_ts = ulid_obj.timestamp  # milliseconds since epoch
-            except Exception:
-                # If not a valid ULID, use current time
+            checkpoint_ts = checkpoint_id_timestamp(checkpoint_id)
+            if checkpoint_ts is None:
+                # The id encodes no time, use current time
                 import time
 
                 checkpoint_ts = time.time() * 1000

langgraph/checkpoint/redis/aio.py

@@ -39,7 +39,6 @@
 from redisvl.query import FilterQuery
 from redisvl.query.filter import Num, Tag
 from redisvl.redis.connection import RedisConnectionFactory
-from ulid import ULID
 
 from langgraph.checkpoint.redis.base import (
     CHECKPOINT_PREFIX,
@@ -54,6 +53,7 @@
 )
 from langgraph.checkpoint.redis.util import (
     EMPTY_ID_SENTINEL,
+    checkpoint_id_timestamp,
     from_storage_safe_id,
     from_storage_safe_str,
     to_storage_safe_id,
@@ -673,14 +673,11 @@ async def alist(
         if before:
             before_checkpoint_id = get_checkpoint_id(before)
             if before_checkpoint_id:
-                try:
-                    before_ulid = ULID.from_str(before_checkpoint_id)
-                    before_ts = before_ulid.timestamp
+                before_ts = checkpoint_id_timestamp(before_checkpoint_id)
+                # An id that encodes no time (e.g. a uuid4) cannot be placed, so no filter.
+                if before_ts is not None:
                     # Use numeric range query: checkpoint_ts < before_ts
                     filter_expression.append(Num("checkpoint_ts") < before_ts)
-                except Exception:
-                    # If not a valid ULID, ignore the before filter
-                    pass
 
         # Combine all filter expressions
         combined_filter = filter_expression[0] if filter_expression else "*"
@@ -947,16 +944,12 @@ async def aput(
         # Store checkpoint data with cluster-aware handling
         try:
             # Store checkpoint data WITH inline channel values
-            # Extract timestamp from checkpoint_id (ULID)
+            # Extract timestamp from checkpoint_id (ULID or LangGraph's UUIDv6)
             checkpoint_ts = None
             if checkpoint_id:
-                try:
-                    from ulid import ULID
-
-                    ulid_obj = ULID.from_str(checkpoint_id)
-                    checkpoint_ts = ulid_obj.timestamp  # milliseconds since epoch
-                except Exception:
-                    # If not a valid ULID, use current time
+                checkpoint_ts = checkpoint_id_timestamp(checkpoint_id)
+                if checkpoint_ts is None:
+                    # The id encodes no time, use current time
                     import time
 
                     checkpoint_ts = time.time() * 1000

langgraph/checkpoint/redis/ashallow.py

@@ -26,7 +26,6 @@
 from redisvl.query import FilterQuery
 from redisvl.query.filter import Num, Tag
 from redisvl.redis.connection import RedisConnectionFactory
-from ulid import ULID
 
 from langgraph.checkpoint.redis.base import (
     CHECKPOINT_PREFIX,
@@ -36,6 +35,7 @@
     aexpire_with_retry,
 )
 from langgraph.checkpoint.redis.util import (
+    checkpoint_id_timestamp,
     to_storage_safe_id,
     to_storage_safe_str,
 )
@@ -200,17 +200,12 @@ async def aput(
         }
 
         try:
-            # Extract timestamp from checkpoint_id (ULID) or fallback to checkpoint's ts field
-            # Note: LangGraph may generate checkpoint IDs in different formats (ULID, UUIDv6, etc.)
-            # We try ULID first, then fall back gracefully without warnings (Issue #136)
+            # Extract timestamp from checkpoint_id (ULID or LangGraph's UUIDv6), or fall back
+            # to the checkpoint's ts field for ids that encode no time (Issue #136)
             checkpoint_ts = None
             if checkpoint["id"]:
-                try:
-                    ulid_obj = ULID.from_str(checkpoint["id"])
-                    checkpoint_ts = ulid_obj.timestamp  # milliseconds since epoch
-                except Exception:
-                    # Not a valid ULID - this is expected for UUIDv6 and other formats
-                    # Fall back to checkpoint's timestamp field or current time
+                checkpoint_ts = checkpoint_id_timestamp(checkpoint["id"])
+                if checkpoint_ts is None:
                     checkpoint_ts = self._extract_fallback_timestamp(checkpoint)
 
             # Store channel values inline in the checkpoint
@@ -311,14 +306,11 @@ async def alist(
         if before:
             before_checkpoint_id = get_checkpoint_id(before)
             if before_checkpoint_id:
-                try:
-                    before_ulid = ULID.from_str(before_checkpoint_id)
-                    before_ts = before_ulid.timestamp
+                before_ts = checkpoint_id_timestamp(before_checkpoint_id)
+                # An id that encodes no time (e.g. a uuid4) cannot be placed, so no filter.
+                if before_ts is not None:
                     # Use numeric range query: checkpoint_ts < before_ts
                     query_filter.append(Num("checkpoint_ts") < before_ts)
-                except Exception:
-                    # If not a valid ULID, ignore the before filter
-                    pass
 
         combined_filter = query_filter[0] if query_filter else "*"
         for expr in query_filter[1:]:

langgraph/checkpoint/redis/shallow.py

@@ -23,7 +23,6 @@
 from redisvl.query import FilterQuery
 from redisvl.query.filter import Num, Tag
 from redisvl.redis.connection import RedisConnectionFactory
-from ulid import ULID
 
 from langgraph.checkpoint.redis.base import (
     CHECKPOINT_PREFIX,
@@ -33,6 +32,7 @@
     expire_with_retry,
 )
 from langgraph.checkpoint.redis.util import (
+    checkpoint_id_timestamp,
     from_storage_safe_str,
     to_storage_safe_id,
     to_storage_safe_str,
@@ -143,17 +143,12 @@ def put(
             }
         }
 
-        # Extract timestamp from checkpoint_id (ULID) or fallback to checkpoint's ts field
-        # Note: LangGraph may generate checkpoint IDs in different formats (ULID, UUIDv6, etc.)
-        # We try ULID first, then fall back gracefully without warnings (Issue #136)
+        # Extract timestamp from checkpoint_id (ULID or LangGraph's UUIDv6), or fall back
+        # to the checkpoint's ts field for ids that encode no time (Issue #136)
         checkpoint_ts = None
         if checkpoint["id"]:
-            try:
-                ulid_obj = ULID.from_str(checkpoint["id"])
-                checkpoint_ts = ulid_obj.timestamp  # milliseconds since epoch
-            except Exception:
-                # Not a valid ULID - this is expected for UUIDv6 and other formats
-                # Fall back to checkpoint's timestamp field or current time
+            checkpoint_ts = checkpoint_id_timestamp(checkpoint["id"])
+            if checkpoint_ts is None:
                 checkpoint_ts = self._extract_fallback_timestamp(checkpoint)
 
         # Parse metadata from string to dict to avoid double serialization
@@ -251,14 +246,11 @@ def list(
         if before:
             before_checkpoint_id = get_checkpoint_id(before)
             if before_checkpoint_id:
-                try:
-                    before_ulid = ULID.from_str(before_checkpoint_id)
-                    before_ts = before_ulid.timestamp
+                before_ts = checkpoint_id_timestamp(before_checkpoint_id)
+                # An id that encodes no time (e.g. a uuid4) cannot be placed, so no filter.
+                if before_ts is not None:
                     # Use numeric range query: checkpoint_ts < before_ts
                     filter_expression.append(Num("checkpoint_ts") < before_ts)
-                except Exception:
-                    # If not a valid ULID, ignore the before filter
-                    pass
 
         # Combine all filter expressions
         combined_filter = filter_expression[0] if filter_expression else "*"

langgraph/checkpoint/redis/util.py

@@ -11,11 +11,45 @@
 configured with decode_responses.
 """
 
-from typing import Any
+import uuid
+from typing import Any, Optional
+
+from ulid import ULID
 
 EMPTY_STRING_SENTINEL = "__empty__"
 EMPTY_ID_SENTINEL = "00000000-0000-0000-0000-000000000000"
 
+# 100-nanosecond intervals between the UUID epoch (1582-10-15) and the Unix epoch.
+_UUID_EPOCH_OFFSET = 0x01B21DD213814000
+
+
+def checkpoint_id_timestamp(checkpoint_id: str) -> Optional[float]:
+    """Return the timestamp a checkpoint id encodes, or None if it encodes none.
+
+    This is the value stored as ``checkpoint_ts`` and compared by ``list(before=...)``,
+    so both sides must come from here.
+
+    A ULID returns ``ULID.timestamp``, the value the savers have always stored for one.
+    LangGraph's own checkpoint ids are UUIDv6, whose 60-bit timestamp is returned in
+    milliseconds since the Unix epoch, the unit already stored for non-ULID ids. Any
+    other id (for example a uuid4) carries no time and returns None.
+    """
+    if not isinstance(checkpoint_id, str):
+        return None
+    try:
+        return ULID.from_str(checkpoint_id).timestamp
+    except ValueError:
+        pass
+    try:
+        parsed = uuid.UUID(checkpoint_id)
+    except ValueError:
+        return None
+    if parsed.version != 6:
+        return None
+    n = parsed.int
+    ticks = ((n >> 96) << 28) | (((n >> 80) & 0xFFFF) << 12) | ((n >> 64) & 0x0FFF)
+    return (ticks - _UUID_EPOCH_OFFSET) / 10_000
+
 
 def to_storage_safe_str(value: str) -> str:
     """

tests/test_list_before_uuid6.py

@@ -0,0 +1,107 @@
+"""list(before=...) must filter for the checkpoint ids LangGraph actually generates.
+
+LangGraph creates checkpoint ids with uuid6() (36 characters). The savers built the
+``before`` filter by parsing the id as a ULID, and when that failed they dropped the
+filter, so ``get_state_history(config, before=...)`` returned the whole history.
+"""
+
+import operator
+import uuid
+from typing import Annotated, TypedDict
+
+import pytest
+from langgraph.graph import END, START, StateGraph
+
+from langgraph.checkpoint.redis import RedisSaver
+from langgraph.checkpoint.redis.aio import AsyncRedisSaver
+
+
+class State(TypedDict):
+    messages: Annotated[list, operator.add]
+
+
+def _reply(state: State) -> dict:
+    return {"messages": [f"reply to {state['messages'][-1]}"]}
+
+
+def _graph(saver):
+    builder = StateGraph(State)
+    builder.add_node("reply", _reply)
+    builder.add_edge(START, "reply")
+    builder.add_edge("reply", END)
+    return builder.compile(checkpointer=saver)
+
+
+def _config() -> dict:
+    return {"configurable": {"thread_id": f"before-{uuid.uuid4()}"}}
+
+
+def test_get_state_history_before_uuid6_checkpoint(redis_url: str) -> None:
+    with RedisSaver.from_conn_string(redis_url) as saver:
+        saver.setup()
+        app = _graph(saver)
+        config = _config()
+        for turn in range(3):
+            app.invoke({"messages": [f"turn {turn}"]}, config)
+
+        history = list(app.get_state_history(config))
+        ids = [s.config["configurable"]["checkpoint_id"] for s in history]
+        assert len(ids[0]) == 36  # LangGraph's own uuid6 ids, not ULIDs
+        middle = history[len(history) // 2]
+
+        before = list(app.get_state_history(config, before=middle.config))
+        before_ids = [s.config["configurable"]["checkpoint_id"] for s in before]
+
+        expected = [
+            i for i in ids if i < middle.config["configurable"]["checkpoint_id"]
+        ]
+        assert before_ids == expected
+        assert middle.config["configurable"]["checkpoint_id"] not in before_ids
+
+
+@pytest.mark.asyncio
+async def test_aget_state_history_before_uuid6_checkpoint(redis_url: str) -> None:
+    async with AsyncRedisSaver.from_conn_string(redis_url) as saver:
+        await saver.asetup()
+        app = _graph(saver)
+        config = _config()
+        for turn in range(3):
+            await app.ainvoke({"messages": [f"turn {turn}"]}, config)
+
+        history = [s async for s in app.aget_state_history(config)]
+        ids = [s.config["configurable"]["checkpoint_id"] for s in history]
+        middle = history[len(history) // 2]
+
+        before = [s async for s in app.aget_state_history(config, before=middle.config)]
+        before_ids = [s.config["configurable"]["checkpoint_id"] for s in before]
+
+        expected = [
+            i for i in ids if i < middle.config["configurable"]["checkpoint_id"]
+        ]
+        assert before_ids == expected
+
+
+def test_checkpoint_id_timestamp_formats() -> None:
+    import time
+
+    from langgraph.checkpoint.base import empty_checkpoint
+    from ulid import ULID
+
+    from langgraph.checkpoint.redis.util import checkpoint_id_timestamp
+
+    # LangGraph's uuid6 ids decode to milliseconds since the epoch, in order.
+    first, second = empty_checkpoint()["id"], empty_checkpoint()["id"]
+    first_ts, second_ts = checkpoint_id_timestamp(first), checkpoint_id_timestamp(
+        second
+    )
+    assert first_ts is not None and second_ts is not None
+    assert abs(first_ts - time.time() * 1000) < 60_000
+    assert first_ts < second_ts
+
+    # ULIDs keep the value the savers have always stored for them.
+    ulid = ULID()
+    assert checkpoint_id_timestamp(str(ulid)) == ulid.timestamp
+
+    # Ids that encode no time give None, and the caller falls back.
+    assert checkpoint_id_timestamp(str(uuid.uuid4())) is None
+    assert checkpoint_id_timestamp("not-an-id") is None