Streaming LLM responses

Rindle can deliver generated text as live events while storing periodic checkpoints. The live events reduce display latency. The checkpoints let another client recover the stored text through an ordinary query.

This guide provides integration modules for an existing synced app. It assumes SQL migrations, generated TypeScript tables, a browser client, and an HTTP server that can return a Fetch Response. See the synced-app quickstart for those foundations.

The example uses public response records and a deterministic demo producer. It does not call a model provider. A private chat app must add authenticated ownership checks, as described in the final section.

# Understand the two sources of text

Offsets count UTF-16 code units, the unit used by JavaScript string lengths. The default checkpoint triggers are 512 code units or 750 milliseconds. Checkpoints run serially. A slow database can produce fewer, larger checkpoints. These defaults are not latency or durability guarantees.

On successful completion, one transaction writes the whole response to body and removes its chunk rows. The client combines the stored prefix with live text without counting the checkpoint twice. Live text that has not reached a checkpoint can be lost after a process failure.

# Add the tables

Add these tables in a new migration:

CREATE TABLE response (
  id TEXT NOT NULL PRIMARY KEY,
  prompt TEXT NOT NULL,
  body TEXT NOT NULL DEFAULT '',
  seq INTEGER NOT NULL DEFAULT 0,
  status TEXT NOT NULL DEFAULT 'pending',
  cancelRequested INTEGER NOT NULL DEFAULT 0,
  host TEXT NOT NULL DEFAULT ''
);

CREATE TABLE response_chunk (
  id TEXT NOT NULL PRIMARY KEY,
  streamId TEXT NOT NULL,
  seq INTEGER NOT NULL,
  text TEXT NOT NULL
);
CREATE INDEX response_chunk_order ON response_chunk (streamId, seq);

Apply the migration and regenerate shared/schema.gen.ts through your existing rindle dev --migrate --gen command. The examples use its schema, response, and response_chunk exports. They do not modify generated files by hand.

seq must start at zero and cannot be NULL. The stream uses it to avoid applying a checkpoint twice. The mapped host column also lets competing processes resolve which producer opened the response. The stream owns body, seq, and status after opening.

# Define the shared writes and read

A start mutation creates the response row before generation starts. A cancel mutation changes durable state that the producer reads during a checkpoint.

// shared/responses.ts
import { defineMutators, defineQuery, newQueryBuilder } from "@rindle/client";
import { z } from "zod";
import { schema, response_chunk } from "./schema.gen.ts";

export const startResponseArgs = z.object({
  id: z.string().uuid(),
  prompt: z.string().min(1).max(4000),
});
export type StartResponseArgs = z.infer<typeof startResponseArgs>;

const responseIdArgs = z.object({ id: z.string().uuid() });
const { shared } = defineMutators(schema);

export const responseMutators = {
  startResponse: shared(startResponseArgs, function* (tx, args) {
    yield tx.insert("response", {
      id: args.id, prompt: args.prompt, body: "", seq: 0,
      status: "pending", cancelRequested: 0, host: "",
    });
  }),
  cancelResponse: shared(responseIdArgs, function* (tx, args) {
    yield tx.update("response", { id: args.id, cancelRequested: 1 });
  }),
};

const q = newQueryBuilder(schema);
export const responseQuery = defineQuery(
  "response",
  (raw) => responseIdArgs.parse(raw),
  ({ id }) => q.response.where.id(id)
    .select("id", "prompt", "body", "seq", "status", "cancelRequested")
    .sub("chunks", response_chunk, { parent: ["id"], child: ["streamId"] }, (chunks) =>
      chunks.select("seq", "text").orderBy("seq", "asc"),
    )
    .one(),
);

Add responseMutators to the browser client’s mutator registry. For example, merge it with the quickstart’s existing registry as { ...mutators, ...responseMutators }. Keep the same combined registry type in your client module.

# Start the producer after the row commits

This producer defines the input shape required by pump: an AsyncIterable<string>. Replace its implementation with your model SDK adapter when integrating a real model.

// server/demo-text.ts
import { setTimeout as delay } from "node:timers/promises";

export async function* demoText(prompt: string): AsyncIterable<string> {
  const words = `This is a demonstration response to: ${prompt}`.split(" ");
  for (const word of words) {
    await delay(50);
    yield `${word} `;
  }
}

