A run is not a request (agent runs on plain Postgres)

Every agent demo shares a lie: that the run will finish. The user asks, the loop spins, tokens stream — and then someone refreshes the browser. Or the deploy lands. Or the laptop lid closes. In the demo, none of that happens. In production, all of it happens before lunch.

A dropped web request costs nothing; the page re-renders. A dropped agent run is different. It was minutes of paid model calls and real tool side effects, and it is not reproducible — the retry will not think the same thoughts. That asymmetry is the whole argument.

The fix is one move: a run is not a request — it’s a row. The moment a turn becomes a first-class object in the database (enqueued, claimed, leased, journaled), every reliability property in this article falls out of that one decision. The industry has quietly converged on the same move — runs became first-class queued objects in every serious agent platform — and the ladder below is why. Each rung is a production failure, the mechanism that answers it, and the canonical form that mechanism echoes. All of it on plain Postgres: no Redis, no RabbitMQ, no workflow engine. One less system is the point.

(This piece is a deep dive on one layer of a production agent harness — the map of all thirteen systems is in Building a production-ready AI agent.)

A run is not a request

Point the agent loop at an HTTP handler and you have bolted a minutes-long computation onto a connection built for milliseconds. The loop runs inside the request — when the request ends, the loop ends with it. And requests end for reasons that have nothing to do with the work. The reader refreshes the browser. The deploy lands and drains the pod. The laptop lid closes on the train. Each one severs the socket mid-thought, and with it goes minutes of already-billed model calls and tool side effects that already touched the world. Worse than losing the work: nothing records that the work ever happened. No terminal state, no audit row. The run didn’t fail — it stopped existing.

So don’t do the work inside the request. Record it, then answer. POST /runs does three cheap things and returns: validate the input, INSERT one agent_run row — status='queued', plus a snapshot of the request — and reply 202 {run_id}. Not 200; 202, accepted. Nothing has run yet, but a durable intent to run now exists. The client takes the run_id and either polls it or tails its event stream — a stream you can drop and rejoin, a few rungs down. Either way the connection is disposable: it can drop, reconnect, or land on a different pod next request, and the run will not notice. Its life is the row’s life, not the socket’s. And that row has a lifecycle:

Here is what “record it” looks like when your queue and your database are the same Postgres:

async def enqueue_run(db, conversation_id, user_id, content):
    message = Message(conversation_id=conversation_id, role="user",
                      content=content, status="queued")
    db.add(message); await db.flush()

    run = AgentRun(conversation_id=conversation_id, user_id=user_id,
                   user_message_id=message.id, status="queued")
    db.add(run); await db.flush()

    db.add(RunEvent(run_id=run.id, seq=1, type="user_message_created",
                    payload={"message_id": str(message.id)}))
    db.add(RunEvent(run_id=run.id, seq=2, type="run_queued",
                    payload={"run_id": str(run.id)}))
    await db.exec(select(func.pg_notify("run_enqueued", str(conversation_id))))

    await db.commit()   # message + run + journal + wake: one atomic unit
    return run          # -> 202 {"run_id": run.id, "status": "queued"}

Read the last two lines. The user’s message, the agent_run, its first two run_event rows, and the wake signal all commit in one transaction — together, or not at all. There is no instant when the run exists without its message, and no queued job pointing at domain rows that were never written. With a separate broker — Redis, SQS, RabbitMQ — that’s two systems to keep in step: write the rows to Postgres, publish the job to the broker, and hope the process survives the gap between them. The standard patch for that gap is the outbox pattern — a relay table and a poller whose whole job is to make two writes look like one. When the broker is a Postgres table, the gap never opens. The dual-write problem isn’t solved here; it’s unrepresentable. The INSERTs and the pg_notify ride one transaction because they ride one connection — the queue cannot disagree with the data, because the queue is the data.

The canonical form. This isn’t a local trick; it’s where the industry landed. OpenAI made the run a first-class, queued object — the same kind of status vocabulary (queued, in_progress, terminal states), with background execution you poll or stream against by run id, the connection deliberately separated from the work. LangGraph ships background runs the same way: start one, get an id back, poll or attach a stream. Different stacks, one conclusion — the durable run object is the primitive, and every reliability property below hangs off it. Those properties are the rest of this ladder, one failure at a time.

Two workers, one job

