Multi-replica streaming (SQL, no Redis)

The default token sink buffers a run's frames in process memory, so a run can only be watched — and re-attached — on the replica that started it. The Redis transport removes that limit. The Lucid sink removes it too, for a deployment that has a SQL database and no Redis: frames are rows in the app's database, and every replica reads them.

Select itCopy Anchor Link

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

export default defineConfig({
  model: () => aiSdkModel(openai('gpt-4o-mini')),
  store: 'lucid',
  stores: { lucid: stores.lucid() },
  sink: tokenSinks.lucid(),
})

Nothing about the routes, the SSE envelope or the client changes. The sink works with any store (it only needs a Lucid connection). It is tested on SQLite, Postgres 16 and MySQL 8.4.

What it doesCopy Anchor Link

• Write. The replica running the turn appends each frame to agent_stream_frame (run_id, seq, frame, created_at; primary key (run_id, seq)). seq starts at 1 per run and has no gaps. The row whose frame is NULL is the end marker.
• Read. A subscriber reads the rows past its cursor, in order, every pollIntervalMs, until it meets the end marker. One that attaches late — or after the run ended — replays from the first row. Each read also re-reads the last row the subscriber already has, which is how it notices that the rows it read are gone: when the run has no rows left (purged past its TTL, or close()d) the subscriber ends, and when the run's numbering started over (a run parked past its TTL that then wrote again) it replays from the new first row instead of skipping frames.
• Coalesce. A model streams token by token, and a row per token is more writes than a chat answer deserves. Consecutive deltas written within flushMs are stored 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 merged frame renders the same. Any other frame, and the end of the run, write the gathered delta first, so order is kept.
• Disconnect. When the browser closes the SSE connection, the route ends its subscription, so a run parked on a person is not polled for a client that left.

ConfigurationCopy Anchor Link

CleanupCopy Anchor Link

A run's rows lapse ttlSeconds after its last write — an ended run past its replay window and a run that crashed without ending alike. autoPurge deletes them as a side effect of streaming, which is enough for a replica that keeps serving turns. To purge on a schedule instead (or as well), call the sink yourself:

import { LucidTokenStreamSink } from '@adonis-agora/agent'
import db from '@adonisjs/lucid/services/db'

// from a scheduled job, on any replica
await new LucidTokenStreamSink(db).purgeExpired()

close(runId) deletes one run's rows at once.

The purge reads the (created_at, run_id) index on the frame table to find runs with a row older than the cutoff, then checks those candidates for a newer row through the primary key. Its cost follows the number of rows that aged out, not the size of the table.

What it costsCopy Anchor Link

• Latency. Delivery is polled: a frame reaches the browser up to pollIntervalMs (plus flushMs for text) after it was produced. Text arrives in small bursts rather than token by token.
• Database load. A streamed answer is roughly twenty inserts a second while the model is talking, and each open SSE connection is four small indexed reads a second (one, once the run goes quiet). That is fine for tens of concurrent chats on a database that is already there; it is the wrong place for hundreds.
• A crash window. Text a writer is still gathering lives in the process for at most flushMs. A process that dies in that window loses it from the live stream; the persisted message is unaffected.
• A parked run outlives its buffer. The TTL counts from the run's last write, so a run parked on a person for longer than ttlSeconds has its buffered frames purged while it waits. It still resumes and writes what comes next, but a browser attaching afterwards replays only from there — the persisted thread is the record of what came before. Raise ttlSeconds (or set 0 and purge on your own schedule) if people routinely answer later than that.

Use Redis when you already run it, when you want pushed rather than polled delivery, or when the number of concurrent streams makes polling your database the bottleneck. There is no LISTEN/NOTIFY wake-up on Postgres: the sink stays on the portable query surface every Lucid dialect has.

Several replicas without RedisCopy Anchor Link

The sink makes the stream portable. A run parked on a person — an action approval, or a question set from ask / intake — also has to be resumable from whichever replica receives the reply. Under the inline runner that wait is a promise in one process; under the durable runner it is a journaled signal. So the whole recipe is three lines of config:

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

export default defineConfig({
  model: () => aiSdkModel(openai('gpt-4o-mini')),
  store: 'lucid',
  stores: { lucid: stores.lucid() },
  durable: true,                    // parking is a durable signal, not a promise in one process
  sink: tokenSinks.lucid(),         // any replica serves and resumes any run's stream
  defaultAgent: { ask: true },      // the model may put a question set to the user
})

with @adonis-agora/durable configured on a store and a transport that are themselves shared between replicas without Redis — its Lucid store and its db transport (config/durable.ts).

What then happens to a question:

1. Replica A runs the turn. The model calls ask; the loop writes the elicitation frame and the workflow suspends on the signal tool:<runId>:<callId>. Nothing is held in memory.
2. The browser's SSE connection — on A, on B, or re-opened later with ?after= — reads the frame from the table.
3. POST <path>/tool-call/answer lands on replica B, which signals the engine. Whichever replica's worker claims the run resumes it, and its frames go to the same table.

The same holds for POST <path>/tool-call/approve and /reject.

For the underlying class, LucidTokenStreamSink is exported from the package root.