# Streaming LLM responses

Model tokens want to be on screen in milliseconds and in the database almost never. The stream plane splits a response into a live SSE tail and checkpointed rows in your own tables, joined by one monotone offset — so a reload, a late joiner, or a second device always converges on exactly the text the model produced. Includes cancel and a `useStreamedText` hook.

A language-model response arrives as hundreds of tiny deltas per second. Every one of them wants to
be on a screen immediately, and none of them wants to be a durable write. Routing each delta
through the mutation path costs a transaction, a replication frame, and a view-maintenance pass
**per token**. Routing none of them means the chat doesn't survive a refresh.

The **stream plane** (`streams:` on the [API server](/docs/api-server)) splits the response in two:

- The **durable plane** holds a checkpointed prefix: every ~512 characters / 750ms a **chunk row**
  commits to *your own tables* through the ordinary write path, so it fans out through sync to every
  subscribed client like any other write.
- The **live plane** carries the tail: each delta is fanned to SSE subscribers the moment the model
  emits it.

Both are measured in the same monotone offset (`seq`, a count of UTF-16 code units), so merging them
on the client is a one-liner. Two prefixes of the same string merge by taking the longer. **The
fallback is the load-bearing part**: a client that can't reach the live stream — wrong instance,
dropped connection, no `EventSource` — still watches the message grow at checkpoint granularity
through its ordinary query. The live leg only makes it smooth.

When the stream closes, the plane **compacts**: one transaction writes the whole body onto the
message row and deletes the chunk rows. Steady state is one row per message, and the client renders
the identical string across the handoff — no flicker.

## The tables are yours

The plane writes to *your* schema. The message row carries whatever else your app wants (`chatId`,
`role`, token counts), and the chunk table is reachable from your chat query as an ordinary
relationship. You map the columns. Only `body`, `seq`, and `status` are required (`cancelRequested`
is opt-in — naming it is what turns cancellation on). `seq` must be `INTEGER NOT NULL DEFAULT 0`:
it's the compare-and-swap column that makes checkpoints idempotent, and nothing matches `NULL`.

```sql
-- migrations/0005_chat.sql
CREATE TABLE IF NOT EXISTS message (
  id TEXT NOT NULL,
  chatId TEXT NOT NULL,
  role TEXT NOT NULL,                          -- 'user' | 'assistant'
  body TEXT NOT NULL DEFAULT '',               -- the compacted response
  seq INTEGER NOT NULL DEFAULT 0,              -- the CAS'd durable length
  status TEXT NOT NULL DEFAULT 'pending',      -- 'streaming' → 'complete' | 'cancelled' | 'error' | 'interrupted'
  cancelRequested INTEGER NOT NULL DEFAULT 0,  -- opt-in: the stop button
  createdAt REAL NOT NULL,
  PRIMARY KEY (id)
);
CREATE INDEX IF NOT EXISTS message_chat_order ON message (chatId, createdAt, id);

-- exactly what streamChunkTableDdl({ message: "message", chunks: "message_chunk" }, sqliteDialect) emits
CREATE TABLE IF NOT EXISTS "message_chunk" (
  "id" TEXT PRIMARY KEY, "streamId" TEXT NOT NULL, "seq" INTEGER NOT NULL, "text" TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS "message_chunk_stream_seq" ON "message_chunk" ("streamId", "seq");
```

`rindle schema gen` picks both tables up like any other [schema](/docs/schema) change.

## Wire the plane

Tell `createRindleApiServer` where things live. `authorize` is required — subscribing to a stream is
reading someone's chat. It runs *before* existence is checked, so a denial can't be used to
probe for stream ids:

```ts
export const api = createRindleApiServer<User>({
  // …queries, mutators, authorizers as usual…
  streams: {
    checkpoint: {
      tables: {
        message: "message",
        chunks: "message_chunk",
        columns: { cancel: "cancelRequested" },
      },
    },
    authorize: async ({ user, streamId }) => user !== undefined && ownsMessage(user, streamId),
  },
});
```

The mapping is validated loudly at construction: a missing table or column fails the deploy, not a
3am generation. Checkpoint cadence defaults to 512 chars / 750ms and is tunable via
`streams.policy`.

## Send is one predicted mutation