The row from rung 1 is sitting in agent_run with status = 'queued', and the run_enqueued notify has already fired. Something has to pick it up — and in production that something is plural. You run more than one worker pod, because one pod is both a single point of failure and a throughput ceiling. So two pods wake on the same notify and reach for the same table in the same instant.

Write the obvious code and you get a race. SELECT the oldest queued row, then UPDATE it to running: under one worker, fine; under two, both selects return row 101 before either update lands, both flip it to running, and the run executes twice. Double the token spend, double the tool side effects, two emails to the same customer — the un-reproducible retry from rung 1, except now you caused it. The reflex fix is SELECT … FOR UPDATE: worker A locks row 101, worker B blocks until the lock clears. Correct — but every idle worker that polls now piles up behind the head row, waiting on a lock instead of working. You traded a double-claim for a convoy.

The way out is to stop treating the claim as two steps. It is one statement:

UPDATE agent_run SET
    status            = 'running',
    claim_token       = gen_random_uuid(),
    claim_expires_at  = now() + make_interval(secs => :lease_s),   -- 30s
    worker_id         = :worker_id,
    started_at        = coalesce(started_at, now()),
    last_heartbeat_at = now()
WHERE id IN (
    SELECT id FROM agent_run
    WHERE status = 'queued'
    ORDER BY created_at
    LIMIT 1
    FOR UPDATE SKIP LOCKED      -- contended rows turn invisible, not blocking
)
RETURNING id;

Read it inside out. The inner SELECT takes the oldest queued row and locks it FOR UPDATE. SKIP LOCKED is the hinge: a row another worker already holds is not waited on — it is invisible, passed over as if it weren’t in the table, so worker B’s select slides past 101 and returns 102. The outer UPDATE flips that locked row to running and stamps the claim columns — worker_id, the lease, the token — in the same statement that found it, and RETURNING id hands it straight to the worker. No separate queue table, no ack protocol, no broker handshake: the claim is a lock and an index.

One of those stamped columns is worth flagging now: the UUID written into claim_token is the fence every later write must present — rung 4 is about why.

Claiming a row is only half the job; the worker still has to run it. A pod with every model slot already busy can win a claim anyway, flip a row to running, and then let it sit there executing nothing — alive on paper, stalled in fact. The fix is ordering: acquire-before-pull. Take the concurrency semaphore first, and only reach for a row once you already hold a slot to run it in.

async def run_forever(self):
    while True:
        await self.sem.acquire()            # backpressure BEFORE the claim
        try:
            async with self.session() as db:
                row = await self.claim_next(db)
        except Exception:
            self.sem.release(); await self.idle_wait(); continue
        if row is None:                     # queue empty
            self.sem.release(); await self.idle_wait(); continue
        task = asyncio.create_task(self.run_and_release(row))
        self.in_flight.add(task)            # tracked for the drain (rung 3)
        task.add_done_callback(self.in_flight.discard)

The semaphore gates the claim, not the execution. A saturated pod blocks at sem.acquire() and never touches the table, so it cannot strand a row it has no capacity to advance. Overload stops hiding inside claimed-but-stalled rows and surfaces where you can actually watch it — as queue depth, rows still queued, waiting for a pod with a free slot.

The canonical form. SKIP LOCKED shipped in PostgreSQL 9.5, with concurrent work queues named as the motivating use case. It is not an exotic corner of the engine: it is the claim primitive under Rails 8’s default queue, Solid Queue, and under the mature Postgres-backed job libraries — pg-boss, graphile-worker, GoodJob, River. Different languages, different schemas, the same two keywords carrying the load. When the queue is a Postgres table, “claim one job, exactly once, without stalling the fleet” is a single UPDATE — the database already knows how to run it.

Workers die

The claim succeeded. A worker holds the row, flipped it to running, and is two tool calls into the turn — when its pod OOMs, or a deploy lands and the pod takes a SIGKILL. The Python process vanishes mid-thought; the row does not. It still says running, and left alone it says running forever — a tombstone for a worker that no longer exists. Nothing is coming back for it: no one else will claim a row that already looks alive and owned.

That stranded row is worse than one lost run. A conversation runs one turn at a time (rung 7), so a running row that never resolves wedges every later turn behind it. And you can’t ask whether the pod is alive — there is no cross-pod process registry. Pods share a database, not a process table; they can’t see each other’s processes. Whether that worker still runs has to become a question the row can answer.

