How to Implement Transactional State Rollbacks for AI Agents in LangGraph and PostgreSQL

Picture this. Your support agent is halfway through a refund workflow. The model calls three tools in one step: create a credit memo, deduct the customer's balance, and queue a card refund. The first two succeed. The third receives a hallucinated argument, a customer_id that is a string with a stray quote in it, and blows up. The agent framework dutifully passes the error back to the LLM, the LLM says "let me try that again," and calls all three tools again.

Congratulations, you just issued two credit memos and deducted the balance twice.

This is the failure mode nobody puts in the demo video. Agents are not scripts. They crash, they time out, and they occasionally invent arguments that look plausible and are completely wrong. When those agents write to a real database, "just retry" is a dangerous strategy unless the database is guaranteed to be back in a known-good state first.

In this guide, we will build that guarantee. We will wrap multi-tool LangGraph node executions in PostgreSQL transactions with savepoints, detect partial failures, roll back deterministically, and only then let the model try again. Along the way we will deal with the two hardest parts of the problem: external side effects that a rollback cannot undo, and crashes that happen between your commit and LangGraph's checkpoint.

By the end you will have a working reference implementation you can adapt to your own stack, plus a checklist of the mistakes that cause the most production incidents.

Figure 1: Without transactional rollbacks, a mid-workflow failure leaves partial writes behind. With savepoints, the database returns to a known-good state before the LLM retries.

Why AI Agents Corrupt Databases in the First Place

Traditional backend code is deterministic. A developer decides the order of operations, and each step is written with its failure modes in mind. Agent code inverts that. The model decides what to call, in what order, and with what arguments, and it does so at runtime.

That creates four distinct sources of corruption:

• Hallucinated or malformed arguments. The model passes a valid-looking but wrong value: an ID that belongs to another tenant, a negative amount, a date in the wrong format. Some of these get past naive validation and only fail deep inside a query.
• Parallel tool calls with hidden dependencies. Modern models routinely emit several tool calls in a single step. Those calls are often logically dependent (create the invoice, then apply the credit), but the framework may execute them independently.
• Mid-node crashes. Pods get evicted, deploys roll out, and connections drop. If a node performs three writes and the process dies after the second, you have a half-state.
• Retries by design. The whole point of an agent loop is that the model reacts to errors and tries again. Retries are a feature, which means every tool must be safe to run after a partial failure.

Any one of these is manageable. Together, in a system that talks to real money, they demand a transactional discipline that most agent tutorials skip.

Core Concepts You Need Before Writing Code

ACID transactions and why they are not enough on their own

A PostgreSQL transaction groups several statements into one atomic unit. Either every change commits, or none of them do. That sounds like the whole solution, but there is a catch specific to PostgreSQL: once any statement inside a transaction raises an error, the transaction enters an aborted state. Every subsequent command fails with current transaction is aborted, commands ignored until end of transaction block (SQLSTATE 25P02) until you roll back.

For an agent, this matters because a batch of tool calls is not a single logical operation. You want to know exactly which call failed, and you may want to keep the calls that succeeded. A plain transaction gives you an all-or-nothing hammer and nothing finer.

Savepoints: nested undo points inside a transaction

A savepoint marks a point inside a transaction that you can roll back to without abandoning the whole transaction. The flow looks like this:

BEGIN;
SAVEPOINT tool_call_1;
-- ... writes for tool call 1 ...
RELEASE SAVEPOINT tool_call_1;     -- success, keep changes

SAVEPOINT tool_call_2;
-- ... writes for tool call 2 raise an error ...
ROLLBACK TO SAVEPOINT tool_call_2; -- undo only call 2, transaction is usable again

COMMIT;                            -- or ROLLBACK to discard everything

Figure 2: One outer transaction, one savepoint per tool call. A failed call rolls back to its own savepoint, and the atomic policy then discards the entire transaction.

With savepoints, each tool call becomes an independently revertible unit inside one outer transaction. You then choose a policy:

For anything involving money, default to the atomic policy. A half-applied financial workflow is almost always worse than a clean failure.

Where LangGraph fits in

LangGraph models an agent as a graph of nodes that read and update a shared state, with edges (including conditional ones) that decide what runs next. When you compile the graph with a checkpointer, LangGraph saves state at the boundaries between steps, which enables resuming, time travel, and human-in-the-loop flows.

