Idempotent Consumers: Delivering Agent Events Exactly Once

Every message broker promises at-least-once delivery, and every team eventually asks for exactly-once. Both statements describe the same reality from opposite ends: your event consumer will see the same agent event twice, and the only component that can make that harmless is the consumer itself. For agent systems the stakes are higher than for ordinary event processing — a duplicated event can mean a second LLM run costing real money, a duplicate email to a customer, or worse, a second payment.

This post is about building consumers that are idempotent by construction: choosing deduplication keys that survive redelivery, placing the dedup where it can share a transaction with the side effect, handling the ways agent frameworks re-execute work behind your back, and testing it all by deliberately redelivering events.

Why “exactly once” is a lie at the transport level

The fundamental window is this:

consume event ──► do side effect ──► ack
                     ▲
                     │ crash here, after the effect but before the ack
                     ▼
              redelivery → side effect runs twice

No broker can close this window for you, because the side effect and the ack live in different systems (your database, the customer’s inbox, a payment API) with no shared transaction. Kafka can give you exactly-once within its own commits; SQS gives you at-least-once with a visibility timeout; Redis streams give you consumer groups with pending-entry recovery. All of them re-deliver after a crash, a slow consumer exceeding its visibility window, or a rebalance. The transport layer delivers at least once, always. The consumer’s job is to make “at least once” observationally equivalent to “exactly once.”

Choosing deduplication keys

A dedup key must be stable across redeliveries of the same logical event and distinct across different ones. Three families, in increasing robustness:

Key type	Example	Weakness
Transport-assigned ID	SQS MessageId, Kafka offset	New ID on redelivery in some paths; not stable across sources
Source-assigned event ID	evt_3f8a... (Stripe, GitHub)	Best when the source has one; vanishes for internal producers
Natural key	(order_id, action) or (source, seq)	Survives anything, but needs domain discipline to be unique

For internal producers that lack an event ID, (source, seq) pairs — a monotonic per-producer sequence number — give you both dedup and a way to reason about ordering. Natural keys like ("billing", order_id, "refund_issued") are the most durable: they encode what happened, not which copy of the message you received. Whatever you pick, make it a primary key or unique constraint somewhere; an in-memory set() of recent IDs is a cache, not a dedup strategy, and it forgets exactly when a rebalance is most likely to redeliver.

Where to dedupe: the inbox table

The workhorse pattern is the transactional inbox: a table in the same database as your side effects, with the dedup key as a unique constraint. Insert first, process only if the insert actually created a row:

-- schema.sql
CREATE TABLE agent_event_inbox (
    event_id    text PRIMARY KEY,          -- dedup key: source event id or natural key
    source      text NOT NULL,
    seq         bigint,                    -- optional (source, seq) ordering info
    payload     jsonb NOT NULL,
    received_at timestamptz NOT NULL DEFAULT now(),
    processed_at timestamptz,              -- NULL until side effects commit
    last_error  text
);

-- Partial index keeps the hot "still working" set small.
CREATE INDEX inbox_inflight ON agent_event_inbox (received_at)
    WHERE processed_at IS NULL;

# consumer.py
import json
import asyncpg

async def handle_event(conn: asyncpg.Pool, event: dict) -> bool:
    """Returns True if this delivery was processed, False if it was a duplicate."""
    event_id = event["id"]
    async with conn.transaction():
        inserted = await conn.fetchval(
            """
            INSERT INTO agent_event_inbox (event_id, source, seq, payload)
            VALUES ($1, $2, $3, $4)
            ON CONFLICT (event_id) DO NOTHING
            RETURNING event_id
            """,
            event_id, event["source"], event.get("seq"), json.dumps(event),
        )
        if inserted is None:
            return False  # duplicate delivery: someone else owns it (or already did)

        # Same transaction: side effect and "processed" commit atomically.
        await run_agent_side_effects(conn, event)
        await conn.execute(
            "UPDATE agent_event_inbox SET processed_at = now() WHERE event_id = $1",
            event_id,
        )
    return True

Two properties make this work. First, the insert and the side effects share a transaction, so the crash window collapses: either the inbox row and the side effects both exist, or neither does, and a redelivery retries cleanly. Second, the duplicate path is a single index probe — no lock held during the agent run, so redeliveries of hot events are cheap.

If the side effect lives in a different system (a payment API, a SaaS mailbox), you cannot share the transaction, and you need the second pattern: the transactional outbox on the producer side and fencing keys on the consumer side. The outbox writes the event and the state change in one local transaction; a relay publishes it to the queue, so an event cannot exist in your state but not the queue, or vice versa.

Checkpointing and replay: the hidden second producer

Agent frameworks add a duplicate-source that catches teams by surprise. LangGraph’s persistence layer — checkpointers that persist a thread’s graph state, with PostgresSaver for production — is explicitly built for recovery, time travel, and fault tolerance. Those are features and duplicate-event generators:

