← Back to Blog
Oct 24, 2026 · 10 min read

Your Agent Dies Mid-Run. LangGraph and Postgres Pick It Back Up

AI AgentsLangGraphPostgresPythonTutorial

The approval step is where agents stop being a demo. A checkout agent validates an order, calls interrupt() to ask a human about a 499 dollar charge, and the worker gets redeployed before anyone clicks approve. The pod disappears. So does the pending approval, because it lived in a Python dict.

The customer retries. Now there are two charges and one confused support thread.

None of that is a model failure. It is a state failure, and it has a boring fix: write the run to Postgres at every step so a different process can pick it up. This guide builds a three-node graph that pauses for approval, ends the process, resumes it in a fresh container, and then goes through the four traps that make this pattern break in production.

Prerequisites

  • Docker for the Postgres container, or any Postgres 14+
  • Python 3.10+
  • About 15 minutes
pip install -U langgraph "langgraph-checkpoint-postgres" "psycopg[binary,pool]"

Everything below runs against langgraph 1.2.16, which is what pip installed when this was written.

Step 1: Postgres that outlives your process

docker run -d --name agent-pg -e POSTGRES_PASSWORD=postgres -p 5432:5432 postgres:16-alpine

Then export the connection string:

export DB_URI="postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

sslmode=disable is fine for a local container. Keep TLS on for anything remote.

Step 2: a graph that needs a human

Save this as demo.py. Three nodes: validate, charge, notify. Only charge interrupts.

import operator
import os
from typing import Annotated, TypedDict

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt

DB_URI = os.environ["DB_URI"]
THREAD = "order-7f3a"


class State(TypedDict):
    order_id: str
    total_cents: int
    log: Annotated[list[str], operator.add]
    approved: bool


def validate(state: State) -> dict:
    return {"log": [f"validated {state['order_id']}"]}


def charge(state: State) -> dict:
    decision = interrupt(
        {
            "question": "Charge this card?",
            "order_id": state["order_id"],
            "amount_cents": state["total_cents"],
        }
    )
    if not decision.get("approved"):
        return {"log": ["charge rejected by human"], "approved": False}
    return {"log": [f"charged {state['total_cents']} cents"], "approved": True}


def notify(state: State) -> dict:
    return {"log": ["customer notified"]}


def build(checkpointer):
    g = StateGraph(State)
    g.add_node("validate", validate)
    g.add_node("charge", charge)
    g.add_node("notify", notify)
    g.add_edge(START, "validate")
    g.add_edge("validate", "charge")
    g.add_edge("charge", "notify")
    g.add_edge("notify", END)
    return g.compile(checkpointer=checkpointer)


if __name__ == "__main__":
    config = {"configurable": {"thread_id": THREAD}}
    with PostgresSaver.from_conn_string(DB_URI) as cp:
        cp.setup()
        graph = build(cp)
        graph.invoke(
            {"order_id": "A-1042", "total_cents": 49900, "log": []},
            config,
            durability="sync",
        )
        snap = graph.get_state(config)
        print("next nodes:", snap.next)
        print("interrupt payload:", snap.tasks[0].interrupts[0].value)
        print("state log:", snap.values["log"])

Run it:

next nodes: ('charge',)
interrupt payload: {'question': 'Charge this card?', 'order_id': 'A-1042', 'amount_cents': 49900}
state log: ['validated A-1042']

The script exits while the run is still unfinished. snap.next is ('charge',), meaning the graph is waiting to run that node again, this time with an answer attached.

setup() is what creates the four tables, checkpoints, checkpoint_blobs, checkpoint_writes, and checkpoint_migrations. Call it once as a migration step in your deploy, not on every request.

Step 3: resume in a fresh process

Save this as resume.py:

import os

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import Command

from demo import THREAD, build

DB_URI = os.environ["DB_URI"]
config = {"configurable": {"thread_id": THREAD}}

with PostgresSaver.from_conn_string(DB_URI) as cp:
    graph = build(cp)
    before = graph.get_state(config)
    print("state before resume:", before.values, "next:", before.next)
    out = graph.invoke(Command(resume={"approved": True}), config, durability="sync")
    print("final log:", out["log"])
    print("checkpoints stored:", len(list(graph.get_state_history(config))))

Run it as a separate process, not from inside demo.py:

state before resume: {'order_id': 'A-1042', 'total_cents': 49900, 'log': ['validated A-1042']} next: ('charge',)
final log: ['validated A-1042', 'charged 49900 cents', 'customer notified']
checkpoints stored: 5

I ran those two scripts as two separate containers sharing only the Postgres instance. The second one had no memory of the first, and still knew the order id, the amount, and that validation had already passed. Five checkpoint rows for a three-node run: one for the input, then one per super-step.

Step 4: put it behind HTTP

The restart story only pays off if your API can find the paused run later. Two things matter: a connection pool instead of a single connection, and setup() at boot.

import os

from fastapi import FastAPI
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import Command
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool

from demo import build