Two facts about that design shape everything that follows:

1. Checkpoints are saved between steps, not inside them. If your tool node crashes halfway, LangGraph resumes from the last checkpoint and runs the node again from the start.
2. The checkpointer uses its own database connection and transaction. Even if it points at the same PostgreSQL instance as your business data, the checkpoint write and your business commit are two separate transactions. There is a tiny window between them where a crash leaves the business data committed but the graph unaware.

The first fact means nodes must be safe to replay. The second means we need idempotency, not just transactions. We will address both.

A quick note on naming. You may see references to "LangGraph error handlers." LangGraph does not ship a single API by that name. What it provides is a set of composable mechanisms: exceptions caught inside nodes, the handle_tool_errors option on the prebuilt ToolNode, per-node retry policies, and conditional edges. In this article, "error handler" means the custom layer we build from those parts.

The Architecture at a Glance

Here is the flow we are going to implement:

   ┌──────────┐   tool calls    ┌───────────────────────────────────────┐
   │  agent   │ ──────────────▶ │  transactional tools node             │
   │  (LLM)   │                 │  1. validate args (no DB touched)     │
   └──────────┘                 │  2. BEGIN + advisory lock             │
        ▲                       │  3. SAVEPOINT per call + ledger row   │
        │                       │  4. COMMIT or ROLLBACK                │
        │  sanitized errors     └──────────────┬────────────────────────┘
        │  + clean DB state                    │
        └──────────────────────────────────────┤
                                               │ too many rollbacks
                                               ▼
                                        ┌────────────┐
                                        │  escalate  │
                                        └────────────┘

   Post-commit (outside the graph):  outbox worker ──▶ Stripe (idempotency key)

The important properties:

• The LLM never holds a database connection. The transaction opens and closes entirely inside the tools node.
• Every tool call receives a response. Even calls that were rolled back because a sibling failed get an explicit "not applied" message, otherwise most model providers reject the next request.
• External effects are decoupled. The transaction only records the intent to charge a card. A separate worker performs the real call after commit.

Step 1: Design the Schema for Safety, Not Just Storage

Good rollbacks start with a schema that lets the database itself reject nonsense. Constraints are your last line of defense against hallucinated arguments.

CREATE TABLE customers (
    id            BIGINT PRIMARY KEY,
    email         TEXT   NOT NULL,
    credit_cents  BIGINT NOT NULL DEFAULT 0 CHECK (credit_cents >= 0)
);

CREATE TABLE invoices (
    id             BIGSERIAL PRIMARY KEY,
    customer_id    BIGINT NOT NULL REFERENCES customers(id),
    billing_period DATE   NOT NULL,
    amount_cents   BIGINT NOT NULL CHECK (amount_cents > 0),
    status         TEXT   NOT NULL DEFAULT 'draft',
    UNIQUE (customer_id, billing_period)
);

