From 993ba33d76ea6de9d78b4ef8c671824ee5cbbbb9 Mon Sep 17 00:00:00 2001 From: William Fu-Hinthorn <13333726+hinthornw@users.noreply.github.com> Date: Thu, 19 Feb 2026 18:12:26 -0800 Subject: [PATCH] feat: add build-checkpointer skill for building conformant checkpoint savers Adds a Claude Code skill that guides implementation of LangGraph checkpoint savers for any storage backend (SQL, document stores, key-value stores). Includes interface reference, critical contracts, and SQLite patterns. Co-Authored-By: Claude Opus 4.6 --- .claude/skills/build-checkpointer/SKILL.md | 177 ++++++++++++++ .../build-checkpointer/critical-contracts.md | 86 +++++++ .../build-checkpointer/interface-reference.md | 228 ++++++++++++++++++ .../build-checkpointer/sqlite-reference.md | 174 +++++++++++++ 4 files changed, 665 insertions(+) create mode 100644 .claude/skills/build-checkpointer/SKILL.md create mode 100644 .claude/skills/build-checkpointer/critical-contracts.md create mode 100644 .claude/skills/build-checkpointer/interface-reference.md create mode 100644 .claude/skills/build-checkpointer/sqlite-reference.md diff --git a/.claude/skills/build-checkpointer/SKILL.md b/.claude/skills/build-checkpointer/SKILL.md new file mode 100644 index 000000000..857cc8aa8 --- /dev/null +++ b/.claude/skills/build-checkpointer/SKILL.md @@ -0,0 +1,177 @@ +--- +name: build-checkpointer +description: Build a LangGraph checkpoint saver implementation that passes all conformance tests. Use when creating a new checkpointer for any storage backend (Redis, DynamoDB, MongoDB, etc.) or wrapping an existing storage client. +disable-model-invocation: true +user-invocable: true +argument-hint: [storage-backend] +--- + +# Build a Conformant LangGraph Checkpointer + +You are building a LangGraph checkpoint saver for the **$ARGUMENTS** storage backend. Your goal is FULL conformance: all 82 tests across 8 capabilities must pass. + +Read [interface-reference.md](interface-reference.md) for method signatures, data structures, and the conformance test harness template. +Read [critical-contracts.md](critical-contracts.md) for the 8 most common failure points. +Read [sqlite-reference.md](sqlite-reference.md) for patterns from a working implementation. + +## Ground Rules + +**You are not done until ALL conformance tests pass.** Do not stop after writing code — you must run the tests, read failures, fix, and re-run in a loop until you see FULL conformance. If you hit a wall, try a different approach rather than giving up. + +**No hacks or shortcuts.** Specifically: +- Do NOT skip or xfail tests to make the suite "pass" +- Do NOT weaken assertions or modify the conformance test suite itself +- Do NOT use `# type: ignore` to paper over real type mismatches +- Do NOT store data in global/module-level dicts to fake persistence — use the actual storage backend +- Do NOT disable serialization or store raw Python objects — use `self.serde.dumps_typed` / `loads_typed` +- Do NOT catch and swallow exceptions to hide failures + +**Flag security concerns.** As you implement: +- Ensure all queries use parameterized statements — never interpolate user-provided values (thread_id, checkpoint_id, etc.) into SQL or query strings +- Check for injection risks in metadata filtering (JSON path queries, NoSQL operators, etc.) +- Ensure connection credentials are not hardcoded in the implementation — accept them as constructor args +- Flag any backend client library that has known CVEs or security advisories +- If the backend requires TLS/auth, note it prominently in the constructor docstring + +**Ask the user for help when you need it.** Don't guess or assume — ask when: +- You need database connection details, credentials, or access +- You're unsure which client library or driver to use +- You need the user to start/stop a database service +- You're stuck on a test failure after multiple attempts +- You're unsure about a design decision (e.g., schema layout, indexing strategy) + +**Safety checks — ask the user to confirm:** +- "Is this database safe to use for testing? Please confirm it is NOT a production database." (before running any tests that create/delete tables) +- "I'm about to create tables and run destructive test operations (INSERT, DELETE, DROP). Is this OK?" (before first test run) +- "What connection string / credentials should I use?" (never assume defaults for non-local databases) + +## Step 1: Understand the target + +Determine the storage backend from the arguments. If no arguments were provided, ask the user: +- What storage backend? (Redis, DynamoDB, MongoDB, Cassandra, etc.) +- From scratch, or wrapping an existing client/library? +- Any connection/authentication requirements? +- How do I connect to a test instance? (Docker compose, local install, cloud sandbox, etc.) + +Install the backend's Python client library if needed. + +## Step 2: Scaffold the package + +Create `libs/checkpoint-/` with this structure: + +``` +libs/checkpoint-/ + pyproject.toml + Makefile + langgraph/ + checkpoint/ + / + __init__.py # Main implementation + tests/ + test_conformance.py # Conformance harness +``` + +The `pyproject.toml` should depend on: +- `langgraph-checkpoint` (the base interfaces) +- The backend's client library +- `langgraph-checkpoint-conformance` as a test dependency + +Model the `Makefile` after `libs/checkpoint-sqlite/Makefile`. + +## Step 3: Implement the checkpointer + +Subclass `BaseCheckpointSaver` and implement ALL 8 async methods: + +**Required (5):** `aput`, `aget_tuple`, `alist`, `aput_writes`, `adelete_thread` +**Extended (3):** `adelete_for_runs`, `acopy_thread`, `aprune` + +Key implementation guidance: + +1. **Storage layout depends on your backend.** Choose the layout that fits your backend's strengths: + - **SQL databases (Postgres, MySQL, SQLite):** Use 3 tables — checkpoints, checkpoint_blobs (channel values keyed by version), checkpoint_writes. The blobs table avoids re-serializing unchanged large values on every checkpoint write. Inline primitive channel values (str, int, float, bool, None) in the checkpoint JSON; store non-primitives as blobs keyed by `(thread_id, checkpoint_ns, channel, version)`. + - **Document stores (MongoDB, DynamoDB, Firestore):** Use 2 collections — checkpoints (with channel values embedded) and writes. Serialize the full checkpoint including all channel values. The blob optimization adds complexity without much benefit in document stores. + - **Key-value stores (Redis, etcd):** Use composite keys to namespace checkpoints and writes. Store serialized checkpoint + writes as values. + + See `critical-contracts.md` for composite key design. + +3. **Serialize blobs and writes with `self.serde`** — use `self.serde.dumps_typed(value)` which returns `(type_str, bytes)` and `self.serde.loads_typed((type_str, bytes))` for deserialization. For CPU-bound serialization, use `asyncio.to_thread()` to avoid blocking the event loop. + +4. **Serialize metadata as JSON** — use `get_checkpoint_metadata(config, metadata)` to merge config metadata before storing, then `json.dumps()`. Deserialize with `json.loads()`. Metadata is small enough to store inline (no blob table needed). + +5. **Handle `new_versions` correctly** — this is the #1 source of failures. The checkpoint's `channel_values` contains ALL channels, but `new_versions` only lists CHANGED channels. If using a blob table (SQL pattern), only write blobs for channels in `new_versions` and reference all versions in the checkpoint JSON. If storing the full checkpoint (document/KV pattern), just serialize all of `checkpoint["channel_values"]` — simpler and correct. + +6. **Handle `WRITES_IDX_MAP`** — special channels (ERROR, INTERRUPT, SCHEDULED, RESUME) use fixed negative indices. Regular writes use their positional index. Special channel writes should UPSERT (replace on conflict); regular writes should be idempotent (ignore on conflict). + +7. **Return correct `parent_config`** — the `checkpoint_id` in the incoming config to `aput` is the parent. When returning `CheckpointTuple`, set `parent_config` to a config with that parent checkpoint_id, or None if there was no parent. + +### Production-quality patterns + +Go beyond "just passing tests" — build something that performs well at scale: + +- **Connection pooling.** Accept both a single connection and a connection pool in the constructor. Use a pool for production workloads. For Postgres, use `psycopg_pool.AsyncConnectionPool`. For Redis, use the client's built-in pool. Document which to use. +- **Use native backend features.** Don't treat the backend as a dumb key-value store. Examples: + - Postgres: use JSONB containment (`@>`) for metadata filtering, `COPY FROM STDIN` for bulk inserts, `DISTINCT ON` for pruning, pipeline mode for batching + - Redis: use Lua scripts for atomic operations, sorted sets for ordering, hash fields for channel blobs + - DynamoDB: use query vs scan appropriately, batch write items, GSIs for metadata filtering + - MongoDB: use `$match` aggregation stages, bulk write operations, compound indexes +- **Batch writes where possible.** In `aput`, group the checkpoint insert and blob upserts into a single round-trip (pipeline, transaction, or batch write). Don't make N separate calls for N blobs. +- **Fetch writes alongside checkpoints in a single query.** Use subqueries, JOINs, or array aggregation to avoid N+1 patterns where you fetch N checkpoints then query writes for each one separately. +- **Use `asyncio.to_thread()` for CPU-bound serialization** — `serde.dumps_typed` and `serde.loads_typed` can be expensive for large values. Offload to a thread to keep the event loop responsive. +- **Add appropriate indexes.** At minimum: primary/unique keys on all collections, and an index on `thread_id` for `adelete_thread`. For `adelete_for_runs`, consider an index on the metadata `run_id` field if the backend supports it. + +## Step 4: Run conformance and iterate — DO NOT STOP UNTIL GREEN + +```bash +cd libs/checkpoint- +pip install -e ".[test]" +python -m pytest tests/test_conformance.py -x -v +``` + +Or run via `make test` if your Makefile is set up. + +**This is the core of the task.** You MUST loop: + +1. Run the conformance tests +2. Read the failure output carefully — it tells you exactly which contract was violated +3. Understand WHY it failed — read the test source in `libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/` if the error message isn't clear +4. Fix the implementation with a proper solution (not a hack — see Ground Rules) +5. Re-run. Go back to step 1. + +**Do not stop until `report.passed_all()` returns True.** If you've been through 5+ iterations and are still failing, step back and re-read the critical-contracts.md and the failing test source code. The answer is always in the test — it specifies exactly what the contract requires. + +**If you're blocked, ask the user.** Common things to ask about: +- "The database isn't reachable — can you check the connection / start the service?" +- "I'm stuck on this test failure after N attempts — here's what I've tried, can you help?" +- "I need to install this package / run this command — is that OK?" + +Common failure patterns: +- `test_put_incremental_channel_update` fails → you're not storing all channel values, only the ones in `new_versions` +- `test_put_writes_idempotent` fails → your write upsert logic is wrong, check `WRITES_IDX_MAP` handling +- `test_list_global_search` fails → you're requiring a thread_id when config is None +- `test_get_tuple_pending_writes` fails → writes not ordered by `(task_id, idx)` or missing `task_id` in tuple +- `test_list_metadata_filter_*` fails → metadata filtering not checking all keys, or not handling custom keys + +## Step 5: Final verification and review + +Run the full suite one more time with verbose output: + +```bash +python -m pytest tests/test_conformance.py -v +``` + +Confirm the output shows FULL conformance (all 82 tests pass). The report should show: +- PUT: all pass +- PUT_WRITES: all pass +- GET_TUPLE: all pass +- LIST: all pass +- DELETE_THREAD: all pass +- DELETE_FOR_RUNS: all pass +- COPY_THREAD: all pass +- PRUNE: all pass + +Then do a final review of your implementation: + +1. **Run `make lint` and `make format`** to clean up the code +2. **Security review** — check for SQL/query injection, hardcoded credentials, unvalidated inputs +3. **Performance review** — check for N+1 query patterns (fetching writes per checkpoint in a loop), missing indexes on frequently-filtered columns, unnecessary full-table scans +4. **Report findings** — tell the user about any security concerns, performance considerations, or caveats about the implementation diff --git a/.claude/skills/build-checkpointer/critical-contracts.md b/.claude/skills/build-checkpointer/critical-contracts.md new file mode 100644 index 000000000..658e9511a --- /dev/null +++ b/.claude/skills/build-checkpointer/critical-contracts.md @@ -0,0 +1,86 @@ +# Critical Contracts — Common Failure Points + +These are the 8 requirements most likely to cause test failures. Get these right and you'll pass. + +## 1. Store the FULL checkpoint, not just the diff + +`aput` receives `new_versions` which lists only CHANGED channels. But `checkpoint["channel_values"]` contains ALL channels. You must store all of them. The `new_versions` parameter is informational — some implementations use it to optimize blob storage by only writing changed blobs, but the simplest correct approach is to serialize and store the entire checkpoint. + +**Failing test:** `test_put_incremental_channel_update`, `test_put_new_channel_added`, `test_put_channel_removed` + +## 2. Write idempotency with WRITES_IDX_MAP + +The unique key for a write is `(thread_id, checkpoint_ns, checkpoint_id, task_id, idx)`. + +The `idx` comes from `WRITES_IDX_MAP.get(channel, positional_index)`: +- Special channels: `__error__` → -1, `__interrupt__` → -3, `__scheduled__` → -2, `__resume__` → -4 +- Regular channels: use their positional index in the writes list (0, 1, 2, ...) + +For **special channels** (all writes are in WRITES_IDX_MAP): use UPSERT (replace on conflict) because these channels get updated in place. + +For **regular channels**: use INSERT-ignore-on-conflict to be idempotent — calling `aput_writes` twice with the same `(task_id, idx)` must not create duplicates. + +```python +if all(w[0] in WRITES_IDX_MAP for w in writes): + # UPSERT — replace existing +else: + # INSERT OR IGNORE — idempotent +``` + +**Failing test:** `test_put_writes_idempotent`, `test_put_writes_special_channels` + +## 3. Namespace isolation + +`checkpoint_ns` (from `config["configurable"].get("checkpoint_ns", "")`) is part of the composite key for BOTH checkpoints and writes. Default to empty string `""` if not present. + +Two checkpoints with the same `thread_id` and `checkpoint_id` but different `checkpoint_ns` are DIFFERENT checkpoints. + +**Failing test:** `test_put_child_namespace`, `test_put_writes_across_namespaces`, `test_get_tuple_respects_namespace` + +## 4. Metadata round-trip + +Before storing metadata, call `get_checkpoint_metadata(config, metadata)` which merges additional keys from config. Store the result as JSON. When loading, deserialize back to dict. + +ALL keys must survive — standard ones (`source`, `step`, `parents`, `run_id`) AND custom keys the caller added. + +**Failing test:** `test_put_preserves_metadata`, `test_list_metadata_custom_keys` + +## 5. Global search: `alist(None, filter=...)` + +When `config` is `None`, `alist` must search across ALL threads. Don't require `thread_id`. Filter by metadata keys if `filter` is provided. + +**Failing test:** `test_list_global_search` + +## 6. parent_config in CheckpointTuple + +When `aput(config, checkpoint, ...)` is called, `config["configurable"].get("checkpoint_id")` is the PARENT checkpoint ID. Store this as `parent_checkpoint_id`. + +When returning `CheckpointTuple`: +- If `parent_checkpoint_id` exists: set `parent_config = {"configurable": {"thread_id": ..., "checkpoint_ns": ..., "checkpoint_id": parent_checkpoint_id}}` +- If no parent: set `parent_config = None` + +**Failing test:** `test_put_parent_config`, `test_get_tuple_parent_config` + +## 7. Pending writes in CheckpointTuple + +`pending_writes` must be a list of `(task_id, channel, deserialized_value)` tuples, ordered by `(task_id, idx)`. + +Every `aget_tuple` and every tuple yielded by `alist` must include pending writes. Don't forget to query the writes table/collection. + +**Failing test:** `test_get_tuple_pending_writes`, `test_list_includes_pending_writes` + +## 8. Checkpoint ordering in alist + +`alist` must return checkpoints in descending order by `checkpoint_id` (newest first). Checkpoint IDs are UUID-like strings that sort chronologically. Use `ORDER BY checkpoint_id DESC` or equivalent. + +The `before` parameter means: only return checkpoints with `checkpoint_id < before_checkpoint_id`. + +**Failing test:** `test_list_ordering`, `test_list_before`, `test_list_limit_plus_before` + +## 9. Storage design principles + +All backends must key checkpoints by `(thread_id, checkpoint_ns, checkpoint_id)` and writes by `(thread_id, checkpoint_ns, checkpoint_id, task_id, idx)`. + +- **SQL backends:** Consider a 3rd blobs table keyed by `(thread_id, checkpoint_ns, channel, version)` to avoid re-serializing unchanged large channel values. Only write blobs for channels in `new_versions`; reconstruct all values on read via `channel_versions`. +- **Document/KV backends:** Embed all channel values directly in the checkpoint document/value. Serialize the full checkpoint on every `aput` — simpler and correct. +- **All backends need:** descending `checkpoint_id` ordering for `alist`, metadata field filtering for `alist(filter=...)`, delete by `thread_id` for `adelete_thread`, delete by `metadata.run_id` for `adelete_for_runs`. diff --git a/.claude/skills/build-checkpointer/interface-reference.md b/.claude/skills/build-checkpointer/interface-reference.md new file mode 100644 index 000000000..4b30f72c8 --- /dev/null +++ b/.claude/skills/build-checkpointer/interface-reference.md @@ -0,0 +1,228 @@ +# Checkpointer Interface Reference + +## Imports + +```python +from langgraph.checkpoint.base import ( + WRITES_IDX_MAP, + BaseCheckpointSaver, + ChannelVersions, + Checkpoint, + CheckpointMetadata, + CheckpointTuple, + SerializerProtocol, + get_checkpoint_id, + get_checkpoint_metadata, +) +from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer +from langchain_core.runnables import RunnableConfig +``` + +## Data Structures + +```python +# RunnableConfig["configurable"] keys: +# thread_id: str — identifies the conversation thread +# checkpoint_ns: str — namespace (empty string for root, dotted path for subgraphs) +# checkpoint_id: str — unique monotonically-increasing ID (UUID-like, sortable) + +# Checkpoint (TypedDict): +# v: int — format version (currently 1) +# id: str — unique checkpoint ID +# ts: str — ISO 8601 timestamp +# channel_values: dict[str, Any] — serialized state per channel +# channel_versions: ChannelVersions — version number per channel +# versions_seen: dict[str, ChannelVersions] — per-node version tracking + +# CheckpointMetadata (TypedDict): +# source: str — "input" | "loop" | "update" | "fork" +# step: int — -1 for input, 0+ for loop steps +# parents: dict[str, str] — parent checkpoint IDs +# (plus any custom keys the caller adds) + +# CheckpointTuple (NamedTuple): +# config: RunnableConfig +# checkpoint: Checkpoint +# metadata: CheckpointMetadata +# parent_config: RunnableConfig | None +# pending_writes: list[tuple[str, str, Any]] | None +# Each write is (task_id, channel, value) + +# ChannelVersions = dict[str, Any] (typically str or int version numbers) + +# WRITES_IDX_MAP = {"__error__": -1, "__scheduled__": -2, "__interrupt__": -3, "__resume__": -4} +``` + +## Method Signatures + +### Required Methods + +```python +async def aput( + self, + config: RunnableConfig, + checkpoint: Checkpoint, + metadata: CheckpointMetadata, + new_versions: ChannelVersions, +) -> RunnableConfig: + """Store a checkpoint. Return config with checkpoint_id set to checkpoint["id"]. + + The incoming config["configurable"]["checkpoint_id"] is the PARENT checkpoint ID. + new_versions contains only the channels that changed — but checkpoint["channel_values"] + has ALL channels. Store the full checkpoint. + """ + +async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None: + """Retrieve a checkpoint. + + If config has checkpoint_id: return that exact checkpoint. + If no checkpoint_id: return the LATEST checkpoint for the thread+namespace. + Return None if not found. + Include pending_writes as list of (task_id, channel, value) ordered by (task_id, idx). + """ + +async def alist( + self, + config: RunnableConfig | None, + *, + filter: dict[str, Any] | None = None, + before: RunnableConfig | None = None, + limit: int | None = None, +) -> AsyncIterator[CheckpointTuple]: + """List checkpoints, newest first (descending checkpoint_id). + + If config is None: search ALL threads (global search). + If config has thread_id: filter to that thread. + filter: dict of metadata key-value pairs (AND logic). + before: only return checkpoints before this checkpoint_id. + limit: max number to return. + Each yielded tuple must include pending_writes. + """ + +async def aput_writes( + self, + config: RunnableConfig, + writes: Sequence[tuple[str, Any]], + task_id: str, + task_path: str = "", +) -> None: + """Store pending writes for a checkpoint. + + Each write is (channel, value). Use WRITES_IDX_MAP.get(channel, idx) for the index. + Special channels (in WRITES_IDX_MAP) should UPSERT (replace on conflict). + Regular channels should be idempotent (ignore on conflict). + """ + +async def adelete_thread(self, thread_id: str) -> None: + """Delete ALL checkpoints and writes for a thread (all namespaces).""" +``` + +### Extended Methods + +```python +async def adelete_for_runs(self, run_ids: Sequence[str]) -> None: + """Delete checkpoints+writes where metadata.run_id is in run_ids.""" + +async def acopy_thread(self, source_thread_id: str, target_thread_id: str) -> None: + """Copy all checkpoints+writes from source thread to target thread.""" + +async def aprune( + self, + thread_ids: Sequence[str], + *, + strategy: str = "keep_latest", +) -> None: + """Prune checkpoints for given threads. + strategy="keep_latest": keep only the latest checkpoint per thread+namespace. + strategy="delete_all": delete everything for those threads. + """ +``` + +## Conformance Test Harness Template + +Create `tests/test_conformance.py`: + +```python +"""Conformance tests for Saver.""" +from __future__ import annotations + +import pytest +from langgraph.checkpoint.conformance import checkpointer_test, validate +from langgraph.checkpoint.conformance.report import ProgressCallbacks + +# Import your checkpointer +from langgraph.checkpoint. import Saver + + +# Optional: lifespan for one-time setup/teardown (database creation, etc.) +# async def backend_lifespan(): +# # setup +# yield +# # teardown + + +@checkpointer_test(name="Saver") # add lifespan=backend_lifespan if needed +async def backend_checkpointer(): + # Create and yield a fresh checkpointer instance. + # Use async with if your saver needs connection management. + saver = Saver(...) + yield saver + # cleanup (close connections, etc.) + + +@pytest.mark.asyncio +async def test_full_conformance(): + """Saver passes ALL conformance tests.""" + report = await validate( + backend_checkpointer, + progress=ProgressCallbacks.verbose(), + ) + report.print_report() + assert report.passed_all(), f"Conformance failed: {report.to_dict()}" +``` + +## Serialization Pattern + +```python +# In __init__: +super().__init__(serde=serde) + +# Storing metadata (use JSON, not serde): +merged = get_checkpoint_metadata(config, metadata) +serialized_md = json.dumps(merged).encode("utf-8") + +# Loading metadata: +metadata = json.loads(serialized_md_bytes) + +# Storing/loading blob values and write values (use serde): +type_, serialized = self.serde.dumps_typed(value) +value = self.serde.loads_typed((type_, serialized_bytes)) + +# For CPU-bound serde in async context, offload to thread: +type_, serialized = await asyncio.to_thread(self.serde.dumps_typed, value) +value = await asyncio.to_thread(self.serde.loads_typed, (type_, serialized_bytes)) +``` + +## Schema Design by Backend Type + +All backends must store checkpoints keyed by `(thread_id, checkpoint_ns, checkpoint_id)` and writes keyed by `(thread_id, checkpoint_ns, checkpoint_id, task_id, idx)`. + +### SQL backends (Postgres, MySQL, SQLite) + +Use 3 tables: **checkpoints** (checkpoint JSON with primitive channel_values inlined + channel_versions for blob lookup, metadata JSON), **checkpoint_blobs** (non-primitive channel values keyed by `(thread_id, checkpoint_ns, channel, version)`), and **checkpoint_writes** (pending writes). The blobs table avoids re-serializing unchanged large values — only write blobs for channels in `new_versions`. On read, JOIN blobs via `channel_versions` to reconstruct all channel values. PKs on all three tables handle most access patterns; add an index on the metadata `run_id` field for `adelete_for_runs`. Use subqueries/JOINs to fetch writes alongside checkpoints in a single round-trip. + +### Document stores (MongoDB, Firestore, DynamoDB) + +Use 2 collections: **checkpoints** (full checkpoint with all channel_values embedded, metadata as top-level fields) and **writes**. Serialize the full checkpoint including all channel values on every `aput`. Use composite `_id` or PK/SK from the key parts. Required indexes: +- `(thread_id, checkpoint_ns, checkpoint_id DESC)` — for `alist` ordering and `aget_tuple` latest-lookup +- `(thread_id)` — for `adelete_thread` +- `(metadata.run_id)` — for `adelete_for_runs` +- Use native query operators (e.g. MongoDB `$match`, DynamoDB filter expressions) for metadata filtering in `alist(filter=...)` + +### Key-value stores (Redis, etcd) + +Use composite keys like `cp:{thread_id}:{ns}:{id}`. Use sorted sets or equivalent for descending-order listing. **Requires manual secondary indexes** maintained on every write: +- Thread index (`thread:{thread_id}` → set of `{ns}:{checkpoint_id}`) — for `adelete_thread` and `alist` +- Run ID index (`run:{run_id}` → set of checkpoint keys) — for `adelete_for_runs` +- Write index (`writes:{thread_id}:{ns}:{checkpoint_id}` → set of `{task_id}:{idx}`) — for pending writes lookup +- Metadata filtering for `alist(filter=...)` is the hardest: either scan+deserialize, or maintain per-field indexes. For small datasets scanning is acceptable; for large ones consider a search module. diff --git a/.claude/skills/build-checkpointer/sqlite-reference.md b/.claude/skills/build-checkpointer/sqlite-reference.md new file mode 100644 index 000000000..4614a9b7f --- /dev/null +++ b/.claude/skills/build-checkpointer/sqlite-reference.md @@ -0,0 +1,174 @@ +# SQLite Implementation Reference + +Working patterns from `libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py`. + +## Schema + +```sql +CREATE TABLE IF NOT EXISTS checkpoints ( + thread_id TEXT NOT NULL, + checkpoint_ns TEXT NOT NULL DEFAULT '', + checkpoint_id TEXT NOT NULL, + parent_checkpoint_id TEXT, + type TEXT, + checkpoint BLOB, + metadata BLOB, + PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id) +); + +CREATE TABLE IF NOT EXISTS writes ( + thread_id TEXT NOT NULL, + checkpoint_ns TEXT NOT NULL DEFAULT '', + checkpoint_id TEXT NOT NULL, + task_id TEXT NOT NULL, + idx INTEGER NOT NULL, + channel TEXT NOT NULL, + type TEXT, + value BLOB, + PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) +); +``` + +## aput pattern + +```python +async def aput(self, config, checkpoint, metadata, new_versions): + thread_id = config["configurable"]["thread_id"] + checkpoint_ns = config["configurable"].get("checkpoint_ns", "") + parent_checkpoint_id = config["configurable"].get("checkpoint_id") + + type_, serialized_checkpoint = self.serde.dumps_typed(checkpoint) + serialized_metadata = json.dumps( + get_checkpoint_metadata(config, metadata), ensure_ascii=False + ).encode("utf-8", "ignore") + + # UPSERT checkpoint row + await db.execute( + "INSERT OR REPLACE INTO checkpoints (...) VALUES (...)", + (thread_id, checkpoint_ns, checkpoint["id"], parent_checkpoint_id, + type_, serialized_checkpoint, serialized_metadata), + ) + + return { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": checkpoint_ns, + "checkpoint_id": checkpoint["id"], + } + } +``` + +## aget_tuple pattern + +```python +async def aget_tuple(self, config): + checkpoint_ns = config["configurable"].get("checkpoint_ns", "") + + if checkpoint_id := get_checkpoint_id(config): + # Fetch specific checkpoint + query = "... WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?" + else: + # Fetch latest + query = "... WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1" + + row = await fetch_one(query, ...) + if not row: + return None + + # Fetch pending writes for this checkpoint + writes = await fetch_all( + "SELECT task_id, channel, type, value FROM writes " + "WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? " + "ORDER BY task_id, idx", + ... + ) + + return CheckpointTuple( + config={"configurable": {"thread_id": ..., "checkpoint_ns": ..., "checkpoint_id": ...}}, + checkpoint=self.serde.loads_typed((type_, blob)), + metadata=json.loads(metadata_blob), + parent_config=( + {"configurable": {"thread_id": ..., "checkpoint_ns": ..., "checkpoint_id": parent_id}} + if parent_id else None + ), + pending_writes=[ + (task_id, channel, self.serde.loads_typed((type_, value))) + for task_id, channel, type_, value in writes + ], + ) +``` + +## alist pattern + +```python +async def alist(self, config, *, filter=None, before=None, limit=None): + # Build WHERE clause dynamically + where_clauses = [] + params = [] + + if config is not None: + where_clauses.append("thread_id = ?") + params.append(config["configurable"]["thread_id"]) + if checkpoint_ns := config["configurable"].get("checkpoint_ns"): + where_clauses.append("checkpoint_ns = ?") + params.append(checkpoint_ns) + + if filter: + # Filter on metadata JSON — for each key-value pair: + for key, value in filter.items(): + where_clauses.append(f"json_extract(metadata, '$.{key}') = ?") + params.append(json.dumps(value) if not isinstance(value, (str, int, float)) else value) + + if before: + before_id = before["configurable"]["checkpoint_id"] + where_clauses.append("checkpoint_id < ?") + params.append(before_id) + + where = "WHERE " + " AND ".join(where_clauses) if where_clauses else "" + query = f"SELECT ... FROM checkpoints {where} ORDER BY checkpoint_id DESC" + if limit: + query += " LIMIT ?" + params.append(limit) + + # For each checkpoint row, also fetch its writes (same as aget_tuple) +``` + +## aput_writes pattern + +```python +async def aput_writes(self, config, writes, task_id, task_path=""): + thread_id = config["configurable"]["thread_id"] + checkpoint_ns = config["configurable"].get("checkpoint_ns", "") + checkpoint_id = config["configurable"]["checkpoint_id"] + + # Choose UPSERT vs INSERT-ignore based on channel types + if all(w[0] in WRITES_IDX_MAP for w in writes): + query = "INSERT OR REPLACE INTO writes (...) VALUES (...)" + else: + query = "INSERT OR IGNORE INTO writes (...) VALUES (...)" + + rows = [ + (thread_id, checkpoint_ns, checkpoint_id, task_id, + WRITES_IDX_MAP.get(channel, idx), channel, + *self.serde.dumps_typed(value)) + for idx, (channel, value) in enumerate(writes) + ] + await executemany(query, rows) +``` + +## adelete_thread pattern + +```python +async def adelete_thread(self, thread_id): + await execute("DELETE FROM checkpoints WHERE thread_id = ?", (thread_id,)) + await execute("DELETE FROM writes WHERE thread_id = ?", (thread_id,)) +``` + +## Key takeaway + +The SQLite implementation is ~300 lines and is the simplest correct reference. It uses 2 tables and serializes the full checkpoint as a single blob. + +- SQLite patterns show the simplest correct implementation of every contract +- SQL backends can add a 3rd blobs table for performance (see `interface-reference.md` Schema Design section) +- NoSQL backends should adapt the contracts to native idioms — focus on `critical-contracts.md` +- Don't port SQL patterns to NoSQL; use your backend's native features (document embedding, sorted sets, composite keys, etc.)