Persisting LangGraph state with checkpointers

LangGraph checkpointer state persistence is the mechanism that lets your multi-agent workflows survive restarts, scale horizontally, and resume from arbitrary points. Without it, every graph execution starts from scratch — fine for demos, fatal for production. This guide walks through configuring, customizing, and verifying checkpointers end to end.

Step 1: Understand what a checkpointer actually does

A checkpointer implements the BaseCheckpointSaver interface. It serializes the graph’s State object at each step — or at explicit interrupt points — and writes it to a backend. On resume, it deserializes the latest checkpoint and restores the graph’s position, including the message history, tool outputs, and any custom fields you added to the state schema.

Key methods you’ll interact with:

• put(config, checkpoint, metadata, new_versions) — write a checkpoint
• get(config) — read the latest checkpoint for a thread
• list(config) — iterate checkpoints for a thread (for time-travel debugging)
• put_writes(config, writes, task_id) — store intermediate writes from a node

The config parameter carries thread_id (required) and optionally checkpoint_ns (namespace) and checkpoint_id (specific version). This design lets you branch conversations, implement undo, or run A/B comparisons on the same thread.

Step 2: Set up a minimal graph to experiment with

Create a new project and install the core packages:

mkdir langgraph-checkpoint-demo && cd langgraph-checkpoint-demo
python -m venv .venv && source .venv/bin/activate
pip install langgraph langchain-openai langchain-core

Define a simple state schema and graph. Save this as graph.py:

# graph.py
from typing import Annotated, List
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