-- Idempotency ledger: one row per successfully applied tool call
CREATE TABLE tool_executions (
    idempotency_key TEXT PRIMARY KEY,
    tool_name       TEXT        NOT NULL,
    result          JSONB       NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Transactional outbox: intents to call external systems
CREATE TABLE outbox (
    id            BIGSERIAL PRIMARY KEY,
    topic         TEXT        NOT NULL,
    payload       JSONB       NOT NULL,
    dedupe_key    TEXT        NOT NULL UNIQUE,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    dispatched_at TIMESTAMPTZ
);

Notice what each object does for us:

• The CHECK (credit_cents >= 0) constraint means an agent that tries to deduct more credit than exists gets a hard database error, not a silently negative balance.
• The UNIQUE (customer_id, billing_period) constraint means the same billing period cannot be invoiced twice, no matter how many times the model asks.
• The tool_executions ledger makes replays safe. It is written in the same transaction as the business changes, so it exists if and only if those changes do.
• The outbox table turns "charge the card" into a row that can be rolled back, which is the trick that prevents duplicate SaaS billing.

Step 2: Define Tools With Strict Schemas and Typed Errors

Validate arguments before you touch the database. A Pydantic model catches the cheap failures, such as wrong types and out-of-range numbers, without opening a transaction at all.

# tools.py
from dataclasses import dataclass
from datetime import date
from typing import Awaitable, Callable

from psycopg import AsyncConnection
from psycopg.types.json import Jsonb
from pydantic import BaseModel, Field


class ToolBusinessError(Exception):
    """An expected failure that the LLM can understand and correct."""


class ApplyCredit(BaseModel):
    """Deduct account credit from a customer."""
    customer_id: int = Field(gt=0)
    amount_cents: int = Field(gt=0, le=1_000_000)


class CreateInvoice(BaseModel):
    """Create a draft invoice for a customer and billing period."""
    customer_id: int = Field(gt=0)
    billing_period: date
    amount_cents: int = Field(gt=0, le=1_000_000)


class QueueCharge(BaseModel):
    """Queue a card charge for an existing draft invoice."""
    invoice_id: int = Field(gt=0)


async def apply_credit(conn: AsyncConnection, args: ApplyCredit) -> dict:
    cur = await conn.execute(
        "UPDATE customers SET credit_cents = credit_cents - %s "
        "WHERE id = %s RETURNING credit_cents",
        (args.amount_cents, args.customer_id),
    )
    row = await cur.fetchone()
    if row is None:
        raise ToolBusinessError(f"Customer {args.customer_id} does not exist.")
    return {"remaining_credit_cents": row[0]}


async def create_invoice(conn: AsyncConnection, args: CreateInvoice) -> dict:
    cur = await conn.execute(
        "INSERT INTO invoices (customer_id, billing_period, amount_cents) "
        "VALUES (%s, %s, %s) RETURNING id",
        (args.customer_id, args.billing_period, args.amount_cents),
    )
    (invoice_id,) = await cur.fetchone()
    return {"invoice_id": invoice_id, "status": "draft"}


async def queue_charge(conn: AsyncConnection, args: QueueCharge) -> dict:
    cur = await conn.execute(
        "SELECT customer_id, amount_cents, status FROM invoices "
        "WHERE id = %s FOR UPDATE",
        (args.invoice_id,),
    )
    row = await cur.fetchone()
    if row is None:
        raise ToolBusinessError(f"Invoice {args.invoice_id} does not exist.")
    customer_id, amount_cents, status = row
    if status != "draft":
        raise ToolBusinessError(f"Invoice {args.invoice_id} is already {status}.")

    await conn.execute(
        "UPDATE invoices SET status = 'charge_queued' WHERE id = %s",
        (args.invoice_id,),
    )
    # The intent is recorded inside the transaction. The real Stripe call
    # happens later, in a worker, after this transaction commits.
    await conn.execute(
        "INSERT INTO outbox (topic, payload, dedupe_key) VALUES (%s, %s, %s) "
        "ON CONFLICT (dedupe_key) DO NOTHING",
        (
            "billing.charge",
            Jsonb({"invoice_id": args.invoice_id,
                   "customer_id": customer_id,
                   "amount_cents": amount_cents}),
            f"charge:invoice:{args.invoice_id}",
        ),
    )
    return {"invoice_id": args.invoice_id, "status": "charge_queued"}


@dataclass(frozen=True)
class ToolSpec:
    args_model: type[BaseModel]
    handler: Callable[[AsyncConnection, BaseModel], Awaitable[dict]]


REGISTRY: dict[str, ToolSpec] = {
    spec.args_model.__name__: spec
    for spec in (
        ToolSpec(ApplyCredit, apply_credit),
        ToolSpec(CreateInvoice, create_invoice),
        ToolSpec(QueueCharge, queue_charge),
    )
}
TOOL_SCHEMAS = [spec.args_model for spec in REGISTRY.values()]

Two design choices are worth calling out. First, the tools receive a connection instead of opening their own. That is what lets the node control the transaction boundary. Second, queue_charge never talks to Stripe. It writes a row. That single decision is the difference between a recoverable failure and a duplicate charge.

Step 3: Build the Error Handler That Talks to the Model

When a tool fails, the LLM needs to know what went wrong and what to do about it. It does not need your SQL, your table names, or a stack trace. Raw database errors leak internals and often confuse the model into worse retries.

# errors.py
from psycopg import errors as pgerr

from tools import ToolBusinessError

# Failures the model can act on. Connection-level errors are deliberately
# excluded so that LangGraph's RetryPolicy can handle them instead.
RECOVERABLE = (
    ToolBusinessError,
    pgerr.IntegrityError,
    pgerr.DataError,
    pgerr.QueryCanceled,
    pgerr.LockNotAvailable,
)


def describe_failure(exc: Exception) -> str:
    """Translate an exception into a safe, actionable message for the LLM."""
    if isinstance(exc, ToolBusinessError):
        return str(exc)
    if isinstance(exc, pgerr.UniqueViolation):
        return ("That record already exists. Do not create it again; "
                "look it up or choose a different value.")
    if isinstance(exc, pgerr.CheckViolation):
        return ("A business rule rejected this change, for example "
                "insufficient credit. Check the amounts and try again.")
    if isinstance(exc, pgerr.ForeignKeyViolation):
        return "A referenced record does not exist. Verify the IDs."
    if isinstance(exc, (pgerr.QueryCanceled, pgerr.LockNotAvailable)):
        return "The operation timed out because the data was busy. Try again."
    return "The database rejected the change. No changes were saved."

This is the heart of the "error handler" concept. It converts low-level exceptions into stable, model-friendly language. Notice that we exclude generic OperationalError from the recoverable set. A dropped connection is an infrastructure problem, and we want LangGraph's retry policy to handle it rather than asking the LLM to reason about it.

Step 4: Write the Transactional Tools Node

Now the main piece. The node has three phases: validate, execute inside a transaction, and translate the outcome into messages.

# nodes.py
import json

from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import MessagesState
from psycopg.types.json import Jsonb
from psycopg_pool import AsyncConnectionPool
from pydantic import ValidationError

from errors import RECOVERABLE, describe_failure
from tools import REGISTRY

pool = AsyncConnectionPool(conninfo="postgresql://app@localhost/app",
                           min_size=2, max_size=10, open=False)

ATOMIC = True  # all-or-nothing per batch of tool calls


class AgentState(MessagesState):
    rollback_count: int


class AtomicAbort(Exception):
    """Raised inside the transaction to force a full rollback."""

    def __init__(self, call_id: str, cause: Exception):
        super().__init__(str(cause))
        self.call_id = call_id
        self.cause = cause


async def _run_batch(conn, thread_id: str, batch: list, atomic: bool):
    results: dict[str, dict] = {}
    failures: dict[str, Exception] = {}

    for call, args in batch:
        key = f"{thread_id}:{call['id']}"
        try:
            async with conn.transaction():  # psycopg emits a SAVEPOINT here
                cur = await conn.execute(
                    "SELECT result FROM tool_executions "
                    "WHERE idempotency_key = %s", (key,))
                seen = await cur.fetchone()
                if seen:                     # replay after a crash: reuse result
                    results[call["id"]] = seen[0]
                    continue

                out = await REGISTRY[call["name"]].handler(conn, args)
                await conn.execute(
                    "INSERT INTO tool_executions "
                    "(idempotency_key, tool_name, result) VALUES (%s, %s, %s)",
                    (key, call["name"], Jsonb(out)))
                results[call["id"]] = out
        except RECOVERABLE as exc:           # savepoint already rolled back
            failures[call["id"]] = exc
            if atomic:
                raise AtomicAbort(call["id"], exc) from exc
    return results, failures

A key detail here: in psycopg 3, calling conn.transaction() while already inside a transaction block automatically creates a savepoint, and leaving the block by exception rolls back to that savepoint. You get savepoint semantics without hand-writing SAVEPOINT strings, and the driver keeps the bookkeeping straight. See the psycopg transaction docs for the details.

Also note that the idempotency ledger insert lives inside the same savepoint as the tool's writes. If the tool is rolled back, its ledger row disappears too. The two can never disagree.

Now the node itself:

def _msg(call, content: str, error: bool = False) -> ToolMessage:
    return ToolMessage(content=content, tool_call_id=call["id"],
                       name=call["name"], status="error" if error else "success")


async def transactional_tools_node(state: AgentState, config: RunnableConfig) -> dict:
    ai_msg = state["messages"][-1]
    thread_id = config["configurable"]["thread_id"]
    prior = state.get("rollback_count", 0)

    # Phase 1: validate everything before touching the database.
    batch, invalid = [], {}
    for call in ai_msg.tool_calls:
        spec = REGISTRY.get(call["name"])
        if spec is None:
            invalid[call["id"]] = f"Unknown tool {call['name']}."
            continue
        try:
            batch.append((call, spec.args_model.model_validate(call["args"])))
        except ValidationError as exc:
            first = exc.errors()[0]
            invalid[call["id"]] = (f"Invalid argument "
                                   f"{'.'.join(map(str, first['loc']))}: {first['msg']}.")

    if invalid:
        msgs = []
        for call in ai_msg.tool_calls:
            if call["id"] in invalid:
                msgs.append(_msg(call, "FAILED: " + invalid[call["id"]], True))
            else:
                msgs.append(_msg(call, "NOT APPLIED: another call in this step had "
                                       "invalid arguments. Resend the corrected batch.", True))
        return {"messages": msgs, "rollback_count": prior + 1}

    # Phase 2: one transaction, one savepoint per tool call.
    try:
        async with pool.connection() as conn:
            async with conn.transaction():                       # BEGIN
                await conn.execute("SELECT pg_advisory_xact_lock(hashtext(%s))",
                                   (thread_id,))                 # serialize per thread
                await conn.execute("SET LOCAL statement_timeout = '5s'")
                await conn.execute("SET LOCAL lock_timeout = '2s'")
                results, failures = await _run_batch(conn, thread_id, batch, ATOMIC)
                                                                 # COMMIT on exit
    except AtomicAbort as abort:                                 # full ROLLBACK happened
        msgs = []
        for call in ai_msg.tool_calls:
            if call["id"] == abort.call_id:
                msgs.append(_msg(call, "FAILED: " + describe_failure(abort.cause), True))
            else:
                msgs.append(_msg(call, "NOT APPLIED: rolled back because a related call "
                                       "failed. The database is unchanged. Retry the "
                                       "full batch with corrected arguments.", True))
        return {"messages": msgs, "rollback_count": prior + 1}

    # Phase 3: translate the committed outcome.
    msgs = []
    for call in ai_msg.tool_calls:
        cid = call["id"]
        if cid in results:
            msgs.append(_msg(call, json.dumps(results[cid])))
        else:
            msgs.append(_msg(call, "FAILED: " + describe_failure(failures[cid]), True))
    return {"messages": msgs,
            "rollback_count": prior + 1 if failures else 0}

Let us walk through the decisions that matter.

Validation happens first. If any call has invalid arguments, we never open a transaction. There is nothing to roll back because nothing happened. This is the cheapest and most common failure path, so it should be the fastest.

A per-thread advisory lock serializes execution. pg_advisory_xact_lock is released automatically at commit or rollback, so it cannot leak, and it prevents two concurrent runs of the same conversation from racing each other. It is also safe behind transaction-mode connection poolers because it is scoped to the transaction.

Timeouts are set locally. SET LOCAL limits statement_timeout and lock_timeout to the current transaction only. A runaway query from a confused agent cannot hold locks for minutes.

Every tool call gets a response. Whether it succeeded, failed, or was rolled back as collateral, the model receives a ToolMessage for each call ID. Skipping one causes provider-side validation errors on the next model call.

The rollback counter feeds the router. Successful batches reset it to zero. Failures increment it, which brings us to the graph.

Step 5: Wire the Graph With Routing and Retry Policy

# graph.py
import psycopg
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, SystemMessage
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy

from nodes import AgentState, pool, transactional_tools_node
from tools import TOOL_SCHEMAS

MAX_ROLLBACKS = 3

model = init_chat_model("anthropic:claude-sonnet-5").bind_tools(TOOL_SCHEMAS)
SYSTEM = SystemMessage(
    "You are a billing operations agent. When a tool reports a failure, read the "
    "message, fix the arguments, and resend the complete batch. Never assume a "
    "failed or rolled-back call took effect."
)


async def call_model(state: AgentState) -> dict:
    # No database connection is held while the model is thinking.
    reply = await model.ainvoke([SYSTEM, *state["messages"]])
    return {"messages": [reply]}


async def escalate(state: AgentState) -> dict:
    return {"messages": [AIMessage(
        content="I could not complete this safely after several attempts. "
                "No partial changes were saved, and a human will review the request.")]}


def route_after_model(state: AgentState) -> str:
    return "tools" if state["messages"][-1].tool_calls else END


def route_after_tools(state: AgentState) -> str:
    return "escalate" if state.get("rollback_count", 0) >= MAX_ROLLBACKS else "agent"


builder = StateGraph(AgentState)
builder.add_node("agent", call_model)
builder.add_node(
    "tools",
    transactional_tools_node,
    retry_policy=RetryPolicy(max_attempts=3, retry_on=psycopg.OperationalError),
)
builder.add_node("escalate", escalate)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route_after_model, ["tools", END])
builder.add_conditional_edges("tools", route_after_tools, ["agent", "escalate"])
builder.add_edge("escalate", END)


