Skip to content

fix: make Redis checkpoint listing conformant #186

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
CreativeSystemsDevelopment wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix: make Redis checkpoint listing conformant #186

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5494d2e
fix: make redis checkpoint listing conformant
Apr 26, 2026
193ded7
fix: address checkpoint post-filter review
Apr 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 33 additions & 17 deletions langgraph/checkpoint/redis/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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(
Expand All @@ -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 "*"
Expand All @@ -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,

@chatgpt-codex-connector chatgpt-codex-connector Bot Apr 26, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Respect caller limit in post-filter query window

In post-filter mode the search always fetches only the first 10,000 docs, then applies metadata/before filtering in Python. If matching checkpoints are beyond that initial window (or the caller requests limit > 10000), they can never be returned even though they exist, so results become silently incomplete. This affects pagination/filter correctness whenever histories are larger than 10k entries.

Useful? React with 👍 / 👎.

sort_by=("checkpoint_id", "DESC"),
)

Expand Down Expand Up @@ -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": {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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:
Expand Down
50 changes: 33 additions & 17 deletions langgraph/checkpoint/redis/aio.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand All @@ -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 "*"
Expand All @@ -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"),
)

Expand Down Expand Up @@ -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": {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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:
Expand Down
28 changes: 28 additions & 0 deletions langgraph/checkpoint/redis/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

@chatgpt-codex-connector chatgpt-codex-connector Bot Apr 26, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Skip run_id in metadata post-filter checks

When list/alist falls back to post-filtering (for example filter={"run_id": "r1", "score": 42}), _metadata_filter_matches re-validates filter keys against metadata but only skips thread_id. run_id is indexed separately and is often absent from metadata when it comes from config, so valid results are dropped after the Redis query already matched run_id. This makes mixed indexed+custom filters return empty/incomplete results in both sync and async listing paths.

Useful? React with 👍 / 👎.

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]):
Expand Down
Loading