Three columns and a background loop do it. The lease came first: every claim in rung 2 already stamped claim_expires_at = now() + 30s — the worker’s promise to finish or check in by then. The heartbeat keeps it: every 10s, while the turn runs, the worker rolls the lease forward:

UPDATE agent_run SET
    last_heartbeat_at = now(),
    claim_expires_at  = now() + make_interval(secs => :lease_s)
WHERE id = :run_id
  AND claim_token = :claim_token;  -- a stale worker's heartbeat updates 0 rows

Note the fence in the WHERE: claim_token = :claim_token. A worker whose claim was superseded updates zero rows and learns it is stale — rung 4’s subject. A healthy worker renews its lease with room to spare; only a dead one lets it lapse.

The reaper waits for that lapse. It is a loop on every pod, not a singleton — no reaper node to lose — sweeping for rows where status = 'running' AND claim_expires_at < now() - grace and finalizing each failed with a terminal run_event. The grace window — 15 to 60 seconds past the lease — is slack, so a worker in a GC pause isn’t reaped mid-turn. The same sweep runs once at boot, the reconciler: a full-cluster restart stops every heartbeat at once with no loop left to notice, so each booting pod clears the orphans the outage left.

Crashes are one thing; deploys are constant, and they are not failures — reaping a healthy run because a rollout recycled its pod is self-inflicted loss. A deploy deserves grace: a chance to finish the work in flight. That is a two-phase drain, fired by the SIGTERM the orchestrator sends ahead of its SIGKILL deadline:

async def stop(self, drain_timeout_s):
    self.claim_loop.cancel()                 # phase 1: stop claiming (ms)
    with contextlib.suppress(asyncio.CancelledError):
        await self.claim_loop

    done, pending = await asyncio.wait(      # phase 2: drain in-flight work
        self.dispatcher.in_flight, timeout=drain_timeout_s)
    for task in pending:                     # budget spent: cancel the rest;
        task.cancel()                        # the reaper finalizes them

Phase one cancels the claim loop. It returns in milliseconds, because that loop spends its life parked — on the semaphore or in the idle wait — so tearing it down claims nothing new. Phase two is the point: await the in-flight tasks against a drain budget and let live turns finish. Whatever outlives the budget is cancelled and left to the reaper. Skip phase two and you get the textbook broken shutdown — the budget spent cancelling a sleeping loop while the real work runs unwaited, until SIGKILL kills it mid-turn.

A reaped run and a drained-but-unfinished run land in the same state: failed, not queued. An orphaned run fails; nothing auto-requeues. An agent run is not an idempotent job: it sends messages, writes rows, calls tools, and when it dies it has already done some unknowable prefix of that. Re-running half of one is how you send the email twice. So the default is at-most-once, never at-least-once — mark the orphan failed, tell the user, let a human decide. Retry isn’t free infrastructure from the broker; it is a product feature with idempotency requirements, not a queue default.

The canonical form. The lease is the visibility timeout, rebuilt on two timestamp columns. SQS hands a worker a message and hides it for a visibility-timeout window; check in before it closes, or the message reappears. Pub/Sub calls the same clock an ack deadline. Rebuild it as claim_expires_at and last_heartbeat_at and you have the mechanism exactly — visible in a plain SELECT instead of buried in a managed broker. The retry stance has a lineage too, from durable execution: Temporal retries only what you declare safe to retry. Same instinct, one layer down.

Dead, or just slow?

Rung 2 stamped a claim_token and promised its reason to rung 4. Rung 3 built the reaper on a lease and deferred the same debt — why every write drags that token through its WHERE. Both come due on the one question the reaper cannot answer: is a silent worker dead, or just slow?

It cannot tell. A stop-the-world GC pause, an event loop blocked on a synchronous call, a network partition between the pod and Postgres — from the database’s side, every one is indistinguishable from a worker that died mid-turn. All the reaper ever sees is a lease that lapsed. It has to act on that — an unreaped lease is a slow leak of wedged conversations — but acting on half the picture breeds zombies.

Two of them. In the first, the reaper marks a stalled run failed; then the “dead” worker wakes from its pause and finishes the turn, writing completed over the reaper’s failed. The journal now holds two endings for one run. In the second, the reaper kills a worker that was never dead — it went quiet for a beat and comes back to find its own run reaped out from under it.

