Checkpointing

Serializable agent state: persistence, recovery, time-travel, and migration.

Why Checkpointing?

State Schema

Minimal Serializable State

@dataclass
class AgentState:
    # Version for migration
    schema_version: int = 2

    # Identity
    thread_id: str
    agent_id: str
    user_id: str

    # Task
    goal: str
    plan: list[PlanStep] | None
    current_step: int

    # Conversation
    messages: list[Message]  # Full history or compressed

    # Memory
    working_memory: WorkingMemory
    short_term_memory: ConversationMemory
    long_term_refs: list[MemoryRef]  # IDs, not full content

    # Tools
    available_tools: list[ToolSpec]
    tool_call_history: list[ToolCall]

    # Execution
    iterations: int
    tokens_used: int
    cost_usd: float
    start_time: datetime
    last_checkpoint: datetime

    # Control
    status: Literal["running", "paused", "completed", "failed", "awaiting_human"]
    checkpointer_metadata: dict

State Reducers (Merge Strategies)

class StateReducers:
    @staticmethod
    def messages(old: list, new: list) -> list:
        # Append with deduplication by ID
        seen = {m.id for m in old}
        return old + [m for m in new if m.id not in seen]

    @staticmethod
    def working_memory(old: WorkingMemory, new: WorkingMemory) -> WorkingMemory:
        return new  # Replace (latest wins)

    @staticmethod
    def tool_calls(old: list, new: list) -> list:
        return old + new  # Append all

    @staticmethod
    def counters(old: dict, new: dict) -> dict:
        return {k: old.get(k, 0) + new.get(k, 0) for k in set(old) | set(new)}

Checkpointer Backends

1. In-Memory (Development)

class InMemoryCheckpointer:
    def __init__(self):
        self.store: dict[str, AgentState] = {}

    async def save(self, thread_id: str, state: AgentState):
        self.store[thread_id] = state

    async def load(self, thread_id: str) -> AgentState | None:
        return self.store.get(thread_id)

    async def list(self, prefix: str = "") -> list[str]:
        return [k for k in self.store if k.startswith(prefix)]

2. SQLite (Single-Node Production)