class State(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    user_name: str
    turn_count: int

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def chat_node(state: State) -> State:
    response = llm.invoke(state["messages"])
    return {
        "messages": [response],
        "turn_count": state.get("turn_count", 0) + 1,
    }

def greet_node(state: State) -> State:
    name = state.get("user_name", "there")
    greeting = AIMessage(content=f"Hello, {name}! How can I help?")
    return {"messages": [greeting]}

builder = StateGraph(State)
builder.add_node("greet", greet_node)
builder.add_node("chat", chat_node)
builder.add_edge(START, "greet")
builder.add_edge("greet", "chat")
builder.add_edge("chat", END)

graph = builder.compile()

Run a quick smoke test:

# test_smoke.py
from graph import graph

result = graph.invoke(
    {"messages": [HumanMessage(content="Hi")], "user_name": "Ada"},
    config={"configurable": {"thread_id": "test-1"}}
)
print(result["messages"][-1].content)

Execute python test_smoke.py. You should see a greeting. The graph works. Now add persistence.

Step 3: Add an in-memory checkpointer for development

LangGraph ships with MemorySaver — an in-memory checkpointer useful for local iteration and tests. It requires zero infrastructure.

Update graph.py to accept a checkpointer at compile time:

# graph.py (updated compile section)
from langgraph.checkpoint.memory import MemorySaver

# ... existing imports and graph definition ...

def build_graph(checkpointer=None):
    builder = StateGraph(State)
    builder.add_node("greet", greet_node)
    builder.add_node("chat", chat_node)
    builder.add_edge(START, "greet")
    builder.add_edge("greet", "chat")
    builder.add_edge("chat", END)
    return builder.compile(checkpointer=checkpointer)

# Default for scripts
graph = build_graph(MemorySaver())

Create a script that demonstrates checkpointing across invocations:

# test_memory_checkpoint.py
from graph import build_graph
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage

checkpointer = MemorySaver()
graph = build_graph(checkpointer)

config = {"configurable": {"thread_id": "demo-thread"}}

# Turn 1
result = graph.invoke(
    {"messages": [HumanMessage(content="What's 2+2?")], "user_name": "Ada", "turn_count": 0},
    config=config
)
print(f"Turn 1: {result['messages'][-1].content}")
print(f"Turn count: {result['turn_count']}")

# Turn 2 — same thread_id, state persists
result = graph.invoke(
    {"messages": [HumanMessage(content="And 3+3?")]},
    config=config
)
print(f"Turn 2: {result['messages'][-1].content}")
print(f"Turn count: {result['turn_count']}")

# Inspect checkpoints
print("\n--- Checkpoint history ---")
for checkpoint in checkpointer.list(config):
    print(f"  checkpoint_id: {checkpoint.config['configurable']['checkpoint_id']}")
    print(f"  metadata: {checkpoint.metadata}")
    print(f"  values keys: {list(checkpoint.checkpoint['channel_values'].keys())}")

Run it. You’ll see the turn count increment across invocations and the checkpoint history printed. The MemorySaver stores everything in a Python dict — fine for a single process, lost on restart.

Step 4: Switch to SQLite for local persistence

For a developer machine or single-container deployment, SqliteSaver gives you durability without a database server. It writes to a .sqlite file.

pip install langgraph-checkpoint-sqlite

Update your test script:

# test_sqlite_checkpoint.py
from graph import build_graph
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import HumanMessage
import sqlite3

# Use a file-backed database
conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
checkpointer = SqliteSaver(conn)

graph = build_graph(checkpointer)

config = {"configurable": {"thread_id": "sqlite-demo"}}

# First run
result = graph.invoke(
    {"messages": [HumanMessage(content="Remember: my favorite color is blue")], "user_name": "Ada", "turn_count": 0},
    config=config
)
print(f"Run 1: {result['messages'][-1].content}")

# Simulate process restart — new connection, same file
conn2 = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
checkpointer2 = SqliteSaver(conn2)
graph2 = build_graph(checkpointer2)

# Resume — state should be there
result = graph2.invoke(
    {"messages": [HumanMessage(content="What's my favorite color?")]},
    config=config
)
print(f"Run 2 (after restart): {result['messages'][-1].content}")

Run it twice — once to create the checkpoint, once to verify resume. The second run prints the color from the first run. Open the SQLite file to inspect:

sqlite3 checkpoints.sqlite ".schema"
sqlite3 checkpoints.sqlite "SELECT thread_id, checkpoint_id, type, metadata FROM checkpoints;"

You’ll see tables for checkpoints, checkpoint_writes, and checkpoint_blobs. The blobs table stores large serialized values (message lists) separately to keep the main table lean.

Step 5: Use Postgres for production workloads

Production systems need connection pooling, concurrent access, and operational tooling. PostgresSaver uses psycopg with connection pooling and advisory locks to coordinate writers.

pip install langgraph-checkpoint-postgres psycopg[pool]

Start a Postgres instance (Docker is easiest):

docker run -d --name pg-checkpoint \
  -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=langgraph \
  -p 5432:5432 \
  postgres:16

Wait for it to be ready, then run the migration to create tables:

# migrate_postgres.py
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    "postgresql://postgres:postgres@localhost:5432/langgraph",
    max_size=10,
    kwargs={"autocommit": True}
)

checkpointer = PostgresSaver(pool)
checkpointer.setup()  # Creates tables if they don't exist
print("Migration complete")

Run python migrate_postgres.py. Now use it in your graph:

# test_postgres_checkpoint.py
from graph import build_graph
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool
from langchain_core.messages import HumanMessage

pool = ConnectionPool(
    "postgresql://postgres:postgres@localhost:5432/langgraph",
    max_size=5,
    kwargs={"autocommit": True}
)

checkpointer = PostgresSaver(pool)
graph = build_graph(checkpointer)

config = {"configurable": {"thread_id": "prod-thread-1"}}

result = graph.invoke(
    {"messages": [HumanMessage(content="Store this in Postgres")], "user_name": "Ada", "turn_count": 0},
    config=config
)
print(f"Stored: {result['messages'][-1].content}")

# Verify from a second pool (simulates another worker)
pool2 = ConnectionPool(
    "postgresql://postgres:postgres@localhost:5432/langgraph",
    max_size=5,
    kwargs={"autocommit": True}
)
checkpointer2 = PostgresSaver(pool2)
graph2 = build_graph(checkpointer2)

