Multi-replica streaming (Redis)

The default token sink buffers a run's tokens in process memory — great for a single instance, but it means a run started on one pod can only be re-attached on that same pod. The Redis transport removes that constraint: any replica can serve any run's SSE stream, with the SSE envelope kept byte-identical.

Select itCopy Anchor Link

import { defineConfig, tokenSinks } from '@adonis-agora/agent'

export default defineConfig({
  model: () => aiSdkModel(openai('gpt-4o-mini')),
  sink: tokenSinks.redis({ connection: 'main' }),
})

tokenSinks.redis (aliased streamTransports.redis) fans a run's tokens across replicas over Redis pub/sub plus a replayable list, so a late or cross-instance subscriber replays the buffered chunks first and then follows live. Requires @adonisjs/redis installed and configured (config/redis.ts).

ConfigurationCopy Anchor Link

ttlSeconds is the knob that keeps Redis from filling up: the framework never calls the sink's close(), so a run's replay buffer has no natural end of life. One hour is long enough that a client re-attaching after a browser restart still replays the whole run, and short enough that yesterday's transcripts aren't sitting in memory. Raise it if you re-attach on a longer horizon; set 0 only if something else prunes the keys.

What it doesCopy Anchor Link

• Write. Each frame is an RPUSH onto <keyPrefix>:<runId>:chunks, a refresh of that key's TTL, and a PUBLISH on <keyPrefix>:<runId> to wake the subscribers. The element is the frame's JSON with one extra property, "#", a write id no other element shares (the writing process's random id and a counter); subscribers remove it, so they yield exactly the frame written. The end of the run sets <keyPrefix>:<runId>:state and publishes end.
• Coalesce. Consecutive deltas written within 50 ms are pushed as one frame holding their concatenation: text frames, reasoning deltas, and the tool-input-deltas of one tool call. Every reader appends these deltas, so the rendered text is the same; a client receives fewer, larger frames. Any other frame, and the end of the run, push the gathered delta first, so order is kept. A process that dies inside that window loses the gathered delta from the live stream; the persisted message is unaffected.
• Read. A subscriber replays the list, then follows the channel. It reads the end marker once when it attaches; after that it relies on the end message. After 60 s without a message it reads the list on its own, because expiring keys publish nothing: when the run's keys are gone it ends, and when the list started over (a run parked past ttlSeconds that then wrote again) it replays from the new first element instead of skipping frames. It tells a restarted list apart by the write id of the last element it read, so a restart that repeats the run's first frames (a step-start, say) is still caught. Elements without a write id, as earlier versions wrote them, are read the same way and compared as they are.
• Disconnect. When the browser closes the SSE connection, the route ends its subscription and the subscriber unsubscribes.

Bring your own clientCopy Anchor Link

Passing a client lets you drive a non-Adonis Redis driver or a test double directly — the factory uses it verbatim. Under the hood the adapter uses a dedicated duplicated connection per subscribe (a subscribed Redis connection can't run other commands).

RedisStreamClient.append(key, value, ttlSeconds, channel, message) is optional. It does a frame's RPUSH, EXPIRE (skipped when ttlSeconds is 0) and PUBLISH, in that order, in one round trip — a MULTI or a pipeline. The @adonisjs/redis adapter implements it with MULTI. A client without it gets the three commands one after another, a round trip each.

Pair with the durable runnerCopy Anchor Link

The Redis sink solves stream portability across replicas; the durable runner solves run portability and crash-survival (a suspended HITL approval resumes on any pod). Together they give you a horizontally-scaled, crash-proof agent: sink: tokenSinks.redis(...) + durable: true.

No Redis?Copy Anchor Link

tokenSinks.lucid() gives the same cross-replica stream over the app's SQL database, polled instead of pushed. It is slower to deliver and it writes to your database on every streamed answer, so prefer Redis when you have it.

For the underlying class, RedisTokenStreamSink (and the RedisStreamClient interface) are exported from the package root.