The first zombie dies by fencing. Every write a worker makes to its run is guarded — not WHERE id = :run_id, but WHERE id = :run_id AND claim_token = :claim_token AND status = 'running'. The token proves the writer still holds the claim; the status proves the run has not already ended. Whoever writes a terminal state first wins, and terminal states are sticky — nothing moves a row out of completed or failed. So the revived worker’s completed finds status = 'failed', matches zero rows, and disappears. That zero rowcount is the notice: someone already ended this run — stand down. And the fence is not only on the final write. Every journal append carries it too (rung 6 walks the append path), so a fenced-out worker cannot even narrate into a run that has stopped being its own.

The second zombie is subtler, and worth telling as it happened. The heartbeat renews the lease — it pushes claim_expires_at forward — but it deliberately does not rotate the token. Rotating on every beat would fence out the worker’s own in-flight writes, already composed against the current token. So the token holds steady for the life of the claim, and that steadiness is the hole.

Here is the sequence, in production. A worker stalls on a long GC pause. Its lease expires. The reaper’s sweep runs, reads the orphan list, and finds the row — expired, running, holding a token — and builds its reap. In that same window the pause ends, the worker comes back, and its next heartbeat lands, rolling the lease forward, healthy again. Then the reap fires. It presents a token that still matches, against a status still running. The heartbeat beat it by a tenth of a second. Everything the reaper read is still true in its WHERE — and with nothing more, it kills a run that just proved itself alive.

The fix is to stop trusting the read. The staleness the reaper decided on — this lease was expired when I looked — has to be re-tested at the instant of the write, inside the same atomic statement, against the row as it is now. The reap carries its own liveness clause: the lease must still be expired now. A recovered worker’s heartbeat has already pushed claim_expires_at into the future, so the clause is false, the UPDATE touches zero rows, and the reap quietly no-ops. The lesson outlives this one query: check liveness in the write, not before it.

async def write_terminal_if_claimed(db, *, run_id, claim_token, status,
                                    error=None, require_expired_grace_s=None):
    conditions = [
        AgentRun.id == run_id,
        AgentRun.claim_token == claim_token,   # the fence
        AgentRun.status == "running",          # terminal states are sticky
    ]
    if require_expired_grace_s is not None:    # reaper path: still dead *now*?
        cutoff = now() - timedelta(seconds=require_expired_grace_s)
        conditions.append(AgentRun.claim_expires_at < cutoff)

    stmt = (update(AgentRun).where(*conditions)
            .values(status=status, error=error, finished_at=now())
            .returning(AgentRun.id))
    applied = (await db.exec(stmt)).scalar() is not None
    await db.commit()
    return applied   # False -> someone else already ended this run; stand down

The canonical form. This is Kleppmann’s fencing token — the argument at the heart of the locks-and-leases chapter in Designing Data-Intensive Applications, and of his Redlock critique: a lease alone cannot make a distributed decision safe, because the holder can pause past its expiry and act on a claim it no longer owns. The cure is a token the resource checks on every write — one a superseded holder cannot forge. Our claim_token is that token; Postgres is that resource. This is the piece most homegrown database queues skip: they lease, they heartbeat, they reap, and it all works right up until the first long GC pause writes a second ending into a run. The fence is what makes the lease safe.

Waking workers without hammering the database

The dispatcher loop from rung 2 works the instant it holds a row; between rows it calls idle_wait(), twice — once when a claim errors, once when the queue is empty. The simplest body is a poll: sleep the 10s tick, look again. It works, and it is slow in precisely the wrong place. A run enqueued a moment after a poll goes untouched until the next one — half a tick later on average, so a 10s tick staples ~5s of median latency to the front of every chat turn, before a single token streams.

Cranking the tick down only moves the pain. Every pod runs a dispatcher and every dispatcher polls, so a 1s tick multiplies the empty-queue load by pods × dispatchers — a whole fleet asking an idle table “anything yet?” in a tight loop — and 1s still isn’t instant. You cannot poll your way to low latency without hammering the database.

So stop polling for the answer and let Postgres deliver it. LISTEN/NOTIFY is a lossy wake-hint: a worker parks on a named channel, and a NOTIFY to that channel wakes it. The signal is already in place — the wake signal has ridden in the enqueue transaction since rung 1, where pg_notify('run_enqueued', …) commits in the same atomic unit as the row. Because NOTIFY is transactional it fires on commit, so a rolled-back enqueue wakes nobody. The idle wait blocks on that channel and keeps the tick as its timeout:

