Checkpointing long agent runs without bloating your database

Long-running agents accumulate state fast. Naive checkpointing agent runs database designs dump every intermediate step into a table, and within a day you’ve got millions of rows slowing down queries and inflating storage costs. This guide shows how to store only what you need to resume, prune the rest, and keep your database lean.

Step 1: Define a minimal checkpoint schema

Start by separating immutable run metadata from the mutable state blob. A checkpoint should answer three questions: where am I, what happened last, and how do I get back? Use a narrow table with a JSONB column for state and a parent pointer for chaining.

CREATE TABLE agent_checkpoints (
    run_id UUID NOT NULL,
    step_id BIGINT NOT NULL,
    parent_step_id BIGINT,
    created_at TIMESTAMPTZ DEFAULT now(),
    state_json JSONB NOT NULL,
    is_snapshot BOOLEAN DEFAULT FALSE,
    PRIMARY KEY (run_id, step_id)
);

Index run_id, step_id for fast resume queries. Do not store full conversation transcripts or tool outputs inline if they exceed a few KB—more on that in Step 4.

In Python, map this with a tiny dataclass:

from dataclasses import dataclass
import json

@dataclass
class Checkpoint:
    run_id: str
    step_id: int
    parent_step_id: int | None
    state: dict
    is_snapshot: bool = False

    def to_row(self):
        return {
            "run_id": self.run_id,
            "step_id": self.step_id,
            "parent_step_id": self.parent_step_id,
            "state_json": json.dumps(self.state),
            "is_snapshot": self.is_snapshot,
        }

Step 2: Serialize agent state with delta compression

Writing the full state every step is the primary cause of bloat. If your agent state is a dictionary that changes incrementally, store JSON Merge Patch (RFC 7386) deltas instead of full dumps. Keep the last full snapshot in memory or in the row marked is_snapshot.

def make_delta(prev: dict, curr: dict) -> dict:
    delta = {}
    for k, v in curr.items():
        if k not in prev or prev[k] != v:
            delta[k] = v
    for k in prev:
        if k not in curr:
            delta[k] = None  # RFC 7386: null means delete
    return delta

def apply_delta(base: dict, delta: dict) -> dict:
    merged = base.copy()
    for k, v in delta.items():
        if v is None:
            merged.pop(k, None)
        else:
            merged[k] = v
    return merged

On each step, compute delta = make_delta(last_snapshot, current_state), write a checkpoint row with state_json set to the delta (not the full state), and set is_snapshot=False. This cuts row size by 10–100x for typical agent loops where only a few keys change per step.

Step 3: Implement periodic full snapshots and prune intermediates

Deltas are cheap but replaying 500 deltas on resume is slow and error-prone. Take a full snapshot every SNAPSHOT_INTERVAL steps (e.g., 20). After writing a snapshot, delete all non-snapshot checkpoints for that run with step_id less than the new snapshot’s step_id.

SNAPSHOT_INTERVAL = 20

def save_checkpoint(conn, cp: Checkpoint, step: int):
    with conn.cursor() as cur:
        if step % SNAPSHOT_INTERVAL == 0:
            cp.is_snapshot = True
            cur.execute(
                "INSERT INTO agent_checkpoints (run_id, step_id, parent_step_id, state_json, is_snapshot) "
                "VALUES (%s, %s, %s, %s, TRUE)",
                (cp.run_id, cp.step_id, cp.parent_step_id, json.dumps(cp.state))
            )
            cur.execute(
                "DELETE FROM agent_checkpoints WHERE run_id=%s AND step_id <%s AND is_snapshot=FALSE",
                (cp.run_id, cp.step_id)
            )
        else:
            delta = make_delta(load_last_snapshot(conn, cp.run_id), cp.state)
            cur.execute(
                "INSERT INTO agent_checkpoints (run_id, step_id, parent_step_id, state_json, is_snapshot) "
                "VALUES (%s, %s, %s, %s, FALSE)",
                (cp.run_id, cp.step_id, cp.parent_step_id, json.dumps(delta))
            )
    conn.commit()

Resuming becomes a two-row read: the latest snapshot plus any deltas after it.

Step 4: Offload large artifacts to object storage

Agent state often includes retrieved documents, generated files, or embedding vectors. Storing these in Postgres JSONB will bloat your checkpointing agent runs database faster than anything else. Push anything >4 KB to S3 (or equivalent) and keep only a URI and checksum in the state.

import boto3, hashlib

s3 = boto3.client("s3")

def offload_blob(run_id: str, key: str, data: bytes) -> str:
    obj_key = f"{run_id}/{key}"
    s3.put_object(Bucket="agent-artifacts", Key=obj_key, Body=data)
    return f"s3://agent-artifacts/{obj_key}"

def checkpoint_with_file(run_id: str, step: int, file_bytes: bytes, state: dict):
    uri = offload_blob(run_id, f"step_{step}.bin", file_bytes)
    state["artifact_uri"] = uri
    state["artifact_sha"] = hashlib.sha256(file_bytes).hexdigest()
    return state

On resume, lazy-load the artifact only if the step needs it. This keeps checkpoint rows under a few hundred bytes.

Step 5: Wire checkpointing into the agent loop

A robust agent loop wraps each step in a try/except that persists state before executing side effects. If the process dies, a separate worker picks up the last checkpoint and continues.

def run_agent(run_id: str, max_steps: int, conn):
    step = load_resume_step(conn, run_id)
    state = load_latest_state(conn, run_id) or {"history": []}
    while step < max_steps:
        try:
            # optionally route LLM calls through a gateway that honors
            # client routing directives; n4n.ai forwards provider cache-control
            # hints so prompt prefixes stay cached across retries.
            action = plan_next(state)
            state = execute_action(state, action)
            step += 1
            save_checkpoint(conn, Checkpoint(run_id, step, step-1, state), step)
        except TransientError as e:
            log.warning("step %d failed, will retry from checkpoint", step)
            state = load_latest_state(conn, run_id)
            continue
        except FatalError:
            break

The key discipline: never mutate state without immediately writing a checkpoint. If execute_action has side effects (sending email, calling API), make those idempotent keyed by step_id.

Step 6: Verify success without guessing

You need proof that your checkpointing agent runs database strategy works and stays small. Two tests:

1. Crash resume test. Force-kill the agent process at step 37. Restart with the same run_id. Assert that step resumes at 38 and the final output matches a non-crashed run.
2. Growth assertion. Run the agent for 1,000 steps with SNAPSHOT_INTERVAL=20. Query the table size:

SELECT pg_total_relation_size('agent_checkpoints') AS bytes,
       count(*) AS rows
FROM agent_checkpoints
WHERE run_id = 'test-run';

Expect ~50 rows (20 snapshots + deltas between) and a size under a few MB even if intermediate states included large artifacts (which are in S3). If row count grows linearly with steps, your pruning is broken.

Add a CI job that fails if rows > max_steps / SNAPSHOT_INTERVAL * 2. That catches regressions where someone disables pruning.

Trade-offs and edge cases

Delta compression assumes state is mostly JSON-serializable and conflicts are rare. If your agent stores binary blobs or non-deterministic objects, offload them entirely. For highly concurrent agents (multiple workers on same run), use SELECT ... FOR UPDATE on the latest snapshot to avoid interleaved deltas.

Checkpointing agent runs database design is not about capturing everything—it’s about capturing enough. Treat your database as a resume log, not an audit trail. If you need full audit, stream raw events to an append-only log or warehouse, not to the hot path.

Following these steps gives you constant-size storage per run regardless of length, fast resume, and a database that won’t page out under agent load.