class SQLiteCheckpointer:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS checkpoints (
                thread_id TEXT PRIMARY KEY,
                state_json TEXT NOT NULL,
                schema_version INTEGER NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        self.conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_updated ON checkpoints(updated_at)
        """)

    async def save(self, thread_id: str, state: AgentState):
        state.last_checkpoint = datetime.utcnow()
        self.conn.execute(
            "INSERT OR REPLACE INTO checkpoints VALUES (?, ?, ?, ?, ?)",
            (thread_id, json.dumps(asdict(state)), state.schema_version, 
             state.start_time, state.last_checkpoint)
        )
        self.conn.commit()

    async def load(self, thread_id: str) -> AgentState | None:
        row = self.conn.execute(
            "SELECT state_json FROM checkpoints WHERE thread_id = ?", 
            (thread_id,)
        ).fetchone()
        if row:
            return migrate_state(json.loads(row[0]))
        return None

3. PostgreSQL (Multi-Node Production)

class PostgresCheckpointer:
    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool

    async def setup(self):
        await self.pool.execute("""
            CREATE TABLE IF NOT EXISTS checkpoints (
                thread_id TEXT PRIMARY KEY,
                state_json JSONB NOT NULL,
                schema_version INTEGER NOT NULL,
                created_at TIMESTAMPTZ DEFAULT NOW(),
                updated_at TIMESTAMPTZ DEFAULT NOW()
            );
            CREATE INDEX IF NOT EXISTS idx_updated ON checkpoints(updated_at);
        """)

    async def save(self, thread_id: str, state: AgentState):
        async with self.pool.acquire() as conn:
            await conn.execute("""
                INSERT INTO checkpoints (thread_id, state_json, schema_version, updated_at)
                VALUES ($1, $2, $3, NOW())
                ON CONFLICT (thread_id) DO UPDATE SET
                    state_json = EXCLUDED.state_json,
                    schema_version = EXCLUDED.schema_version,
                    updated_at = NOW()
            """, thread_id, json.dumps(asdict(state)), state.schema_version)

    async def load(self, thread_id: str) -> AgentState | None:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT state_json FROM checkpoints WHERE thread_id = $1", thread_id
            )
            if row:
                return migrate_state(row['state_json'])
            return None

4. Redis (High-Throughput, Ephemeral)

class RedisCheckpointer:
    def __init__(self, redis: redis.Redis, ttl: int = 86400 * 7):
        self.redis = redis
        self.ttl = ttl

    async def save(self, thread_id: str, state: AgentState):
        key = f"checkpoint:{thread_id}"
        await self.redis.setex(key, self.ttl, json.dumps(asdict(state)))

    async def load(self, thread_id: str) -> AgentState | None:
        data = await self.redis.get(f"checkpoint:{thread_id}")
        if data:
            return migrate_state(json.loads(data))
        return None

Checkpointing Strategies

1. Periodic (Time-Based)

class PeriodicCheckpointer:
    def __init__(self, checkpointer, interval_seconds=30):
        self.checkpointer = checkpointer
        self.interval = interval_seconds

    async def run(self, thread_id: str, get_state: Callable):
        while True:
            await asyncio.sleep(self.interval)
            state = await get_state()
            if state and state.status == "running":
                await self.checkpointer.save(thread_id, state)

2. Event-Based (After Each Turn)

class EventBasedCheckpointer:
    def __init__(self, checkpointer):
        self.checkpointer = checkpointer

    async def on_turn_complete(self, thread_id: str, state: AgentState):
        if state.status in ("running", "paused", "awaiting_human"):
            await self.checkpointer.save(thread_id, state)

3. Hybrid (Smart)

class SmartCheckpointer:
    def __init__(self, checkpointer, min_interval=10, max_interval=300):
        self.checkpointer = checkpointer
        self.min_interval = min_interval
        self.max_interval = max_interval
        self.last_save = {}

    async def maybe_save(self, thread_id: str, state: AgentState, event: str):
        # Always save on critical events
        if event in ("human_interrupt", "error", "tool_call", "plan_change"):
            await self.checkpointer.save(thread_id, state)
            self.last_save[thread_id] = time.time()
            return

        # Time-based with adaptive interval
        last = self.last_save.get(thread_id, 0)
        elapsed = time.time() - last

        # More frequent if active
        interval = self.min_interval if state.iterations > 10 else self.max_interval

        if elapsed >= interval:
            await self.checkpointer.save(thread_id, state)
            self.last_save[thread_id] = time.time()

Migration (Schema Evolution)

Versioned State

def migrate_state(state: dict) -> AgentState:
    version = state.get("schema_version", 1)

    if version == 1:
        # v1 → v2: Added cost tracking, tool call history
        state.setdefault("tokens_used", 0)
        state.setdefault("cost_usd", 0.0)
        state.setdefault("tool_call_history", [])
        state["schema_version"] = 2

    if version == 2:
        # v2 → v3: Added plan field, separated memory refs
        state.setdefault("plan", None)
        state.setdefault("long_term_refs", [])
        # Migrate inline memory to refs
        if "long_term_memory" in state:
            refs = extract_to_refs(state.pop("long_term_memory"))
            state["long_term_refs"] = refs
        state["schema_version"] = 3

    return AgentState(**state)

Migration Best Practices

1. Never delete fields — mark deprecated, keep for backward compat
2. Always provide defaults — setdefault() for new fields
3. Test migrations — Unit test each version upgrade
4. One-way migrations — Never downgrade schema version
5. Log migrations — Track which threads migrated when

Time-Travel & Forking

State History

class HistoryCheckpointer:
    def __init__(self, base_checkpointer):
        self.base = base_checkpointer

    async def save(self, thread_id: str, state: AgentState):
        # Save to history
        history_key = f"{thread_id}:history:{state.last_checkpoint.isoformat()}"
        await self.base.save(history_key, state)
        # Update current
        await self.base.save(thread_id, state)

    async def get_history(self, thread_id: str) -> list[AgentState]:
        keys = await self.base.list(f"{thread_id}:history:")
        states = []
        for key in sorted(keys, reverse=True):
            state = await self.base.load(key)
            if state:
                states.append(state)
        return states

    async def fork(self, thread_id: str, timestamp: datetime, 
                   new_thread_id: str) -> AgentState:
        history_key = f"{thread_id}:history:{timestamp.isoformat()}"
        state = await self.base.load(history_key)
        if state:
            state.thread_id = new_thread_id
            state.start_time = datetime.utcnow()
            state.last_checkpoint = datetime.utcnow()
            await self.base.save(new_thread_id, state)
        return state

Time-Travel Debugging

class TimeTravelDebugger:
    def __init__(self, checkpointer: HistoryCheckpointer):
        self.checkpointer = checkpointer

    async def inspect(self, thread_id: str, timestamp: datetime) -> AgentState:
        return await self.checkpointer.load(f"{thread_id}:history:{timestamp.isoformat()}")

    async def diff(self, thread_id: str, t1: datetime, t2: datetime) -> StateDiff:
        s1 = await self.inspect(thread_id, t1)
        s2 = await self.inspect(thread_id, t2)
        return compute_diff(s1, s2)

    async def replay(self, thread_id: str, from_timestamp: datetime, 
                     to_timestamp: datetime) -> list[StateTransition]:
        # Replay state transitions between timestamps
        pass

Distributed Checkpointing

Consistency Models

Implementation

class DistributedCheckpointer:
    def __init__(self, primary: Checkpointer, replicas: list[Checkpointer]):
        self.primary = primary
        self.replicas = replicas

    async def save(self, thread_id: str, state: AgentState):
        # Write to primary (sync)
        await self.primary.save(thread_id, state)

        # Replicate async
        asyncio.create_task(self._replicate(thread_id, state))

    async def _replicate(self, thread_id: str, state: AgentState):
        await asyncio.gather(*[
            r.save(thread_id, state) for r in self.replicas
        ], return_exceptions=True)

    async def load(self, thread_id: str) -> AgentState | None:
        # Read from primary
        return await self.primary.load(thread_id)

Related Notes

• Session Management — Session lifecycle
• Event Sourcing — Event-based state
• Distributed State — Distributed patterns
• Versioning Migration — Migration deep dive
• LangGraph — LangGraph checkpointer
• AgentKit — AgentKit checkpointer
• Memory Systems — Memory as state
• Reliability — Recovery patterns
• Observability — Checkpoint metrics

Sources: LangGraph checkpointer, AgentKit, temporal.io, event sourcing patterns, SQLite/PostgreSQL/Redis docs, distributed systems patterns (Raft, Paxos), schema migration best practices.