async def idle_wait(self):
    if self.hub is None:                     # hub down -> plain poll
        await asyncio.sleep(self.tick_s)     # 10s: slower, never incorrect
        return
    with contextlib.suppress(TimeoutError):  # tick stays as the fallback
        async with self.hub.subscribe("run_enqueued") as queue:
            await asyncio.wait_for(queue.get(), timeout=self.tick_s)

A freshly enqueued run now wakes a parked dispatcher in milliseconds; the tick stays only as the fallback ceiling on a missed signal.

One LISTEN connection cannot be handed out to every task that wants to wait on a channel. The hub is the answer: one dedicated LISTEN connection per pod, multiplexing every channel the process cares about — dispatcher wakes, the stream tails of rung 6, the cancel watchers of rung 8 — out to in-process subscribers, instead of a Postgres connection per subscriber. A single supervisor task owns every listener add and remove, so a reconnect can’t race a subscribe. The subscriber queues are bounded and drop on overflow rather than block — which is safe, because a dropped payload only ever means “something changed”, and a full queue already implies a wake is pending. The consumer re-reads the rows regardless.

That tolerance is the contract the whole design rests on, and it comes down to one line: signals are hints; rows are truth. Every NOTIFY here has a row behind it — a status, a cancel_requested flag, a journal row — and consumers trust the row, never the signal. Lose a hint and you lose a little latency, never correctness. Take the hub down entirely and every mechanism falls back to its poll interval: the dispatcher to its 10s tick, the tails and watchers to theirs — slower, never wrong. A dropped NOTIFY isn’t a bug to fix; it is the design doing what it was built to do.

The canonical form. This is exactly how the mature Postgres-backed queues use NOTIFY: pg-boss and graphile-worker both layer it over a poll they never drop — a low-latency wake where the poll is what lets a missed notify cost nothing. One hardening note before a connection pooler enters the picture: LISTEN needs a direct, session-lifetime connection, and it does not survive pgbouncer’s transaction pooling, where the session you LISTEN on isn’t the one you get back on the next statement. Plan the hub’s connection to sit outside the pooler before you add one.

A stream you can drop and rejoin

The obvious stream is an SSE response bolted onto the request that started the run — and it breaks three ways. Refresh the page and the stream is gone, but the run is not: it keeps burning model calls behind a blank screen. During a long tool call nothing streams, and a pipe idle for thirty seconds is one some proxy kills. A dropped socket reconnects to nothing — no record of what streamed, nothing to replay. The stream is as fragile as the request underneath it — rung 1’s mistake in a new costume.

The fix is rung 1’s move one level up: decouple the stream from the connection by writing it down. Every event a run emits is appended to a journal — run_event(run_id, seq, type, payload), the same table the enqueue transaction wrote 1–2 into — and the SSE endpoint is just a reader: pull rows after a cursor, format them onto the wire. The socket owns nothing; the table owns everything.

The journal needs one guarantee: a clean owner for seq. Ownership is single-writer-at-a-time, handed down the run’s life. The enqueue transaction wrote seq 1–2 before any worker existed; the claiming worker owns 3..n behind the fence from rung 4; the finalizers — reaper, stop — take max(seq)+1, but only once the fence has fenced the worker out. One writer at a time, so seq never collides.

Journal every event and the next problem lands: one turn emits thousands of token deltas — content_delta after content_delta — and a row apiece is write amplification, tens of thousands of tiny inserts for one answer. The deltas need not each be durable; only their sum does. So coalesce in memory and flush on a clock:

async def emit(self, event):
    if event["type"] in DELTA_TYPES:
        if self.pending and same_stream(self.pending, event):
            self.pending["delta"] += event["delta"]     # merge, don't write
        else:
            await self.flush_pending()
            self.pending = dict(event)
            self.pending_since = time.monotonic()
        if time.monotonic() - self.pending_since >= 0.250:
            await self.flush_pending()                  # 250ms flush clock
        return
    await self.flush_pending()    # structural events flush the buffer...
    await self.write([event])     # ...and write through immediately

Consecutive deltas of the same stream merge into one pending buffer. A structural event — a tool starting, the turn completing — flushes it and writes through immediately, because those are the frames a reader needs promptly; everything else rides the 250ms clock. Hundreds of rows per turn, not tens of thousands, with no loss of smoothness.