async def run(dsn: str, user_request: str, thread_id: str):
    await pool.open()
    async with AsyncPostgresSaver.from_conn_string(dsn) as checkpointer:
        await checkpointer.setup()
        graph = builder.compile(checkpointer=checkpointer)
        return await graph.ainvoke(
            {"messages": [("user", user_request)], "rollback_count": 0},
            config={"configurable": {"thread_id": thread_id}},
        )

There are three layers of error handling working together here, and it helps to be explicit about which one owns which failure:

1. Validation and business errors are caught inside the node, rolled back, converted to messages, and shown to the model. The model decides what to do.
2. Infrastructure errors such as a dropped connection propagate out of the node, where the RetryPolicy re-runs it. That is safe because the transaction either never committed or is protected by the idempotency ledger.
3. Repeated failure is caught by the router. After three consecutive rollbacks, the graph stops burning tokens and escalates to a human.

LangGraph evolves quickly, so double-check the RetryPolicy and checkpointer import paths against the version you have installed. The concepts hold across versions even when module paths shift.

Step 6: Handle External Side Effects With an Outbox Worker

A database rollback cannot recall a charge that already reached Stripe. So we never charge inside the transaction. Instead, a small worker drains the outbox after commit and uses the row's dedupe_key as the payment provider's idempotency key.