result = graph2.invoke(
    {"messages": [HumanMessage(content="What did I just say?")]},
    config=config
)
print(f"Retrieved: {result['messages'][-1].content}")

Run it. The second graph instance, using a separate connection pool, reads the checkpoint written by the first. This is horizontal scaling in action.

Step 6: Implement a custom checkpointer for unusual backends

If you need to write to S3, Redis, DynamoDB, or a custom service, subclass BaseCheckpointSaver. The interface is small but the serialization details matter.

Here’s a skeleton for a Redis checkpointer using redis-py and msgpack for compact serialization:

# redis_checkpointer.py
import msgpack
import redis
from typing import Any, Optional, Iterator, List
from langgraph.checkpoint.base import BaseCheckpointSaver, Checkpoint, CheckpointMetadata, CheckpointTuple
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

class RedisSaver(BaseCheckpointSaver):
    def __init__(self, client: redis.Redis, ttl: int = 86400 * 30):
        super().__init__(serde=JsonPlusSerializer())
        self.client = client
        self.ttl = ttl
        self._lock_lua = client.register_script("""
            if redis.call('setnx', KEYS[1], ARGV[1]) == 1 then
                redis.call('expire', KEYS[1], ARGV[2])
                return 1
            else
                return 0
            end
        """)

    def _key(self, thread_id: str, checkpoint_ns: str = "", checkpoint_id: str = "") -> str:
        parts = ["checkpoint", thread_id, checkpoint_ns]
        if checkpoint_id:
            parts.append(checkpoint_id)
        return ":".join(parts)

    def _lock_key(self, thread_id: str) -> str:
        return f"lock:checkpoint:{thread_id}"

    def put(
        self,
        config: dict,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: dict
    ) -> dict:
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        checkpoint_id = checkpoint["id"]

        # Serialize checkpoint tuple
        tuple_data = (checkpoint, metadata, new_versions)
        serialized = msgpack.packb(tuple_data, use_bin_type=True)

        # Advisory lock via Redis SETNX
        lock_acquired = False
        for _ in range(10):
            lock_acquired = self._lock_lua(
                keys=[self._lock_key(thread_id)],
                args=[checkpoint_id, 10]
            )
            if lock_acquired:
                break
        if not lock_acquired:
            raise RuntimeError("Could not acquire checkpoint lock")

        try:
            pipe = self.client.pipeline()
            pipe.set(self._key(thread_id, checkpoint_ns, checkpoint_id), serialized, ex=self.ttl)
            # Update latest pointer
            pipe.set(self._key(thread_id, checkpoint_ns, "latest"), checkpoint_id, ex=self.ttl)
            pipe.execute()
        finally:
            self.client.delete(self._lock_key(thread_id))

        return {"configurable": {"thread_id": thread_id, "checkpoint_ns": checkpoint_ns, "checkpoint_id": checkpoint_id}}

    def get_tuple(self, config: dict) -> Optional[CheckpointTuple]:
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        checkpoint_id = config["configurable"].get("checkpoint_id")

        key = self._key(thread_id, checkpoint_ns, checkpoint_id or "latest")
        data = self.client.get(key)
        if not data:
            return None

        if checkpoint_id is None:
            # Resolve latest pointer
            checkpoint_id = data.decode()
            data = self.client.get(self._key(thread_id, checkpoint_ns, checkpoint_id))
            if not data:
                return None

        checkpoint, metadata, new_versions = msgpack.unpackb(data, raw=False)
        return CheckpointTuple(
            config={"configurable": {"thread_id": thread_id, "checkpoint_ns": checkpoint_ns, "checkpoint_id": checkpoint_id}},
            checkpoint=checkpoint,
            metadata=metadata,
            parent_config=None,  # Simplified; implement if you need branching
            pending_writes=[]
        )

    def list(self, config: dict, *, filter: Optional[dict] = None, before: Optional[dict] = None, limit: Optional[int] = None) -> Iterator[CheckpointTuple]:
        # Simplified: scan keys matching thread_id prefix
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        pattern = self._key(thread_id, checkpoint_ns, "*")
        for key in self.client.scan_iter(match=pattern, count=100):
            key_str = key.decode()
            if key_str.endswith(":latest"):
                continue
            data = self.client.get(key)
            if data:
                checkpoint, metadata, _ = msgpack.unpackb(data, raw=False)
                cid = key_str.split(":")[-1]
                yield CheckpointTuple(
                    config={"configurable": {"thread_id": thread_id, "checkpoint_ns": checkpoint_ns, "checkpoint_id": cid}},
                    checkpoint=checkpoint,
                    metadata=metadata,
                    parent_config=None,
                    pending_writes=[]
                )

    def put_writes(self, config: dict, writes: List[tuple], task_id: str) -> None:
        # Store writes separately if needed for your use case
        pass