Every append is fenced like everything else here (rung 4) — the journal is not a privileged side channel that skips the ownership check:

INSERT INTO run_event (run_id, seq, type, payload)
SELECT :run_id, :seq, :type, CAST(:payload AS jsonb)
WHERE EXISTS (
    SELECT 1 FROM agent_run
    WHERE id = :run_id
      AND claim_token = :claim_token   -- zombie writers insert zero rows
      AND status = 'running'
);

The INSERT … SELECT … WHERE EXISTS writes only if the run still names this worker’s claim_token and is still running. A zombie writer — reaped, superseded, woken from a pause — carries a token that no longer matches, inserts zero rows, and finds out it is stale. And the batch is all-or-nothing: if the fence trips mid-batch, the whole flush rolls back — including the NOTIFY queued to wake the tails, transactional, so it dies with the rows it would have announced.

Reading the journal back out is the tail, and it inherits rung 5’s hub and doctrine whole:

async def tail(run_id, from_seq, hub, keepalive_s=15):
    cursor = from_seq
    async with hub.subscribe(events_channel(run_id)) as queue:  # BEFORE reading
        while True:
            events = await read_events_after(run_id, cursor)
            for e in events:
                cursor = e.seq
                yield {**e.payload, "seq": e.seq}    # seq = the resume cursor
                if e.payload["type"] == "complete":
                    return                           # first terminal wins
            if not events:
                yield {"type": "keepalive"}          # keep proxies warm
            with contextlib.suppress(TimeoutError):  # poll is the fallback
                await asyncio.wait_for(queue.get(), timeout=keepalive_s)

The tail subscribes to the hub before the catch-up read, never after: an event landing in the gap between read and subscribe reaches no one yet listening, and the tail hangs on a wake that already fired — subscribe first, and the worst case is a redundant wake, not a lost one. Every wire event carries its own seq, the resume cursor: a dropped client reconnects with ?from=N and the tail replays everything after N. The first complete ends the tail; keepalives every 15s keep a proxy from killing a quiet pipe.

The NOTIFY is only a hint here, exactly as in rung 5 — the poll is the truth. That is what the empty poll is for: when a wait times out with no new rows, the tail re-reads the run row. If the run went terminal without a journaled complete — a finalize that died after the status flip but before its last frame — the tail synthesizes the terminal frame from the row, so an interrupted finalize cannot strand a viewer. And because this state lives in Postgres, not in the pod that started the run, any pod can serve any tail — worker pod and SSE pod need not be the same machine.

One caveat keeps the journal honest: it is a delivery buffer, not history. Once a run is terminal and its viewers have caught up, its run_event rows are trimmed after 24h and the run row’s PII snapshot is nulled. Nothing is lost — the journal was never the record of what was said. The durable transcript lives in the message table. Two jobs, two tables: the journal delivers the live stream and evaporates; the message table remembers.

The canonical form. A journal with a per-consumer cursor is a per-run Kafka topic with consumer offsets — partitioned by run_id, ordered by seq, each viewer at its own position — built from one Postgres table instead of a broker. The resume cursor is the shape the platforms converged on: OpenAI’s background mode streams against a sequence_number, the Vercel AI SDK ships resumable streams, and LangGraph lets a client join a run already in flight. Different stacks, one conclusion — the stream is a table, and a cursor is how you rejoin it.

One turn at a time

A conversation is a sequence, not a set. Turn two reads what turn one wrote — the transcript, the tool results, the model’s last sentence. Run the two turns at once and that ordering dissolves. A user fires two messages back-to-back, a pod has two free slots, and two workers claim both turns of one conversation at once. Two loops append to one transcript, interleaved, each reading a history the other is still rewriting. The transcript corrupts, and no single turn misbehaved.

SKIP LOCKED (rung 2) already keeps two workers off the same row — but these are different rows, two turns of one conversation, and nothing yet couples them. The obvious coupling is a clause in the claim’s inner select: take a queued turn only if its conversation has no run already running.

-- the pre-filter (an optimization, not the guarantee)
...
AND NOT EXISTS (
    SELECT 1 FROM agent_run r2
    WHERE r2.conversation_id = agent_run.conversation_id
      AND r2.status = 'running')