# outbox_worker.py
import asyncio
import stripe

async def drain_outbox(pool):
    while True:
        async with pool.connection() as conn:
            async with conn.transaction():
                cur = await conn.execute(
                    "SELECT id, payload, dedupe_key FROM outbox "
                    "WHERE dispatched_at IS NULL "
                    "ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 10")
                rows = await cur.fetchall()
                for row_id, payload, dedupe_key in rows:
                    await asyncio.to_thread(
                        stripe.PaymentIntent.create,
                        amount=payload["amount_cents"],
                        currency="usd",
                        customer=str(payload["customer_id"]),
                        idempotency_key=dedupe_key,   # safe to repeat
                    )
                    await conn.execute(
                        "UPDATE outbox SET dispatched_at = now() WHERE id = %s",
                        (row_id,))
        await asyncio.sleep(1)

This gives you at-least-once delivery on your side and effectively-once behavior on the provider side. If the worker crashes after Stripe accepts the request but before dispatched_at is set, the next attempt sends the same idempotency key and Stripe returns the original result instead of charging again. Read Stripe's idempotent requests documentation for the retention window and edge cases. The general technique is known as the transactional outbox pattern.

Step 7: Test the Failure Paths, Not Just the Happy Path

Rollback logic is only trustworthy if you have proven it. The test below reproduces the opening story: one valid call followed by a call that violates a constraint.