Usage:

# test_redis_checkpointer.py
import redis
from graph import build_graph
from redis_checkpointer import RedisSaver
from langchain_core.messages import HumanMessage

client = redis.Redis(decode_responses=False)
checkpointer = RedisSaver(client)
graph = build_graph(checkpointer)

config = {"configurable": {"thread_id": "redis-thread"}}

result = graph.invoke(
    {"messages": [HumanMessage(content="Redis checkpoint test")], "user_name": "Ada", "turn_count": 0},
    config=config
)
print(f"Result: {result['messages'][-1].content}")

# Verify keys exist
for key in client.scan_iter("checkpoint:redis-thread:*"):
    print(f"  {key.decode()}")

This implementation uses a Lua script for atomic lock acquisition, msgpack for smaller payloads, and a “latest” pointer for fast reads. Adapt the TTL, serialization, and locking strategy to your infrastructure.

Step 7: Handle state migration when your schema evolves

Checkpoints are serialized state. When you add, remove, or rename fields in your State TypedDict, existing checkpoints become incompatible. Plan for this from day one.

Add a version field to your state and a migration function:

# graph.py (updated State and migration)
class State(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    user_name: str
    turn_count: int
    schema_version: int  # NEW

CURRENT_SCHEMA_VERSION = 2

def migrate_state(state: dict) -> dict:
    """Upgrade a checkpointed state to the current schema version."""
    version = state.get("schema_version", 1)
    
    if version == 1:
        # v1 -> v2: added user_name, default to "unknown"
        state["user_name"] = state.get("user_name", "unknown")
        state["schema_version"] = 2
    
    # Future migrations go here as elif blocks
    
    return state

def chat_node(state: State) -> State:
    # Ensure migration runs on resume
    state = migrate_state(state)
    response = llm.invoke(state["messages"])
    return {
        "messages": [response],
        "turn_count": state.get("turn_count", 0) + 1,
        "schema_version": CURRENT_SCHEMA_VERSION,
    }

def greet_node(state: State) -> State:
    state = migrate_state(state)
    name = state.get("user_name", "there")
    greeting = AIMessage(content=f"Hello, {name}! How can I help?")
    return {"messages": [greeting], "schema_version": CURRENT_SCHEMA_VERSION}

When a graph resumes from a v1 checkpoint, migrate_state runs before your node logic, upgrading the state in memory. The next checkpoint written will have schema_version: 2.

For complex migrations (renaming fields, restructuring nested objects), write a standalone migration script that reads all checkpoints, transforms them, and writes them back. Run it as a one-off job during deployment.

Step 8: Verify correctness with a test harness

Don’t rely on manual testing. Build a verification suite that exercises the full checkpoint lifecycle.

# test_checkpoint_verification.py
import pytest
from graph import build_graph, State
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool
from langchain_core.messages import HumanMessage, AIMessage
import sqlite3
import os

@pytest.fixture(params=["memory", "sqlite", "postgres"])
def checkpointer(request):
    if request.param == "memory":
        yield MemorySaver()
    elif request.param == "sqlite":
        conn = sqlite3.connect(":memory:", check_same_thread=False)
        yield SqliteSaver(conn)
        conn.close()
    elif request.param == "postgres":
        pool = ConnectionPool(
            "postgresql://postgres:postgres@localhost:5432/langgraph",
            max_size=2,
            kwargs={"autocommit": True}
        )
        yield PostgresSaver(pool)
        pool.close()

def test_basic_persistence(checkpointer):
    graph = build_graph(checkpointer)
    config = {"configurable": {"thread_id": "verify-1"}}

    # First invocation
    result1 = graph.invoke(
        {"messages": [HumanMessage(content="Hello")], "user_name": "Test", "turn_count": 0},
        config=config
    )
    assert result1["turn_count"] == 1
    assert isinstance(result1["messages"][-1], AIMessage)

    # Second invocation — same thread
    result2 = graph.invoke(
        {"messages": [HumanMessage(content="World")]},
        config=config
    )
    assert result2["turn_count"] == 2
    # History should contain both user messages
    user_msgs = [m for m in result2["messages"] if isinstance(m, HumanMessage)]
    assert len(user_msgs) == 2

def test_resume_after_new_graph_instance(checkpointer):
    """Simulate process restart by creating a new graph with the same checkpointer."""
    config = {"configurable": {"thread_id": "verify-2"}}

    graph1 = build_graph(checkpointer)
    result1 = graph1.invoke(
        {"messages": [HumanMessage(content="Remember: 42")], "user_name": "Test", "turn_count": 0},
        config=config
    )

    # New graph instance (simulates new process)
    graph2 = build_graph(checkpointer)
    result2 = graph2.invoke(
        {"messages": [HumanMessage(content="What number?")]},
        config=config
    )

    assert "42" in result2["messages"][-1].content
    assert result2["turn_count"] == 2

def test_checkpoint_listing(checkpointer):
    graph = build_graph(checkpointer)
    config = {"configurable": {"thread_id": "verify-3"}}

    graph.invoke({"messages": [HumanMessage(content="One")], "user_name": "Test", "turn_count": 0}, config=config)
    graph.invoke({"messages": [HumanMessage(content="Two")]}, config=config)

    checkpoints = list(checkpointer.list(config))
    assert len(checkpoints) == 2
    # Latest should be last
    assert checkpoints[-1].checkpoint["channel_values"]["turn_count"] == 2

def test_parallel_threads_isolated(checkpointer):
    graph = build_graph(checkpointer)
    
    result_a = graph.invoke(
        {"messages": [HumanMessage(content="Thread A")], "user_name": "A", "turn_count": 0},
        config={"configurable": {"thread_id": "thread-a"}}
    )
    result_b = graph.invoke(
        {"messages": [HumanMessage(content="Thread B")], "user_name": "B", "turn_count": 0},
        config={"configurable": {"thread_id": "thread-b"}}
    )

    # Resume A — should not see B's messages
    result_a2 = graph.invoke(
        {"messages": [HumanMessage(content="Who am I?")]},
        config={"configurable": {"thread_id": "thread-a"}}
    )
    assert "A" in result_a2["messages"][-1].content or "Thread A" in str(result_a2["messages"])

if __name__ == "__main__":
    # Run without pytest for quick manual verification
    print("Testing MemorySaver...")
    test_basic_persistence(MemorySaver())
    test_resume_after_new_graph_instance(MemorySaver())
    test_checkpoint_listing(MemorySaver())
    test_parallel_threads_isolated(MemorySaver())
    print("All memory tests passed.")

    print("\nTesting SqliteSaver...")
    conn = sqlite3.connect(":memory:", check_same_thread=False)
    test_basic_persistence(SqliteSaver(conn))
    test_resume_after_new_graph_instance(SqliteSaver(conn))
    test_checkpoint_listing(SqliteSaver(conn))
    test_parallel_threads_isolated(SqliteSaver(conn))
    conn.close()
    print("All SQLite tests passed.")

Run with python test_checkpoint_verification.py for a quick pass, or pytest test_checkpoint_verification.py -v for the full matrix. The parametrized fixture runs every test against all three backends, catching backend-specific bugs early.

Step 9: Configure checkpoint granularity for cost and latency

By default, LangGraph checkpoints after every node. For graphs with many fast nodes, this creates write overhead. Control it with checkpoint_during at compile time:

# graph.py (updated compile)
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["chat"],  # Checkpoint before this node
    interrupt_after=["greet"],   # Checkpoint after this node
)

