From 5494d2e7a1ed48aa25c30baf6fe468683768db1f Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 25 Apr 2026 22:23:11 -0400 Subject: [PATCH 1/2] fix: make redis checkpoint listing conformant --- langgraph/checkpoint/redis/__init__.py | 50 ++++--- langgraph/checkpoint/redis/aio.py | 50 ++++--- langgraph/checkpoint/redis/base.py | 28 ++++ tests/test_checkpoint_filters.py | 175 +++++++++++++++++++++++++ 4 files changed, 269 insertions(+), 34 deletions(-) diff --git a/langgraph/checkpoint/redis/__init__.py b/langgraph/checkpoint/redis/__init__.py index 36d1a58..260fd33 100644 --- a/langgraph/checkpoint/redis/__init__.py +++ b/langgraph/checkpoint/redis/__init__.py @@ -31,8 +31,11 @@ from langgraph.checkpoint.redis.base import ( CHECKPOINT_PREFIX, CHECKPOINT_WRITE_PREFIX, + INDEXED_CHECKPOINT_FILTER_KEYS, REDIS_KEY_SEPARATOR, BaseRedisSaver, + _checkpoint_id_filter_matches, + _metadata_filter_matches, ) from langgraph.checkpoint.redis.key_registry import SyncCheckpointKeyRegistry from langgraph.checkpoint.redis.message_exporter import ( @@ -250,9 +253,12 @@ def list( # Search for checkpoints with any namespace, including an empty # string, while `checkpoint_id` has to have a value. - if checkpoint_ns := config["configurable"].get("checkpoint_ns"): + if "checkpoint_ns" in config["configurable"]: filter_expression.append( - Tag("checkpoint_ns") == to_storage_safe_str(checkpoint_ns) + Tag("checkpoint_ns") + == to_storage_safe_str( + config["configurable"].get("checkpoint_ns", "") + ) ) if checkpoint_id := get_checkpoint_id(config): filter_expression.append( @@ -269,20 +275,20 @@ def list( filter_expression.append(Tag("thread_id") == to_storage_safe_id(v)) elif k == "run_id": filter_expression.append(Tag("run_id") == to_storage_safe_id(v)) - else: - raise ValueError(f"Unsupported filter key: {k}") - 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 - # 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 + requires_post_filter = bool( + filter and any(k not in INDEXED_CHECKPOINT_FILTER_KEYS for k in filter) + ) + before_checkpoint_id = get_checkpoint_id(before) if before else None + + if before_checkpoint_id and not requires_post_filter: + try: + before_ulid = ULID.from_str(before_checkpoint_id) + before_ts = before_ulid.timestamp + # Use numeric range query: checkpoint_ts < before_ts + filter_expression.append(Num("checkpoint_ts") < before_ts) + except Exception: + requires_post_filter = True # Combine all filter expressions combined_filter = filter_expression[0] if filter_expression else "*" @@ -302,7 +308,7 @@ def list( "$.metadata", "has_writes", # Include has_writes to optimize pending_writes loading ], - num_results=limit or 10000, + num_results=10000 if requires_post_filter else limit or 10000, sort_by=("checkpoint_id", "DESC"), ) @@ -430,6 +436,12 @@ def list( else: metadata = cast(CheckpointMetadata, metadata_dict) + if requires_post_filter and ( + not _checkpoint_id_filter_matches(checkpoint_id, before_checkpoint_id) + or not _metadata_filter_matches(metadata, filter) + ): + continue + # Pre-create the config structure more efficiently config_param: RunnableConfig = { "configurable": { @@ -474,6 +486,10 @@ def list( parent_config=parent_config, pending_writes=pending_writes, ) + if limit is not None: + limit -= 1 + if limit <= 0: + break def put( self, @@ -1692,7 +1708,7 @@ def prune( # Validate input if not thread_ids: - raise ValueError("``thread_ids`` must be a non-empty sequence") + return if keep_last < 0: raise ValueError(f"``keep_last`` must be >= 0, got {keep_last}") if max_results < 1: diff --git a/langgraph/checkpoint/redis/aio.py b/langgraph/checkpoint/redis/aio.py index bcd6689..d4af98b 100644 --- a/langgraph/checkpoint/redis/aio.py +++ b/langgraph/checkpoint/redis/aio.py @@ -44,8 +44,11 @@ from langgraph.checkpoint.redis.base import ( CHECKPOINT_PREFIX, CHECKPOINT_WRITE_PREFIX, + INDEXED_CHECKPOINT_FILTER_KEYS, REDIS_KEY_SEPARATOR, BaseRedisSaver, + _checkpoint_id_filter_matches, + _metadata_filter_matches, ) from langgraph.checkpoint.redis.key_registry import ( AsyncCheckpointKeyRegistry as AsyncKeyRegistry, @@ -638,9 +641,12 @@ async def alist( # Search for checkpoints with any namespace, including an empty # string, while `checkpoint_id` has to have a value. - if checkpoint_ns := config["configurable"].get("checkpoint_ns"): + if "checkpoint_ns" in config["configurable"]: filter_expression.append( - Tag("checkpoint_ns") == to_storage_safe_str(checkpoint_ns) + Tag("checkpoint_ns") + == to_storage_safe_str( + config["configurable"].get("checkpoint_ns", "") + ) ) if checkpoint_id := get_checkpoint_id(config): filter_expression.append( @@ -657,20 +663,20 @@ async def alist( filter_expression.append(Tag("thread_id") == to_storage_safe_id(v)) elif k == "run_id": filter_expression.append(Tag("run_id") == to_storage_safe_id(v)) - else: - raise ValueError(f"Unsupported filter key: {k}") - 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 - # 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 + requires_post_filter = bool( + filter and any(k not in INDEXED_CHECKPOINT_FILTER_KEYS for k in filter) + ) + before_checkpoint_id = get_checkpoint_id(before) if before else None + + if before_checkpoint_id and not requires_post_filter: + try: + before_ulid = ULID.from_str(before_checkpoint_id) + before_ts = before_ulid.timestamp + # Use numeric range query: checkpoint_ts < before_ts + filter_expression.append(Num("checkpoint_ts") < before_ts) + except Exception: + requires_post_filter = True # Combine all filter expressions combined_filter = filter_expression[0] if filter_expression else "*" @@ -690,7 +696,7 @@ async def alist( "$.metadata", "has_writes", # Include has_writes to optimize pending_writes loading ], - num_results=limit or 10000, + num_results=10000 if requires_post_filter else limit or 10000, sort_by=("checkpoint_id", "DESC"), ) @@ -822,6 +828,12 @@ async def alist( else: metadata = cast(CheckpointMetadata, metadata_dict) + if requires_post_filter and ( + not _checkpoint_id_filter_matches(checkpoint_id, before_checkpoint_id) + or not _metadata_filter_matches(metadata, filter) + ): + continue + # Pre-create the config structure more efficiently config_param: RunnableConfig = { "configurable": { @@ -866,6 +878,10 @@ async def alist( parent_config=parent_config, pending_writes=pending_writes, ) + if limit is not None: + limit -= 1 + if limit <= 0: + break async def aput( self, @@ -2086,7 +2102,7 @@ async def aprune( # Validate inputs if not thread_ids: - raise ValueError("``thread_ids`` must be a non-empty sequence") + return if keep_last < 0: raise ValueError(f"``keep_last`` must be >= 0, got {keep_last}") if max_results < 1: diff --git a/langgraph/checkpoint/redis/base.py b/langgraph/checkpoint/redis/base.py index 049dfcb..f5eca89 100644 --- a/langgraph/checkpoint/redis/base.py +++ b/langgraph/checkpoint/redis/base.py @@ -33,6 +33,34 @@ REDIS_KEY_SEPARATOR = ":" CHECKPOINT_PREFIX = "checkpoint" CHECKPOINT_WRITE_PREFIX = "checkpoint_write" +INDEXED_CHECKPOINT_FILTER_KEYS = {"source", "step", "thread_id", "run_id"} + + +def _metadata_filter_matches( + metadata: dict[str, Any], filters: Optional[dict[str, Any]] +) -> bool: + """Return whether checkpoint metadata matches user-supplied filters.""" + if not filters: + return True + for key, expected in filters.items(): + if key == "thread_id": + continue + actual = metadata.get(key) + if isinstance(expected, list): + if actual not in expected: + return False + elif actual != expected: + return False + return True + + +def _checkpoint_id_filter_matches( + checkpoint_id: str, before_checkpoint_id: Optional[str] +) -> bool: + """Apply list(before=...) using the same descending checkpoint_id order.""" + if not before_checkpoint_id: + return True + return checkpoint_id < before_checkpoint_id class BaseRedisSaver(BaseCheckpointSaver[str], Generic[RedisClientType, IndexType]): diff --git a/tests/test_checkpoint_filters.py b/tests/test_checkpoint_filters.py index d1fb45f..22a2847 100644 --- a/tests/test_checkpoint_filters.py +++ b/tests/test_checkpoint_filters.py @@ -126,6 +126,181 @@ async def test_alist_filters_run_id_and_thread_id(redis_url: str) -> None: assert len(thread_id_results) == 2 +def test_list_filters_root_namespace(redis_url: str) -> None: + with RedisSaver.from_conn_string(redis_url) as checkpointer: + checkpointer.setup() + + thread_id = "thread-root-namespace-sync" + root_config: RunnableConfig = { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": "", + "checkpoint_id": "1", + } + } + child_config: RunnableConfig = { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": "child:1", + "checkpoint_id": "2", + } + } + + checkpointer.put( + root_config, + _make_checkpoint("1"), + CheckpointMetadata(source="input", step=0, writes={}), + {"messages": "1"}, + ) + checkpointer.put( + child_config, + _make_checkpoint("2"), + CheckpointMetadata(source="input", step=1, writes={}), + {"messages": "1"}, + ) + + root_results = list( + checkpointer.list( + {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} + ) + ) + assert len(root_results) == 1 + assert root_results[0].checkpoint["id"] == "1" + + +def test_list_filters_custom_metadata_and_before(redis_url: str) -> None: + with RedisSaver.from_conn_string(redis_url) as checkpointer: + checkpointer.setup() + + thread_id = "thread-custom-filter-sync" + for checkpoint_id, score in [("1", 42), ("2", 99), ("3", 99)]: + checkpointer.put( + { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": "", + "checkpoint_id": checkpoint_id, + } + }, + _make_checkpoint(checkpoint_id), + CheckpointMetadata( + source="input", + step=int(checkpoint_id), + writes={}, + score=score, + ), + {"messages": "1"}, + ) + + filtered = list( + checkpointer.list( + {"configurable": {"thread_id": thread_id}}, + filter={"score": 42}, + ) + ) + assert len(filtered) == 1 + assert filtered[0].checkpoint["id"] == "1" + + before = {"configurable": {"thread_id": thread_id, "checkpoint_id": "3"}} + paged = list( + checkpointer.list( + {"configurable": {"thread_id": thread_id}}, + before=before, + limit=1, + ) + ) + assert len(paged) == 1 + assert paged[0].checkpoint["id"] == "2" + + +@pytest.mark.asyncio +async def test_alist_filters_root_namespace(redis_url: str) -> None: + async with AsyncRedisSaver.from_conn_string(redis_url) as checkpointer: + thread_id = "thread-root-namespace-async" + root_config: RunnableConfig = { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": "", + "checkpoint_id": "1", + } + } + child_config: RunnableConfig = { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": "child:1", + "checkpoint_id": "2", + } + } + + await checkpointer.aput( + root_config, + _make_checkpoint("1"), + CheckpointMetadata(source="input", step=0, writes={}), + {"messages": "1"}, + ) + await checkpointer.aput( + child_config, + _make_checkpoint("2"), + CheckpointMetadata(source="input", step=1, writes={}), + {"messages": "1"}, + ) + + root_results = [ + item + async for item in checkpointer.alist( + {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} + ) + ] + assert len(root_results) == 1 + assert root_results[0].checkpoint["id"] == "1" + + +@pytest.mark.asyncio +async def test_alist_filters_custom_metadata_and_before(redis_url: str) -> None: + async with AsyncRedisSaver.from_conn_string(redis_url) as checkpointer: + thread_id = "thread-custom-filter-async" + for checkpoint_id, score in [("1", 42), ("2", 99), ("3", 99)]: + await checkpointer.aput( + { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": "", + "checkpoint_id": checkpoint_id, + } + }, + _make_checkpoint(checkpoint_id), + CheckpointMetadata( + source="input", + step=int(checkpoint_id), + writes={}, + score=score, + ), + {"messages": "1"}, + ) + + filtered = [ + item + async for item in checkpointer.alist( + {"configurable": {"thread_id": thread_id}}, + filter={"score": 42}, + ) + ] + assert len(filtered) == 1 + assert filtered[0].checkpoint["id"] == "1" + + before = {"configurable": {"thread_id": thread_id, "checkpoint_id": "3"}} + paged = [ + item + async for item in checkpointer.alist( + {"configurable": {"thread_id": thread_id}}, + before=before, + limit=1, + ) + ] + assert len(paged) == 1 + assert paged[0].checkpoint["id"] == "2" + + # Now test using the higher-level configuration async def test_aget_state_history_run_id_and_thread_id_pregel(redis_url: str) -> None: class State(TypedDict): From 193ded7b6b15a8ba59caa7c25732fbfad3d2e137 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 25 Apr 2026 22:32:12 -0400 Subject: [PATCH 2/2] fix: address checkpoint post-filter review --- langgraph/checkpoint/redis/__init__.py | 52 +++++++++++++++++--------- langgraph/checkpoint/redis/aio.py | 52 +++++++++++++++++--------- langgraph/checkpoint/redis/base.py | 2 +- tests/test_checkpoint_filters.py | 34 +++++++++-------- 4 files changed, 89 insertions(+), 51 deletions(-) diff --git a/langgraph/checkpoint/redis/__init__.py b/langgraph/checkpoint/redis/__init__.py index 260fd33..faa31b9 100644 --- a/langgraph/checkpoint/redis/__init__.py +++ b/langgraph/checkpoint/redis/__init__.py @@ -297,30 +297,48 @@ def list( # Construct the Redis query # Sort by checkpoint_id in descending order to get most recent checkpoints first - query = FilterQuery( - filter_expression=combined_filter, - return_fields=[ - "thread_id", - "checkpoint_ns", - "checkpoint_id", - "parent_checkpoint_id", - "$.checkpoint", - "$.metadata", - "has_writes", # Include has_writes to optimize pending_writes loading - ], - num_results=10000 if requires_post_filter else limit or 10000, - sort_by=("checkpoint_id", "DESC"), - ) + return_fields = [ + "thread_id", + "checkpoint_ns", + "checkpoint_id", + "parent_checkpoint_id", + "$.checkpoint", + "$.metadata", + "has_writes", # Include has_writes to optimize pending_writes loading + ] - # Execute the query - results = self.checkpoints_index.search(query) + if requires_post_filter: + result_docs = [] + offset = 0 + page_size = 10000 + while True: + query = FilterQuery( + filter_expression=combined_filter, + return_fields=return_fields, + num_results=page_size, + sort_by=("checkpoint_id", "DESC"), + ) + query.paging(offset, page_size) + results = self.checkpoints_index.search(query) + result_docs.extend(results.docs) + if len(results.docs) < page_size: + break + offset += page_size + else: + query = FilterQuery( + filter_expression=combined_filter, + return_fields=return_fields, + num_results=limit or 10000, + sort_by=("checkpoint_id", "DESC"), + ) + result_docs = self.checkpoints_index.search(query).docs # Pre-process all docs to collect batch query requirements all_docs_data = [] pending_sends_batch_keys = [] pending_writes_batch_keys = [] - for doc in results.docs: + for doc in result_docs: # Extract all attributes once doc_dict = doc.__dict__ if hasattr(doc, "__dict__") else {} diff --git a/langgraph/checkpoint/redis/aio.py b/langgraph/checkpoint/redis/aio.py index d4af98b..7f570e2 100644 --- a/langgraph/checkpoint/redis/aio.py +++ b/langgraph/checkpoint/redis/aio.py @@ -685,30 +685,48 @@ async def alist( # Construct the Redis query # Sort by checkpoint_id in descending order to get most recent checkpoints first - query = FilterQuery( - filter_expression=combined_filter, - return_fields=[ - "thread_id", - "checkpoint_ns", - "checkpoint_id", - "parent_checkpoint_id", - "$.checkpoint", - "$.metadata", - "has_writes", # Include has_writes to optimize pending_writes loading - ], - num_results=10000 if requires_post_filter else limit or 10000, - sort_by=("checkpoint_id", "DESC"), - ) + return_fields = [ + "thread_id", + "checkpoint_ns", + "checkpoint_id", + "parent_checkpoint_id", + "$.checkpoint", + "$.metadata", + "has_writes", # Include has_writes to optimize pending_writes loading + ] - # Execute the query asynchronously - results = await self.checkpoints_index.search(query) + if requires_post_filter: + result_docs = [] + offset = 0 + page_size = 10000 + while True: + query = FilterQuery( + filter_expression=combined_filter, + return_fields=return_fields, + num_results=page_size, + sort_by=("checkpoint_id", "DESC"), + ) + query.paging(offset, page_size) + results = await self.checkpoints_index.search(query) + result_docs.extend(results.docs) + if len(results.docs) < page_size: + break + offset += page_size + else: + query = FilterQuery( + filter_expression=combined_filter, + return_fields=return_fields, + num_results=limit or 10000, + sort_by=("checkpoint_id", "DESC"), + ) + result_docs = (await self.checkpoints_index.search(query)).docs # Pre-process all docs to collect batch query requirements all_docs_data = [] pending_sends_batch_keys = [] pending_writes_batch_keys = [] - for doc in results.docs: + for doc in result_docs: # Extract all attributes once doc_dict = doc.__dict__ if hasattr(doc, "__dict__") else {} diff --git a/langgraph/checkpoint/redis/base.py b/langgraph/checkpoint/redis/base.py index f5eca89..528634f 100644 --- a/langgraph/checkpoint/redis/base.py +++ b/langgraph/checkpoint/redis/base.py @@ -43,7 +43,7 @@ def _metadata_filter_matches( if not filters: return True for key, expected in filters.items(): - if key == "thread_id": + if key in {"thread_id", "run_id"}: continue actual = metadata.get(key) if isinstance(expected, list): diff --git a/tests/test_checkpoint_filters.py b/tests/test_checkpoint_filters.py index 22a2847..d6c5365 100644 --- a/tests/test_checkpoint_filters.py +++ b/tests/test_checkpoint_filters.py @@ -174,14 +174,15 @@ def test_list_filters_custom_metadata_and_before(redis_url: str) -> None: thread_id = "thread-custom-filter-sync" for checkpoint_id, score in [("1", 42), ("2", 99), ("3", 99)]: + configurable = { + "thread_id": thread_id, + "checkpoint_ns": "", + "checkpoint_id": checkpoint_id, + } + if checkpoint_id == "1": + configurable["run_id"] = "run-1" checkpointer.put( - { - "configurable": { - "thread_id": thread_id, - "checkpoint_ns": "", - "checkpoint_id": checkpoint_id, - } - }, + {"configurable": configurable}, _make_checkpoint(checkpoint_id), CheckpointMetadata( source="input", @@ -195,7 +196,7 @@ def test_list_filters_custom_metadata_and_before(redis_url: str) -> None: filtered = list( checkpointer.list( {"configurable": {"thread_id": thread_id}}, - filter={"score": 42}, + filter={"run_id": "run-1", "score": 42}, ) ) assert len(filtered) == 1 @@ -260,14 +261,15 @@ async def test_alist_filters_custom_metadata_and_before(redis_url: str) -> None: async with AsyncRedisSaver.from_conn_string(redis_url) as checkpointer: thread_id = "thread-custom-filter-async" for checkpoint_id, score in [("1", 42), ("2", 99), ("3", 99)]: + configurable = { + "thread_id": thread_id, + "checkpoint_ns": "", + "checkpoint_id": checkpoint_id, + } + if checkpoint_id == "1": + configurable["run_id"] = "run-1" await checkpointer.aput( - { - "configurable": { - "thread_id": thread_id, - "checkpoint_ns": "", - "checkpoint_id": checkpoint_id, - } - }, + {"configurable": configurable}, _make_checkpoint(checkpoint_id), CheckpointMetadata( source="input", @@ -282,7 +284,7 @@ async def test_alist_filters_custom_metadata_and_before(redis_url: str) -> None: item async for item in checkpointer.alist( {"configurable": {"thread_id": thread_id}}, - filter={"score": 42}, + filter={"run_id": "run-1", "score": 42}, ) ] assert len(filtered) == 1