# test_rollbacks.py
import pytest
from langchain_core.messages import AIMessage

from nodes import transactional_tools_node

CONFIG = {"configurable": {"thread_id": "test-thread-1"}}


@pytest.mark.asyncio
async def test_second_call_failure_rolls_back_first(db):
    # Seed: customer 1 with 5,000 cents of credit.
    ai = AIMessage(content="", tool_calls=[
        {"id": "c1", "name": "CreateInvoice",
         "args": {"customer_id": 1, "billing_period": "2026-09-01",
                  "amount_cents": 4900}},
        {"id": "c2", "name": "ApplyCredit",
         "args": {"customer_id": 1, "amount_cents": 999_999}},  # exceeds balance
    ])

    update = await transactional_tools_node(
        {"messages": [ai], "rollback_count": 0}, CONFIG)

    assert update["rollback_count"] == 1
    assert all(m.status == "error" for m in update["messages"])
    assert await db.scalar("SELECT count(*) FROM invoices") == 0
    assert await db.scalar("SELECT credit_cents FROM customers WHERE id = 1") == 5000
    assert await db.scalar("SELECT count(*) FROM tool_executions") == 0
    assert await db.scalar("SELECT count(*) FROM outbox") == 0


@pytest.mark.asyncio
async def test_replay_does_not_double_apply(db):
    ai = AIMessage(content="", tool_calls=[
        {"id": "c1", "name": "ApplyCredit",
         "args": {"customer_id": 1, "amount_cents": 1000}}])
    state = {"messages": [ai], "rollback_count": 0}

    await transactional_tools_node(state, CONFIG)
    await transactional_tools_node(state, CONFIG)   # simulate a node replay

    assert await db.scalar("SELECT credit_cents FROM customers WHERE id = 1") == 4000