DB_URI = os.environ["DB_URI"]

pool = ConnectionPool(
    conninfo=DB_URI,
    max_size=10,
    kwargs={"autocommit": True, "prepare_threshold": 0, "row_factory": dict_row},
)
pool.wait()

checkpointer = PostgresSaver(pool)
checkpointer.setup()
graph = build(checkpointer)

app = FastAPI()


def cfg(thread_id: str):
    return {"configurable": {"thread_id": thread_id}}


@app.get("/orders/{thread_id}")
def pending(thread_id: str):
    snap = graph.get_state(cfg(thread_id))
    return {"waiting_on": list(snap.next), "state": snap.values}


@app.post("/orders/{thread_id}")
def start(thread_id: str, body: dict):
    graph.invoke(body, cfg(thread_id), durability="sync")
    return pending(thread_id)


@app.post("/orders/{thread_id}/approve")
def approve(thread_id: str, body: dict):
    graph.invoke(Command(resume=body), cfg(thread_id), durability="sync")
    return pending(thread_id)

The routes are plain def so FastAPI runs them in its threadpool, which is what the sync PostgresSaver expects. If the rest of your app is async, use AsyncPostgresSaver and await graph.ainvoke(...) instead. Calling the sync saver from an async def handler blocks the event loop on every database write, and that only shows up under load.

The traps

Trap 1: the node re-runs from the top

Resuming does not continue the middle of a function. The node starts again, and interrupt() returns the resume value where it previously stopped, but every line above that call executes twice. I counted it with a print above the interrupt: one execution before the pause, a second one on resume.

So this is a double charge waiting to happen:

def charge(state: State) -> dict:
    receipt = stripe.PaymentIntent.create(amount=state["total_cents"])
    decision = interrupt({"receipt": receipt.id})

Keep side effects below the interrupt, or pass an idempotency key built from thread_id and the node name so the repeat call becomes a no-op.

Trap 2: new input is not a resume

When a graph is paused, invoke() with fresh input does not raise. It accepts the input, merges it into state, and re-runs the interrupted node. I tested exactly that: no exception, the process kept going, and the string I passed showed up in the final log.

Branch on state before deciding what to do:

snap = graph.get_state(cfg)
if snap.next:
    graph.invoke(Command(resume=answer), cfg, durability="sync")
else:
    graph.invoke(new_input, cfg, durability="sync")

Without that check, a user clicking approve twice, or a retry from your queue, quietly pushes a second payload into the same thread.

Trap 3: the default write is async

durability takes exit, async, or sync. Leave it unset and you get async, which writes checkpoints in the background while the next step runs. Crash inside that window and the last step is gone. sync writes before the next node starts, which is what you want around money, emails, and anything a human approved. It costs an extra database round trip per step, so a 20 step agent pays it 20 times.

One more detail from the source: durability is ignored, with a warning, when the graph has no checkpointer.

Trap 4: one connection is not a pool

PostgresSaver.from_conn_string() opens a single connection and hands it to the saver. Fine for a script. In a server handling concurrent runs, that one connection becomes the bottleneck and a source of cross-talk between requests. Build the saver from a ConnectionPool like the FastAPI example above.

Edit state before you resume

An approval UI usually needs more than a yes or no. Say the human lowers the amount. update_state writes a new checkpoint, and the resumed node sees the edited value:

graph.update_state(cfg, {"total_cents": 19900}, as_node="validate")
graph.invoke(Command(resume={"approved": True}), cfg, durability="sync")

Output from that run:

log after edited resume: ['validated A-1042', 'charged 19900 cents', 'customer notified']

as_node matters. It tells LangGraph which node produced the edit, and that decides what runs next. Get it wrong and you skip or repeat a step.

The same history powers replay. get_state_history() returns newest first, and invoking with an older checkpoint_id re-runs everything after it. Interrupts fire again during replay, which is handy when you are debugging an approval flow.

Storage and cleanup

Four tables, one row per checkpoint, full channel values at each step. Cheap for small state, not cheap for threads carrying long chat histories or big tool outputs. Delete finished threads when you are done with them:

checkpointer.delete_thread(thread_id)

Put that on a schedule, or add a retention job on the checkpoint tables. For memory that should cross threads, like user preferences, use a Store alongside the checkpointer instead of stuffing it into graph state.

Recap

  • Compile the graph with a Postgres checkpointer and pass a thread_id on every call.
  • Call setup() once at deploy time.
  • interrupt() pauses and persists. Command(resume=...) continues, even from another process.
  • Everything above the interrupt() line runs twice. Move side effects below it, or make them idempotent.
  • Check get_state().next before invoking with new input.
  • Use durability="sync" for steps you cannot afford to lose.
  • Build the saver from a ConnectionPool, not a single connection.

The next useful step: move the resume call behind a real webhook from your approval UI, and put a retry policy on the charge node.

References

Need Help Implementing This?

I help teams design and build scalable cloud infrastructure, DevOps pipelines, and production-grade systems.

Book a Free Consultation