Resumable Streams

A streamed agent run can take minutes. A flaky network, a phone that switches cells or a laptop that goes to sleep cuts the SSE connection, and by default the adapter then stops the run. With resumableStreams the run goes on without the client, every event is kept, and the client reconnects to pick the stream up exactly where it stopped. A client that wants the run stopped says so on the cancel route.

It works the same in every adapter: Express, Fastify, Hono, Koa, Tetsu and the Next.js handlers, on /agents/:name/stream and /agents/:name/resume/stream.

Turn it onCopy Anchor Link

// Express
new CogitatorServer({ app, cogitator, agents, config: { resumableStreams: true } });

// Fastify
await fastify.register(cogitatorPlugin, { cogitator, agents, resumableStreams: true });

// Hono and Koa
cogitatorApp({ cogitator, agents, resumableStreams: true });

// Tetsu
cogitatorController({ cogitator, agents, resumableStreams: true });

// Next.js (owner is required next to beforeRun, see below)
export const POST = createChatHandler(cogitator, agent, { resumableStreams: true });

true keeps the events in the memory of the process with the default limits. An object sets them:

Off by default: events carry no ids and a client that disconnects stops the run, as before.

Event idsCopy Anchor Link

Every event frame of a resumable stream carries an SSE id:

id: stream_6f1c2b7e-3a4d-4e8f-9b0a-1c2d3e4f5a6b:3
data: {"type":"text-delta","id":"txt_1","delta":"Hel"}

The id is <streamId>:<seq>. The stream id is random and cannot be guessed, the sequence number starts at 1 and grows by one with each event, so the id alone names the stream and the place in it. Heartbeat comments carry no id.

ReconnectCopy Anchor Link

Send the same route again with the id of the last event the client got, in the Last-Event-ID header:

const res = await fetch('/cogitator/agents/assistant/stream', {
  method: 'POST',
  headers: { 'Last-Event-ID': lastEventId },
});

The request starts no run and its body is not read. The answer replays the events after that id, with their ids, then follows the run live until finish (and data: [DONE]) or error. Several clients may follow the same stream at once. The lastEventId query parameter works too, for clients that cannot set headers, but it puts the id into URLs and access logs: prefer the header.

On STREAM_RESUME_GAP the client should fetch the thread (GET /threads/:id) to catch up, the message names the thread. A client that reads slowly holds at most maxEvents events, one that falls further behind catches up from the store, and gets the gap only for events the store no longer keeps. A store that lost an event in the middle is caught by its sequence numbers the same way: nothing is skipped silently.

A client on another origin sends Last-Event-ID after a CORS preflight, which must allow the header. The Express adapter allows it by default. With the CORS middleware of Fastify, Hono, Koa, Tetsu or Next.js, add it to the allowed headers: CORS_ALLOWED_HEADERS from @cogitator-ai/server-shared lists every header a client of the API sends. The cancel route also needs DELETE among the allowed methods.

CancelCopy Anchor Link

Stopping the client's fetch only disconnects it, the run goes on. To stop the run, send DELETE to the cancel route of its stream, the stream id being the part of an event id before the last ::

await fetch(`/cogitator/agents/assistant/stream/${streamId}`, { method: 'DELETE' });

It answers 204 and stops the run on whatever process runs it, its stream then ends with an error event STREAM_CANCELLED. A stream that already finished is left as it is. The cancel takes the same checks as a reconnect (404 STREAM_NOT_FOUND, 403 THREAD_FORBIDDEN). In Next.js the cancel is DELETE on the route of the handler with ?streamId=, and useCogitatorChat sends it on stop(), reload() and every other interrupt.

Who may reconnectCopy Anchor Link

A stream is bound to the caller who started it, and only that caller may reconnect to it or cancel its run:

• By default the key is every field of what the adapter's auth returned: user id, roles, permissions and metadata such as a tenant. Callers without a user id are still told apart by the rest of their identity, and a caller with no identity (no auth, or auth that returned {}) is the anonymous caller, who reaches only anonymous streams.
• owner: (auth) => key binds streams to a key of your own, for example auth?.userId when the identity carries values that change between requests, such as the expiry of a token.
• After the owner check, a reconnect and a cancel ask authorizeThread about the stream's thread, so a caller it turns away since the run started is refused.
• In Next.js a reconnect and a cancel run neither beforeRun nor deps, whose side effects would repeat. owner(req) reads the request instead, and a handler with beforeRun throws without it. See Next.js.

Detached runsCopy Anchor Link

When the last client leaves, the run goes on and its events keep going into the store. If nobody reconnects within detachedTimeoutMs, the run is stopped, so a client that never comes back does not run up cost. Every run is also stopped when until aborts: pass the signal your server stops on, and abort it before the HTTP server closes, since a close waits for the streams still open. Fastify stops them by itself as the instance starts closing (preClose), Tetsu also on its controller's until, and Next.js also on the signal beforeRun returns. Such a stream ends with an error event STREAM_ABORTED, which a late reconnect still gets.

The clients of the process that runs a stream get every event straight from it, the last one too, also when the store fails. A store that cannot keep an event stops the run.

Room for streamsCopy Anchor Link

A stream whose run goes on is never evicted, since its client could no longer reconnect. When the store keeps maxStreams streams that all go on, a new resumable stream is refused with 503 STREAM_LIMIT_REACHED before anything is sent, and the client may try again once a run finished.

Several processesCopy Anchor Link

The default store lives in the memory of one process. Behind a load balancer, a reconnect or a cancel may land on a process that did not start the run. Give every process one shared store, an implementation of StreamReplayStore from @cogitator-ai/server-shared:

import type { StreamReplayStore } from '@cogitator-ai/server-shared';

const store: StreamReplayStore = {
  create: (stream) => redisCreateUnlessFull(stream),
  append: (streamId, entry) => redisAppendAndPublish(streamId, entry),
  read: (streamId, afterSeq) => redisRead(streamId, afterSeq),
  finish: (streamId) => redisFinishAndPublish(streamId),
  get: (streamId) => redisGet(streamId),
  subscribe: (streamId, subscriber) => redisSubscribe(streamId, subscriber),
  cancel: (streamId) => redisPublishCancel(streamId),
  onCancel: (streamId, listener) => redisSubscribeCancel(streamId, listener),
};

cogitatorApp({ cogitator, agents, resumableStreams: { store } });

• create answers false when the store has no room without evicting a stream that goes on.
• read answers the events after a sequence number, whether some were evicted (gap) and whether the stream finished.
• subscribe and onCancel must reach listeners on every process, which takes pub/sub (Redis PUBLISH/SUBSCRIBE or similar), and subscribe delivers events in order.
• get reports the owner, the subscribers across every process and detachedAt, when the last of them left on any process: a run checks them before it stops as detached, so a client following it on another process keeps it going, and one that came and went there restarts the count. Stamp detachedAt with the clock the hubs use (now, Date.now by default).

retentionMs and maxStreams belong to the in-memory store, next to a store of your own they throw. maxEvents still bounds what waits for one client.

WebSocketCopy Anchor Link

The WebSocket endpoints do not resume. Their frames use their own vocabulary (token, tool-call, complete, cancelled) rather than the stream protocol the store keeps, and a run lives as long as its socket: closing it or sending stop cancels the run. Use the SSE routes for runs a client must be able to resume.