The first test proves the invoice created by call one vanishes when call two fails. The second proves that a replay, the exact scenario where a crash lands between commit and checkpoint, does not deduct credit twice. Add a third test that kills the connection mid-batch (for example, by terminating the backend with pg_terminate_backend) and asserts that the tables are unchanged.

Real-World Scenario: The Refund Workflow, End to End

Let us replay the opening story with everything in place.

1. The model emits three parallel calls: CreateInvoice, ApplyCredit, and QueueCharge.
2. Validation passes. The transaction begins and the advisory lock is taken.
3. CreateInvoice succeeds inside its savepoint. ApplyCredit violates the credit CHECK constraint.
4. psycopg rolls back to the savepoint, the node raises AtomicAbort, and the outer transaction rolls back. The invoice is gone. The ledger rows are gone. No outbox row exists.
5. The model receives three error messages: one explaining the insufficient credit, two saying "NOT APPLIED, database unchanged."
6. The model corrects the amount and resends all three calls. This time everything commits, including one outbox row.
7. The worker sees the outbox row, calls Stripe once with the deterministic idempotency key, and marks it dispatched.

Figure 3: The refund workflow end to end. The first attempt rolls back cleanly, the retry commits, and the outbox worker charges the card exactly once.

Result: one invoice, one credit deduction, one charge. No cleanup script, no support ticket.

Choosing a Strategy: Transactions, Savepoints, or Sagas

Not every workflow fits inside a single database transaction. Here is a quick way to decide.

Compensation logic (undoing a committed step with an inverse operation) is a last resort. It is harder to get right than a rollback because the world may have changed between the original action and the undo. Whenever you can keep the work inside one transaction, do.

Best Practices

• Keep transactions short and inside one node. Open after the model responds, close before control returns to it. Never hold a transaction across an LLM call.
• Validate at three levels. Pydantic for shape, application logic for business rules, and database constraints as the final guard. Assume each of the first two will occasionally miss something.
• Default to atomic batches. Only use partial commit for calls that are truly independent, and document why.
• Give every tool call a deterministic idempotency key. Deriving it from the thread ID and tool call ID works for replays. For business-level uniqueness across model retries, use natural keys enforced by unique constraints, such as customer plus billing period.
• Sanitize errors before the model sees them. Give it what it needs to fix the call and nothing else.
• Cap retries and escalate. A model stuck in a loop is an expensive bug. Three consecutive rollbacks is a reasonable default.
• Log every rollback with context. Record the thread ID, tool name, SQLSTATE, and attempt number, and alert on spikes. A rising rollback rate often signals a prompt regression or a schema change before users notice.
• Use least-privilege database roles. The agent's role should have access only to the tables and operations its tools require.

Common Mistakes to Avoid

1. Calling external APIs inside the transaction. The rollback restores your tables and leaves the charge in place. Always go through an outbox.

2. Swallowing the exception inside the savepoint block. If you catch an error inside async with conn.transaction(): and continue, the driver never sees the failure and will not roll back to the savepoint. Let the exception leave the block, then catch it outside.

3. Forgetting that the checkpoint and your data commit separately. Transactions alone do not protect you from a crash in that window. You need the idempotency ledger.