Or disable automatic checkpoints entirely and call graph.update_state manually:

# Manual checkpointing example
config = {"configurable": {"thread_id": "manual-1"}}
result = graph.invoke({"messages": [HumanMessage(content="Start")], "user_name": "Ada", "turn_count": 0}, config=config)

# ... many fast node executions later ...

# Explicitly save a checkpoint
graph.update_state(config, {"turn_count": result["turn_count"]}, as_node="chat")

This is useful when you have a high-throughput inner loop (e.g., tool-calling agent) and only want durable state at human-visible boundaries.

Step 10: Monitor checkpoint health in production

Add observability around your checkpointer. Key metrics to emit:

• checkpoint_write_latency_seconds — histogram per backend
• checkpoint_read_latency_seconds — histogram per backend
• checkpoint_size_bytes — histogram, alert on growth
• checkpoint_errors_total — counter by error type (timeout, serialization, lock contention)
• active_threads — gauge, tracks conversation volume

Example with prometheus-client:

# metrics.py
from prometheus_client import Histogram, Counter, Gauge
from functools import wraps
import time

CHECKPOINT_WRITE_LATENCY = Histogram(
    "langgraph_checkpoint_write_latency_seconds",
    "Time spent writing checkpoints",
    ["backend", "thread_id"]
)
CHECKPOINT_READ_LATENCY = Histogram(
    "langgraph_checkpoint_read_latency_seconds",
    "Time spent reading checkpoints",
    ["backend", "thread_id"]
)
CHECKPOINT_ERRORS = Counter(
    "langgraph_checkpoint_errors_total",
    "Checkpoint errors",
    ["backend", "error_type"]
)
ACTIVE_THREADS = Gauge(
    "langgraph_active_threads",
    "Number of active conversation threads",
    ["backend"]
)

