# Checkpointing

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

---

## Why Checkpointing?

| Need | Solution |
|---|---|
| **Crash recovery** | Resume from last checkpoint |
| **Human-in-the-loop** | Pause → human reviews → resume |
| **Time-travel debugging** | Fork from any past state |
| **Multi-session continuity** | Resume yesterday's work |
| **Audit trail** | Immutable execution log |
| **A/B testing** | Fork same state, different params |
| **Cost recovery** | Don't redo expensive computation |

---

## State Schema

### Minimal Serializable State
```python
@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)
```python
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)
```python
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)
```python
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)
```python
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)
```python
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)
```python
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)
```python
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)
```python
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
```python
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
```python
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
```python
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
| Model | Consistency | Latency | Use Case |
|---|---|---|---|
| **Async (fire-and-forget)** | Eventual | Lowest | High-throughput |
| **Sync (wait for quorum)** | Strong | Higher | Financial, legal |
| **Async + periodic sync** | Eventual + periodic | Low | Most agents |

### Implementation
```python
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

- [[StateManagement/Session-Management.md]] — Session lifecycle
- [[StateManagement/Event-Sourcing.md]] — Event-based state
- [[StateManagement/Distributed-State.md]] — Distributed patterns
- [[StateManagement/Versioning-Migration.md]] — Migration deep dive
- [[Frameworks/LangGraph.md]] — LangGraph checkpointer
- [[Frameworks/AgentKit.md]] — AgentKit checkpointer
- [[ContextEngineering/Memory-Systems.md]] — Memory as state
- [[Production/Reliability.md]] — Recovery patterns
- [[Production/Observability.md]] — Checkpoint metrics

---

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