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 <noreply@anthropic.com>
2026-08-17 21:25:46 +02:00 · 2026-02-19 18:12:26 -08:00
4 changed files with 665 additions and 0 deletions
@@ -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-<backend>/` with this structure:
+
+```
+libs/checkpoint-<backend>/
+  pyproject.toml
+  Makefile
+  langgraph/
+    checkpoint/
+      <backend>/
+        __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-<backend>
+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
@@ -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`.
@@ -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 <Backend>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.<backend> import <Backend>Saver
+
+
+# Optional: lifespan for one-time setup/teardown (database creation, etc.)
+# async def backend_lifespan():
+#     # setup
+#     yield
+#     # teardown
+
+
+@checkpointer_test(name="<Backend>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 = <Backend>Saver(...)
+    yield saver
+    # cleanup (close connections, etc.)
+
+
+@pytest.mark.asyncio
+async def test_full_conformance():
+    """<Backend>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.
@@ -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.)