def instrument_checkpointer(checkpointer, backend_name: str):
    original_put = checkpointer.put
    original_get = checkpointer.get_tuple
    original_list = checkpointer.list

    @wraps(original_put)
    def put_wrapper(*args, **kwargs):
        start = time.time()
        try:
            return original_put(*args, **kwargs)
        except Exception as e:
            CHECKPOINT_ERRORS.labels(backend=backend_name, error_type=type(e).__name__).inc()
            raise
        finally:
            CHECKPOINT_WRITE_LATENCY.labels(backend=backend_name, thread_id=args[0]["configurable"]["thread_id"]).observe(time.time() - start)

    @wraps(original_get)
    def get_wrapper(*args, **kwargs):
        start = time.time()
        try:
            return original_get(*args, **kwargs)
        except Exception as e:
            CHECKPOINT_ERRORS.labels(backend=backend_name, error_type=type(e).__name__).inc()
            raise
        finally:
            CHECKPOINT_READ_LATENCY.labels(backend=backend_name, thread_id=args[0]["configurable"]["thread_id"]).observe(time.time() - start)

    checkpointer.put = put_wrapper
    checkpointer.get_tuple = get_wrapper
    checkpointer.list = original_list  # List is less critical
    return checkpointer

Wrap your checkpointer at initialization:

from metrics import instrument_checkpointer

checkpointer = instrument_checkpointer(PostgresSaver(pool), "postgres")
graph = build_graph(checkpointer)

Dashboards on these metrics reveal connection pool exhaustion, serialization regressions, and runaway state growth before they cause outages.

You now have a complete LangGraph checkpointer state persistence stack: in-memory for tests, SQLite for local development, Postgres for production, a custom Redis implementation for specialized needs, schema migration strategy, automated verification, granularity control, and production observability. The same graph.invoke call works across all of them — the checkpointer is a swappable implementation detail.