Create a long-lived API instance for the response demo:

// server/response-api.ts
import {
  createRindleApiServer, defineApiMutators, registerQueries,
  runSharedMutation, scoped, sharedApiMutators,
} from "@rindle/api-server";
import type { ApiMutators, RindleApiServer } from "@rindle/api-server";
import { schema } from "../shared/schema.gen.ts";
import { responseMutators, responseQuery, startResponseArgs } from "../shared/responses.ts";
import { demoText } from "./demo-text.ts";

const sharedContext = () => ({ user: "public-demo" });

const apiMutators = defineApiMutators<undefined, ApiMutators<undefined>>({
  ...sharedApiMutators(responseMutators, sharedContext),
  startResponse: scoped(async (scope, raw) => {
    const args = startResponseArgs.parse(raw);
    const start = await scope.transact(async (tx) => {
      if (await tx.row("response", { id: args.id })) return false;
      await runSharedMutation(responseMutators.startResponse, args, sharedContext(), tx);
      return true;
    });
    if (start) {
      void produceResponse(args.id, args.prompt).catch((error) => {
        console.error("Response generation failed", args.id, error);
      });
    }
  }),
});

export const responseApi: RindleApiServer<undefined> = createRindleApiServer({
  rindle: {}, // RINDLE_URL and RINDLE_DATABASE_TOKEN stay on the server.
  schema,
  queries: registerQueries<undefined>([responseQuery]),
  mutators: apiMutators,
  authorizeQuery: () => true,
  authorizeMutation: () => true,
  streams: {
    checkpoint: {
      tables: {
        message: "response",
        chunks: "response_chunk",
        columns: { cancel: "cancelRequested", host: "host" },
      },
    },
    authorize: () => true, // Every response in this demonstration is public.
  },
});

async function produceResponse(streamId: string, prompt: string): Promise<void> {
  const stream = await responseApi.openStream({ user: undefined, streamId });
  try {
    await stream.pump(demoText(prompt));
    await stream.close();
  } catch (error) {
    await stream.fail(error);
  }
}

The scoped mutator completes its database transaction before it starts generation. The transaction guard skips an already-created response. openStream also refuses a missing row, an advanced response, or an active competing producer. A regeneration needs a new response ID.

streams.authorize guards subscribers. It does not authorize openStream. Generation starts through trusted server code after mutation authorization. The example’s public authorizers are deliberate and are unsuitable for private conversations.

The process must remain alive while produceResponse runs. Starting work after a commit is not a durable job queue. A crash between committing the row and starting generation can leave it pending. For reliable background generation, persist a job and run it through your application’s worker lifecycle.

For a real model adapter, release or abort the upstream request in the iterator’s finally block. pump closes the iterator when it observes cancellation, but the adapter owns the provider-specific cleanup. For an ordered tool-result write, call await stream.flush() before storing the separate result row.

# Mount the HTTP operations

This Fetch-style handler defines both the JSON operations and the SSE subscription:

// server/response-handler.ts
import { RindleApiError } from "@rindle/api-server";
import { responseApi } from "./response-api.ts";

export async function handleResponseRequest(request: Request): Promise<Response> {
  const path = new URL(request.url).pathname;
  const context = { user: undefined, request };
  try {
    if (request.method === "GET" && path === responseApi.routes.stream) {
      return await responseApi.streamResponse(request, context);
    }
    if (request.method !== "POST") return new Response("Not found", { status: 404 });
    const body: unknown = await request.json();
    if (path === responseApi.routes.query) {
      return Response.json(await responseApi.handleQueryJson(body, context));
    }
    if (path === responseApi.routes.read) {
      return Response.json(await responseApi.handleReadJson(body, context));
    }
    if (path === responseApi.routes.mutate) {
      return Response.json(await responseApi.handleMutateJson(body, context));
    }
    return new Response("Not found", { status: 404 });
  } catch (error) {
    const status = error instanceof RindleApiError ? error.status
      : error instanceof SyntaxError ? 400 : 500;
    console.error(error);
    return Response.json({ error: "Response request failed" }, { status });
  }
}