4. Returning results for only some tool calls. Providers expect a response for every call ID. Rolled-back siblings need an explicit "not applied" message.

5. Blindly retrying on any exception. Retrying a constraint violation just produces the same violation. Distinguish infrastructure failures (retry) from logical failures (tell the model).

6. Letting the model see raw database errors. They leak schema details and tend to trigger bizarre "fixes." Translate them.

7. Assuming sequences roll back. Auto-increment values consumed by a rolled-back insert are not returned, so you will see gaps in IDs. That is normal and harmless, but do not build features that assume contiguous IDs.

8. Running with no timeouts. Without statement_timeout and lock_timeout, one confused agent holding a lock can stall unrelated traffic.

🚀 Pro Tips

• Use the ledger as an audit trail. Add the thread ID, model name, and prompt version to tool_executions. When someone asks "why did the agent do that?", you can answer with a query.
• Snapshot before you risk it. For workflows that touch many rows, record before-images in an audit table inside the same transaction. Rollback handles most cases, but before-images help with forensics and dispute resolution.
• Make tool results self-describing. Return the new balance, the invoice status, and the IDs created. The model makes better follow-up decisions when it can see the state it produced.
• Prefer the atomic policy, then prove you need partial. Start strict. Relax it only where a real requirement and a test justify it.
• Chaos-test in staging. Randomly kill connections and pods during tool execution and assert your invariants (no duplicate invoices, no negative balances, one outbox row per charge) hold afterward.
• Store a "compensation needed" flag for the unrecoverable. If an external system truly cannot be made idempotent, capture the fact in the database so humans can reconcile it, instead of hoping nothing went wrong.
• Watch the isolation level. The default READ COMMITTED is fine when your updates are conditional or protected by constraints, as they are here. If you add read-then-write logic, use SELECT ... FOR UPDATE or move that transaction to SERIALIZABLE and be ready to retry on serialization failures.

📌 Key Takeaways

• Agents fail in ways scripts do not: hallucinated arguments, parallel dependent calls, mid-node crashes, and retries by design. Treat database safety as a first-class design concern.
• PostgreSQL savepoints give each tool call an independent undo point inside one outer transaction, and they are essential because any error otherwise poisons the whole transaction.
• Validate first, execute inside one short transaction, and default to atomic all-or-nothing batches for dependent calls.
• Give the LLM sanitized, actionable error messages and a response for every tool call, including calls that were rolled back as collateral.
• Never perform irreversible side effects inside the transaction. Use a transactional outbox with idempotency keys so that rollback removes the intent to charge.
• LangGraph checkpoints and your business data commit separately, so pair transactions with an idempotency ledger to make replays harmless.
• Cap consecutive rollbacks and escalate to a human rather than letting an agent loop indefinitely.

Conclusion

The gap between an impressive agent demo and a dependable production agent is mostly plumbing, and the most important plumbing is what happens when something goes wrong. Language models will keep hallucinating arguments and infrastructure will keep failing at inconvenient moments. You cannot prevent either, but you can make sure neither one leaves your database in a state that a human has to untangle at 2 a.m.

The pattern in this guide is straightforward to summarize. Validate before you write. Run tool calls inside a single short transaction with a savepoint per call. Roll back atomically when anything fails. Tell the model exactly what happened in safe language. Keep money-moving side effects in an outbox with idempotency keys. Record every applied call in a ledger so replays are harmless. Cap the retry loop and escalate.

None of this is exotic. It is classic transaction engineering applied to a new kind of caller, one that is creative, fast, and occasionally wrong. Build the safety net once, and every new tool you add inherits it.

If you are starting from an existing agent, do not rewrite everything at once. Pick your single most dangerous tool, the one closest to money or customer data, move it behind the transactional node, add the ledger and the outbox, and write the failure test first. Then expand from there.

References

• PostgreSQL Documentation: SAVEPOINT
• PostgreSQL Documentation: Explicit Locking and Advisory Locks
• Psycopg 3 Documentation: Transactions management
• LangGraph Documentation: Persistence and Checkpointers
• LangGraph on GitHub
• langgraph-checkpoint-postgres on PyPI
• Stripe API: Idempotent Requests
• Microservices.io: Transactional Outbox Pattern
• Microservices.io: Saga Pattern