-- the guarantee
CREATE UNIQUE INDEX one_running_per_conversation
    ON agent_run (conversation_id)
    WHERE status = 'running';

That pre-filter is an optimization, and alone it leaks. Under READ COMMITTED — Postgres’s default — each claimer’s SELECT reads a snapshot fixed at statement start, before the other’s UPDATE commits. So both run the NOT EXISTS check, both see no running run, and both pass. Two UPDATEs to running, one conversation — the race the clause looked like it closed.

The partial unique index closes it for real, because it lives below the snapshot: one row per conversation_id with status = 'running', no more. The first UPDATE lands; the second collides with the index, raises an IntegrityError, rolls back, and claims nothing this round — its turn stays queued for the next pass. The check filtered the easy case and spared contention; the index caught what it could not see. The lesson outlives this one table: application-level checks filter; only the schema guarantees.

One consequence needs wiring. While a turn runs, the pre-filter and the index hold that conversation’s other turns back — queued, unclaimable, by design — so when the running turn ends, something must knock on the queue, or the next turn waits out a full poll tick (rung 5). Every terminal path knocks: worker finalize, the dispatcher’s run_and_release backstop (rung 2, its terminal write when a run’s task exits unfinalized), and reaper each fire run_enqueued as they close a run, so the next turn is claimed in milliseconds, not on the tick.

The canonical form. One active run per conversation is not a local quirk; it is the shape the platforms settled on. OpenAI enforces exactly one active run per thread — start a run against a thread that already has one in progress and the API refuses it — as service policy. Here it is not application code you can forget to write. It is a partial unique index, checked by the database on every claim — nothing routes around it.

Stop means stop

“Stop” fails two ways before it works. The first: treating the closed connection as the stop. But rung 6 pried the stream off the socket — a closed tab ends the SSE and nothing more; the run behind it keeps calling the model and running up the bill for no one. Disconnect is not stop. Stop must be a request the client makes on purpose, against the run id.

The second failure hides in the stop handler. A conversation being stopped may hold one running turn and several queued behind it (rung 7). Cancel them in separate transactions — sweep the queued rows, then flag the running one — and you reopen the gap this article keeps closing. In the gap the running worker finishes and fires its terminal run_enqueued (rung 7’s wake chain); a dispatcher claims the next queued turn the sweep has not reached; and the user who pressed stop watches the conversation open a fresh turn.

You cannot stop a run from outside it. It is mid-loop on some pod with a model stream open, and the database cannot reach in to halt it. So cancel is cooperative and two-path. The truth is a cancel_requested flag on the run row. The fast path is a NOTIFY on cancel_channel(run_id) — a channel the hub already carries (rung 5) — that wakes the worker at once. The worker honors both: it reacts to the notify the moment it arrives, and its 10s heartbeat (rung 3) doubles as the missed-signal poll, re-reading the flag each beat. So a lost NOTIFY costs at most one heartbeat — 10s — of latency, never the stop itself: rung 5’s doctrine (signals are hints, rows are truth), now applied to cancel. Woken, it checkpoints, stops generating, finalizes the run cancelled, and keeps the partial text already journaled (rung 6) — a cut-off answer, not a blank one.

The handler shuts its own gap the way rung 1 shut the enqueue gap — by leaving no gap to shut.

async def stop_conversation(db, conversation_id):
    runs = await db.exec(                    # ONE transaction for the sweep
        select(AgentRun)
        .where(AgentRun.conversation_id == conversation_id,
               AgentRun.status.in_(["queued", "running"]))
        .with_for_update())

    for run in runs:
        if run.status == "queued":           # never started: cancel outright,
            await cancel_queued(db, run)     # hand the draft back to the user
        else:                                # running: flag + fast-path wake;
            run.cancel_requested = True      # worker finalizes 'cancelled'
            await notify(db, cancel_channel(run.id), "cancel")

    await db.commit()  # sweep + flag land together: nothing sneaks in mid-stop

One FOR UPDATE sweep locks every active run in the conversation, queued and running together. A queued turn never started, so it is cancelled outright and its user message handed back to the composer as a draft — that turn simply never happened. The running turn takes the flag and the wake. Because sweep and flag commit in the same transaction, the worker’s terminal wake finds nothing left queued to start. Stop is one transaction, so nothing sneaks in mid-stop.