Mount this function through your HTTP framework’s Request/Response adapter. The default paths are /api/rindle/query, /api/rindle/read, /api/rindle/mutate, and /api/rindle/stream. A proxy must forward the stream response without buffering it to completion.

All handlers must use the same long-lived API instance to access its in-process live streams. Do not construct and close an API instance for each SSE request. To retain existing app operations, combine their query and mutator registries in this same instance. Their authorization policies must also remain explicit.

# Render the stored prefix and live text

This component uses the quickstart’s exported app after adding responseMutators to its browser registry:

// src/ResponseText.tsx
import { assembleDurableText } from "@rindle/client";
import { useQuery, useQueryStatus, useStreamedText } from "@rindle/react";
import { responseQuery } from "../shared/responses.ts";
import { app } from "./rindle-client.ts";

export function ResponseText({ id }: { id: string }) {
  const query = responseQuery({ id });
  const data = useQuery(query);
  const status = useQueryStatus(query);
  const streaming = data?.status === "streaming";
  const text = useStreamedText({
    streamId: id,
    durable: assembleDurableText(data, data?.chunks ?? []),
    live: streaming,
  });

  if (data === null) {
    return <p>{status === "complete" ? "Response not found." : "Loading response…"}</p>;
  }
  return (
    <article>
      <h2>{data.prompt}</h2>
      <p style={{ whiteSpace: "pre-wrap" }}>{text}</p>
      <p>{data.cancelRequested && streaming ? "Stopping…" : data.status}</p>
      {streaming && !data.cancelRequested && (
        <button onClick={() => app.mutate.cancelResponse({ id })}>Stop</button>
      )}
    </article>
  );
}

Render this component under the same Rindle provider as the browser client. Pass a response ID from your route or application state. Start a response from a submit handler with this helper:

// src/start-response.ts
import { startResponseArgs } from "../shared/responses.ts";
import { app } from "./rindle-client.ts";

export function startResponse(prompt: string): string {
  const args = startResponseArgs.parse({ id: crypto.randomUUID(), prompt: prompt.trim() });
  app.mutate.startResponse(args);
  return args.id;
}

The returned ID identifies a local prediction, not an accepted or completed response. Retain it in route or application state so a reload can query the same response. Use the client’s rejection callback to report a refused start or cancel mutation.

The hook subscribes only after the row becomes streaming. Subscribing while it is still pending can reach the server before a producer exists and receive absent. After absent, stale, or end, the hook detaches and continues to use the stored query result. A checkpoint does not reopen the subscription.

Without EventSource, the hook uses only the durable text. The default transport cannot add arbitrary authentication headers. For private streams, use a same-origin authenticated session or supply a custom StreamTransport.

# Cancellation, failures, and shutdown

A cancel mutation sets cancelRequested in the database. The producer observes that flag on a checkpoint round trip. pump then stops as the upstream iterator yields. A stalled iterator or slow database can delay cancellation beyond the configured checkpoint interval.

close() stores the complete produced body and removes the chunks. fail(error) tries to store that body with an error status. Both operations can reject if the final database write fails. The live tail is not durable merely because the client displayed it.

On shutdown, stop accepting new generation work, then call await responseApi.drainStreams() before responseApi.close(). Draining attempts to store active streams and mark them interrupted. A hard process failure cannot perform that step. Use an application recovery job for abandoned pending or streaming rows. See background writes for the write pattern.

Mapped-table mode retains the whole generated response in process memory until sealing and later eviction. It does not support retainChars as a memory limit. Bound generation length and concurrency in the application.

With several API instances, a subscriber can reach a process that does not host its stream. Without a relay, that process returns absent, and the query continues to receive checkpoints. Route to the producing process or configure streams.relay for live delivery across instances. A relay does not replace the database or its authorization rules.

# Make a private conversation private

For a private app, replace all three public authorizers in the example. Derive the user from a verified server session. Scope the response query, start and cancel mutations, and stream subscription to the same ownership policy. The default SSE transport needs authentication that the browser can send on that request.

A subscriber can reach a process without the live stream’s meta value. Authorize from the response ID and durable application data in that case. An unpredictable ID alone is not an ownership check. See authorization for context-scoped queries and guarded writes.