Typing ahead (message queue)

A person can keep typing while a turn is running, the way they do in Claude Code or a chat app. The next message waits in the thread's queue on the server — it survives a reload, a closed tab and a second device — and runs as soon as the turn ahead of it settles, exactly as if it had been sent then.

Nothing to wireCopy Anchor Link

const chat = useAgentChat({ threadId });

<MessageList messages={chat.messages} status={chat.status} queue={chat.queue} />
<ChatInput onSubmit={(text) => chat.sendMessage({ text })} />

sendMessage and composer.submit() queue on their own while a turn runs, and the chat attaches to the queued turn the moment it starts — its user message appears, its answer streams. queue on <MessageList> draws what is waiting (and a Resume button while the queue is paused).

On the server nothing changes either: both runners and every bundled store (in-memory, Drizzle, MikroORM) support it. The SQL stores add an agent_queued_message table and an agent_thread.queue_pause column on boot.

Your own markupCopy Anchor Link

const { transcript, queue } = chat;

{transcript.items.map((item) => <Message key={item.id} item={item} />)}

{transcript.queued.map((item) => (
  <div key={item.id} data-state={item.state /* 'sending' | 'queued' | 'paused' */}>
    {item.text}
    {item.isNext ? <small>next</small> : null}
    {item.remove.available ? <button onClick={item.remove.run}>Remove</button> : null}
  </div>
))}

{queue.paused ? (
  <p>
    Paused ({queue.paused.reason}{queue.paused.message ? `: ${queue.paused.message}` : ''})
    <button onClick={() => queue.resume()}>Resume</button>
  </p>
) : null}

Queue, interrupt or blockCopy Anchor Link

useAgentChat({ whileRunning: 'queue' });     // default — wait for the running turn
useAgentChat({ whileRunning: 'interrupt' }); // cancel the running turn, run this next
useAgentChat({ whileRunning: 'block' });     // refuse — composer.blockedBy === 'busy'

A backend without enqueueMessage behaves as 'block', so an older server keeps working.

whileRunning is the chat's default. One send can answer differently — which is what a composer with a plain Send (queue) and a Send now (interrupt) button needs:

<button onClick={() => chat.composer.submit()}>Send</button>
<button onClick={() => chat.composer.submit({ mode: 'interrupt' })}>Send now</button>

// the same on sendMessage, for a composer of your own
await chat.sendMessage({ text }, { mode: 'interrupt' });

mode is 'queue' or 'interrupt', overrides whileRunning for that send ('block' included), and only matters while a turn is running — with nothing running the message is simply sent. composer.submit({ mode }) clears the draft and its files like any other submit.

Send a waiting message nowCopy Anchor Link

A message that is already in the queue can cut in too:

{transcript.queued.map((item) => (
  <div key={item.id}>
    {item.text}
    {item.state !== 'sending' && (
      <button onClick={() => chat.queue.interrupt(item.id)}>Send now</button>
    )}
  </div>
))}

chat.queue.interrupt(id) is one server call (POST <base>/queue/:messageId/interrupt): the message moves to the head marked as an interrupt, any pause is lifted, and the running turn is cancelled for it. With nothing running it starts at once. The message keeps its id and never leaves the queue.

Files on a waiting messageCopy Anchor Link

A waiting message's files have the shape a sent message's do, so one renderer draws both:

import { messageFiles } from '@dudousxd/nestjs-agent-react';

const sent = messageFiles(message);          // MessageFile[]
const waiting = chat.queue.items[0].files;   // MessageFile[] — same shape
// { url, mediaType, filename, kind, extension, mediaId }

transcript.queued[n].files carries the same fields plus isImage, so it also fits a renderer written for the transcript's file blocks. attachmentFile(attachment) turns an uploaded MessageAttachment into a MessageFile yourself; items[n].attachments is still there for a re-send. While a message waits its files count as referenced, so attachment sweeping (collectableAttachments) never offers them.

A queued message keeps what its send carried — agent, persona, model, page context and the client's uiCapabilities — and runs with them when it starts.

When the queue pausesCopy Anchor Link

The next message's author is checked against the quota as it starts; a spent budget pauses the queue (quota_exceeded) instead. A paused queue keeps every message and starts nothing until it is resumed. A plain send on an idle thread still runs at once while the queue is paused — the pause holds the queue, not the thread. A queue left waiting with nothing running (the process that would have started it went away) is started when the thread loads.

One turn per thread, across podsCopy Anchor Link

The thread's active run is the admission lock, and every write to it is a compare-and-set: a claim succeeds only on a free thread (or for the run that is handing over), a release only for the run that still holds it. So of two sends racing on one thread exactly one starts and the other queues, and a turn that already handed the thread on can never clear its successor. A queued message runs under its own id as the run id, which makes the drain idempotent; under the durable runner the decision is one journaled checkpoint and the start is a replay-safe ctx.startChild, so a replay never starts a message twice. A thread held by a run that is no longer running (a crashed process) is taken over by the next send instead of queueing behind it for ever.

A backend of your ownCopy Anchor Link

The React side only speaks the wire, so a backend that is not this library (a sandboxed runner, a different framework) serves the queue by implementing:

1. POST <base>/chat with mode: 'auto' | 'queue' | 'interrupt', answering 202 { queued: true, threadId, messageId, position, queue, runId?, interrupting? } when it queues ('queue' always answers 202).
2. Persistent per-thread FIFO storage of queued messages (text, resolved attachments, agent, persona, model, page context, UI capabilities, the actor, an interrupt flag, an order) and a per-thread pause.
3. The drain on every settle of a thread's own turn — before the terminal frame, with the table above, using compare-and-set admission.
4. A queue stream frame: { kind: 'queue', queue: { items, paused }, started?: { messageId, runId } } on every change while a turn holds the thread, and with started on a handoff.
5. GET/DELETE <base>/threads/:id/queue, POST <base>/threads/:id/queue/resume, PATCH/DELETE <base>/queue/:messageId, and queue on GET <base>/threads/:id.
6. Optionally POST <base>/queue/:messageId/interrupt for chat.queue.interrupt(id): mark the message an interrupt, move it to the head, lift the pause, cancel the running turn — answering the queue plus interrupting (the cancelled run), or runId when nothing was running and the message started. Never cancel a turn for a message that is already gone, or that is itself the running turn. Without the route, interrupt(id) rejects and everything else works.

The full contract is Message queue in the repo's docs/stream-protocol.md; the store behaviours are CHAT_QUEUE_STORE_CONTRACT in @dudousxd/nestjs-agent-testing, runnable against a store of your own. See Endpoints and Stream events.