• Crash recovery re-executes from the last checkpoint. If a node ran, its tool calls happened, and then the process died before the next checkpoint, the node runs again on recovery — replaying its side effects.
• Time travel and forks re-run history. Debugging by replaying or forking a thread is normal practice; every re-run walks over nodes that already sent emails or created rows in a previous life.
• Durable execution engines re-run activities. The durable-execution integrations (Temporal, DBOS, Prefect, and friends) resume by re-running work items after a restart.

The fix is side-effect fencing: every side effect carries an idempotency key derived from stable identity in the graph, not from “this execution”. f"{thread_id}:{node_name}:{task_path}" — or the checkpoint NPU/task ID the framework hands you — passed as an Idempotency-Key header to every downstream API (Stripe-style), or as a natural key in your own tables. The first execution creates the charge; the replay’s key matches, and the API returns the original result instead of a second charge. Checkpointing gives you cheap recovery; fencing makes recovery safe to use.

Ordering caveats

Dedup solves duplication, not ordering. At-least-once delivery permits seq=7 to arrive before seq=5. If your consumer is order-sensitive, track (source, seq) and either buffer-and-wait for gaps (bounded, with a timeout that gives up and moves on) or apply monotonic guards: UPDATE accounts SET balance = $1 WHERE seq < $2. The guard style is safer under replays but requires idempotent values, not just idempotent inserts — you overwrite with the newest state rather than applying a delta twice. Note the trap: “skip if seq <= last_applied” silently drops events when a gap was never filled, so alert on gaps rather than just skipping.

Window-based dedupe vs permanent ledger

The inbox table is a permanent ledger, and ledgers grow. The cheap alternative is a sliding window: SETNX in Redis with a TTL, or a bloom filter of recent event IDs. The trade-off is honest:

• Window (hours to days): cheap, bounded memory, and sufficient if you can prove redelivery can never happen after the window expires. For webhooks with bounded retry schedules (Stripe retries for days; some providers longer) the window must outlast the schedule, and it buys you nothing during it.
• Permanent ledger: the dedup key doubles as an audit log — you can prove an event was processed exactly once, replay it deliberately, and answer “did this ever run?” during an incident. This is usually the right call when the side effects cost money.

Pick retention deliberately: a common compromise is a permanent ledger for money-moving event types and a TTL window (longer than the source’s max retry horizon) for the rest, pruned by a cron job.

Testing strategy: redeliver on purpose

Idempotency that is not tested does not exist, because the failure only ever happens at 3 a.m. during a rebalance. Three tests every consumer should have:

1. Double delivery. Feed the identical event through your consumer twice and assert exactly one side effect: one inbox row, one email, one charge, one agent run. This is a five-line test and catches the majority of bugs (a forgotten unique constraint, an effect outside the transaction).
2. Crash injection. Kill the consumer between the side effect and the ack — or, better, make the side effect itself crash the test process on the first call and assert the second delivery completes correctly. This verifies the transactional boundary rather than assuming it.
3. Replay under concurrency. Deliver the same event to two consumers simultaneously and assert the ON CONFLICT path held: one processed, one returned False without blocking.

async def test_duplicate_delivery_is_harmless(pool):
    event = {"id": "evt_123", "source": "billing", "seq": 41,
             "action": "refund", "order_id": 88}

    first = await handle_event(pool, event)
    second = await handle_event(pool, event)   # simulate redelivery after crash

    assert first is True
    assert second is False                     # deduped, not re-run
    rows = await pool.fetch(
        "SELECT count(*) FROM refunds WHERE order_id = 88")
    assert rows[0][0] == 1                     # one side effect, provably

Wire the double-delivery test into CI next to your unit tests; it costs milliseconds and pays for itself the first time a deploy takes down a consumer mid-run.

Checklist

• Accept at-least-once delivery as a fact; make the consumer idempotent rather than chasing transport guarantees.
• Use source event IDs or natural keys as dedup keys, enforced by a unique constraint — never an in-memory set.
• Dedupe with a Postgres inbox table so the dedup insert and side effects commit in one transaction; use an outbox for producers.
• Fence side effects with idempotency keys derived from stable graph identity (thread + node), because checkpoint recovery and time travel re-run work by design.
• Handle ordering explicitly with (source, seq) guards that are idempotent in values, and alert on gaps.
• Choose permanent ledger vs TTL window per event class, with retention that outlasts every source’s retry schedule.
• Test by redelivering the same event twice, crashing between effect and ack, and racing two consumers.

Further reading

• LangGraph documentation — persistence, checkpointers, and time travel (the former langchain-ai.github.io/langgraph URL now redirects here)
• LangGraph persistence guide — checkpointers vs stores, production PostgresSaver
• LangChain documentation
• Pydantic AI — durable execution capabilities — surviving restarts on Temporal, DBOS, and other engines