diff --git a/langgraph/checkpoint/redis/__init__.py b/langgraph/checkpoint/redis/__init__.py index c9e4ec1..760b910 100644 --- a/langgraph/checkpoint/redis/__init__.py +++ b/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 diff --git a/langgraph/checkpoint/redis/aio.py b/langgraph/checkpoint/redis/aio.py index faa0c24..b973bfd 100644 --- a/langgraph/checkpoint/redis/aio.py +++ b/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 diff --git a/langgraph/checkpoint/redis/ashallow.py b/langgraph/checkpoint/redis/ashallow.py index 5023361..4af29b5 100644 --- a/langgraph/checkpoint/redis/ashallow.py +++ b/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:]: diff --git a/langgraph/checkpoint/redis/shallow.py b/langgraph/checkpoint/redis/shallow.py index 51e1aa9..babf0cd 100644 --- a/langgraph/checkpoint/redis/shallow.py +++ b/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 "*" diff --git a/langgraph/checkpoint/redis/util.py b/langgraph/checkpoint/redis/util.py index 9e33895..0e1010b 100644 --- a/langgraph/checkpoint/redis/util.py +++ b/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: """ diff --git a/tests/test_list_before_uuid6.py b/tests/test_list_before_uuid6.py new file mode 100644 index 0000000..2ee8534 --- /dev/null +++ b/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