One [mutator](/docs/mutators) writes the user's message **and** an empty assistant row. The
assistant row is the *pointer*: it exists, with `seq = 0`, before any token. That is what the
plane attaches to, and what a late joiner's query finds. Because the mutator is isomorphic, the
browser predicts both rows. The user's message and an empty assistant bubble appear before the
request even reaches the server:

```ts
sendMessage: shared(sendMessageArgs, function* (tx, a: SendMessageArgs): MutationGen {
  yield tx.insert("message", {
    id: a.userMessageId, chatId: a.chatId, role: "user",
    body: a.text, seq: a.text.length, status: "complete", cancelRequested: 0, createdAt: a.now,
  });
  yield tx.insert("message", {
    id: a.assistantMessageId, chatId: a.chatId, role: "assistant",
    body: "", seq: 0, status: "pending", cancelRequested: 0, createdAt: a.now + 1,
  });
}),
```

## Start the generation post-commit

The model call starts from the **same mutation**, not a second request from the browser (a separate
`fetch` can race the mutation queue and kick a row that doesn't exist yet). Override the server side
with [`scoped`](/docs/api-server#work-outside-the-mutation-transaction): decide *inside* the
transaction, run the model *after* it commits:

```ts
const apiMutators = defineApiMutators<User, ApiMutators<User>>({
  ...sharedApiMutators(mutators, sharedCtx),

  sendMessage: scoped(async (scope, raw, ctx) => {
    const a = sendMessageArgs.parse(raw);
    const kick = await scope.transact(async (tx) => {
      // Replay guard: a retried envelope dedupes the WRITE, not the effect —
      // read inside the transaction so a duplicate kick returns undefined.
      if (await tx.row("message", { id: a.assistantMessageId })) return undefined;
      await runSharedMutation(mutators.sendMessage, a, sharedCtx(ctx), tx);
      // Read the prompt in the SAME transaction that wrote the user's turn,
      // so the model sees exactly this history.
      const history = await tx.sql.query<{ role: string; body: string }>(
        "select role, body from message where chatId = ? order by createdAt, id", [a.chatId],
      );
      return { streamId: a.assistantMessageId, history };
    });
    if (kick) void startGeneration(ctx.user, kick);   // reachable only on the committed path
  }),
});

async function startGeneration(user: User, kick: Kick): Promise<void> {
  let s: StreamHandle;
  try {
    s = await api.openStream({ user, streamId: kick.streamId });
  } catch (err) {
    // A refusal IS the single-flight guard working: openStream refuses a row
    // that's missing, already advanced, or already streaming. Stand down.
    console.warn(`generation for ${kick.streamId} not started:`, err);
    return;
  }
  try {
    const upstream = anthropic.messages.stream({
      model: "claude-opus-5", max_tokens: 4096,
      messages: toPromptMessages(kick.history),
    });
    await s.pump(textDeltas(upstream));  // stops early — closing the iterator — on cancel
    await s.close();                     // compacts; seals `cancelled` if asked to stop
  } catch (err) {
    await s.fail(err);                   // seals `error` at whatever was produced
  }
}

async function* textDeltas(upstream: MessageStream): AsyncIterable<string> {
  try {
    for await (const event of upstream) {
      if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
        yield event.delta.text;
      }
    }
  } finally {
    upstream.abort();   // pump breaking early runs this — a cancel reaches the model
  }
}
```

`pump` drains any `AsyncIterable<string>` — the shape every LLM SDK's text stream already has. For
discrete events in the turn (a tool call, a usage record), write ordinary sibling rows with your own
mutators. `await s.flush()` first, so the text precedes them in the store's ordering.

## The subscribe route

One call — it parses the GET (`Last-Event-ID` included), authorizes, subscribes, and encodes SSE:

```ts
// src/routes/api.rindle.stream.ts
GET: async ({ request }) => api.streamResponse(request, { user: await authenticate(request) });
```

The default endpoint the React hook dials is `/api/rindle/stream`. For a custom transport
(WebSocket, chunked JSON), compose the pieces: `streamRequestFromHttp` → `api.subscribeStream` →
`streamFramesToSse`.

## Render the tokens

The durable side is just your chat query — the message plus its not-yet-compacted chunks as a
`related` subquery. The live side is one hook:

```ts
export const MessageFragment = defineFragment(message, (m) =>
  m.select("id", "role", "body", "seq", "status", "cancelRequested")
   .sub("chunks", rels.messageChunks, ChunkFragment, (c) => c.orderBy("seq", "asc")),
);
```

```tsx
import { assembleDurableText, useFragment, useStreamedText } from "@rindle/react";

export const Message = memo(function Message({ message }: { message: MessageRef }) {
  const data = useFragment(MessageFragment, message);
  const streaming = data != null && (data.status === "streaming" || data.status === "pending");
  const text = useStreamedText({
    streamId: data?.id ?? "",
    durable: data ? assembleDurableText(data, data.chunks) : "",  // body ++ un-compacted chunks
    live: streaming,
  });
  if (!data) return null;

  return (
    <li className={`msg msg-${data.role}`}>
      <p>{text}{streaming && <span className="caret" />}</p>
      {streaming && !data.cancelRequested && (
        <button onClick={() => app.mutate.cancelMessage({ messageId: data.id })}>Stop</button>
      )}
      {data.status === "interrupted" && <em>Response was cut short.</em>}
    </li>
  );
});
```

The merge itself is one line (`spliceStreamText` — take the longer prefix). The hook exists for the
subscription traps apps get subtly wrong without it:

- It reads the join offset from a ref, so a checkpoint doesn't reconnect the socket.
- It seeds its accumulator, so the splice can't discard the tail.
- It detaches on the terminal frame, so a finished stream doesn't reopen forever.
- It splices reconnect replays at their own offset instead of appending.

Without `EventSource` (SSR, an older
runtime) it attaches nothing and the reader stays on the durable plane — correct, just chunkier.

## The stop button

Cancel is durable state, not a routed message: `cancelMessage` is an ordinary predicted mutation
setting the mapped `cancel` column, from whatever instance the reader happens to be on. The producer
— already round-tripping to the store on every checkpoint — reads the flag on the same trip, so
tokens stop within the checkpoint cadence (~750ms). `pump` closes the SDK iterator (aborting the
upstream request), and `close()` seals the row `cancelled`.

## Operating it

- **Wire `drainStreams` to shutdown.** A rolling deploy compacts each in-flight response and seals
  it `interrupted` instead of stranding `streaming` rows:
  ```ts
  process.on("SIGTERM", async () => { await api.drainStreams(); api.close(); process.exit(0); });
  ```
- **Honesty.** A live `chunk` frame is *not* a durability claim. If the host dies between
  checkpoints, the tail after the last checkpoint is gone and the row says so. Add an app-side
  sweeper that marks long-`streaming` rows `interrupted` — the
  [background-writes recipe](/docs/background-writes) implements exactly this sweeper.
- **Multiple api-server instances.** A stream is hosted by one process. A subscriber landing
  elsewhere degrades to checkpoint-granularity updates through sync — correct, just chunky. To keep
  it smooth across instances, either route subscribes to the hosting instance (map
  `columns.host` + set `hostId`, and the open write records who's producing) or configure
  `streams.relay`. The relay is a small `publish`/`attach` adapter over your bus (Redis, NATS, a
  Durable Object) that moves the live frames cross-process. The plane conforms whatever the relay yields,
  so a broken relay costs smoothness, never correctness.

## What the user sees

| moment | what happens | where it came from |
| --- | --- | --- |
| click Send | their message + an empty assistant bubble | the predicted mutation, zero round trips |
| ~200ms | tokens begin | SSE `chunk` frames |
| every ~512 chars | nothing visible | a chunk row commits; other devices catch up here |
| click Stop | tokens stop within ~750ms | predicted cancel → the next checkpoint's probe |
| on completion | the same text, now durable | compaction; the splice returns the identical string |
| reload mid-stream | the text so far, then live tokens again | `body ++ chunks`, then a subscribe at that length |
| a second device | checkpoint-granularity text, then smooth | sync fanout, then its own subscribe |

## See also

- [The API server](/docs/api-server) — `scoped()` post-commit effects, authorizers, and the server
  the stream plane attaches to.
- [Isomorphic mutators](/docs/mutators) — the predicted `sendMessage`/`cancelMessage` bodies.
- [Compose the UI with fragments](/docs/fragments) — the `chunks` relationship the durable read
  rides on.
- [Handling rejected writes](/docs/rejected-writes) — what happens when the send itself is refused.

---

[View this page on Rindle](https://rindle.sh/docs/llm-streams)