The canonical form. Cooperative cancellation is what every serious runtime does, because it is the only kind that is safe: you signal a task and let it wind down at a checkpoint instead of killing it mid-write. Python’s asyncio cancels by raising at the next await; Go threads a context.Context cancel down the call stack. And the two-path split beneath it — NOTIFY for latency, the row for truth — is rung 5’s doctrine once more: signal for latency, state for truth, aimed now at stopping a run, not starting one.

Where this design stops

Every mechanism here is tuned for chat-cadence runs — seconds to minutes long, hundreds in flight. Push past that envelope and the seams show. Knowing where they are is the point: each edge names the moment to graduate, and what to graduate to.

At-most-once is a choice, not a gap. A crash mid-run ends as a failed row and a retry button the user presses on purpose, never a silent redelivery that sends the second email (rung 3). Want automatic retries? You owe idempotency keys on every side effect first — earn those and at-least-once is safe to turn on.

Enqueue is not idempotent yet. A client that double-POSTs /runs gets two rows and two runs. The evolution is a client-supplied Idempotency-Key header, stored unique on agent_run, so the retry returns the first run instead of minting a second. The uniqueness, as everywhere here, lives in the schema.

FIFO has no fairness. The claim orders by created_at, so one hot conversation or busy tenant can flood the queue and head-of-line block everyone behind it. The evolution is a weighted claim — ORDER BY on tenant debt instead of arrival — and because the claim is one statement (rung 2), that statement is the only place fairness changes.

LISTEN and connection pooling fight. LISTEN needs a direct, session-lifetime connection; pgbouncer’s transaction pooling hands back a different session on the next statement and breaks it. Keep the hub’s connection out of the pooler, or lose the fast path — the system survives (rung 5), but latency reverts to the poll tick.

The throughput ceiling is real. LIMIT 1 claims and per-batch journal commits are right for chat cadence, wrong for thousands of jobs a second. A number makes it concrete: a 250ms flush is roughly four journal commits a second per active run, and that scales with concurrency. Fine for hundreds of runs; a wall for a firehose.

Clocks are the last soft spot. Leases compare an app server’s clock against itself across pods, so NTP skew between pods enters the comparison — but that skew is far below the 15–60s grace, so the lease holds. The purist fix makes the database’s now() the single time authority and compares nothing across pods.

None of these is a bug; each is an edge with a name and an exit. Outgrow one and you patch it in place. Outgrow two or more at once, and that’s the signal to graduate — to a dedicated broker or a durable-execution engine — and the row-per-run model transfers there intact.

The canonical forms, in one table

Every rung closed on its canonical form — each mechanism is the shape the rest of the industry converged on, not a local trick. The whole ladder in one view:

Problem	Mechanism	Canonical form
Run outlives request	Run row + 202, enqueue in one tx	First-class run objects (OpenAI, LangGraph)
Two workers, one job	FOR UPDATE SKIP LOCKED claim	Postgres queue lineage (PG 9.5 →, Solid Queue)
Workers die	Lease + heartbeat + reaper	SQS visibility timeout
Dead vs. slow	Claim-token fence + grace re-check in the write	Kleppmann’s fencing tokens
Wake latency	Lossy NOTIFY, poll fallback	Signals are hints, rows are truth
Dropped streams	Seq journal + resume cursor, any-pod tail	Kafka offsets; resumable streams
Turn ordering	Partial unique index WHERE running	Invariants live in the schema
Stop	Flag + NOTIFY + atomic sweep	Cooperative cancellation

What transfers

Strip the schema names and the constants, and the rules port to any runtime, language, or database that can lock a row:

• A run is a row. Give the work a durable identity and the connection turns disposable.
• Claim in one statement. One atomic UPDATE, never select-then-update — the double-claim can’t exist.
• Lease for liveness, fence for correctness. The lease catches the dead worker; the token stops the not-quite-dead one.
• Check liveness in the write, not before it. A snapshot is stale the instant you act on it.
• Signals are hints; rows are truth. Lose a hint, lose a little latency — never correctness.
• Streams are tables with cursors. Write the events down and a viewer is a position; any pod can serve it.
• Invariants belong in the schema. Application-level checks filter; only the schema guarantees.
• Stop is a state, not a disconnect. A closed socket is not intent — make cancellation an explicit, durable request.

The harness has thirteen systems; this piece drilled into one. The map is Building a production-ready AI agent; the context layer gets its own deep dive in An unlimited context window (without an unlimited model).