Streaming LLM responses

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) 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.

-- 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 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:

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 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:

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: decide inside the transaction, run the model after it commits:

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:

// 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:

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

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:

  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 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

# See also

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