Durable token streaming

Stream LLM tokens to the browser while running generation as a durable function, so a dropped connection or a crash never loses the response.

Token streaming means the LLM server returns tokens one at a time as it generates them, so your agent produces a progressive "typing" effect instead of delivering everything at once. Users start reading a long output sooner, they can stop the agent midway if it's heading the wrong direction, and a response they watch arrive feels faster than one they wait on. Most implementations reach for WebSockets or SSE (Server-Sent Events) to push near-realtime updates from the agent to the app.

But both depend on one fragile persistent connection: if the socket drops, the run dies with it. Lost progress, half-finished tool calls. This pattern keeps token streaming alive through a dropped connection, using realtime connections and Inngest connect.

§Architecture

Decouple the agent's execution from the transport that carries its tokens.

Start with a standard REST request. The browser POSTs to /api/chat and gets an event id back immediately.

The Inngest execution engine receives the chat/message.sent event and issues run requests to the worker over the connect websocket.

The worker runs the agent's tasks (model turns, tool calls) as step.run and publishes tokens as the model produces them.

The browser subscribes with short-lived subscription tokens and receives the live stream.

Because the Inngest engine tracks progress instead of the socket:

If the worker dies, Inngest retries the function on the next healthy worker.
Finished steps replay instantly, since Inngest memoizes each one as it completes.
The interrupted turn restreams from seq: 0; the UI resets just that bubble.
The browser can re-subscribe at any time; it's only a viewer.

💡 The tradeoff: there's no server-side session store. The browser holds the transcript and POSTs all of it on every /api/chat call. That keeps the server stateless, so any worker instance can pick up any request, and it keeps this example small. The cost is that the browser is the only copy of history. A real app would persist history (e.g. keyed by sessionId) and send only the new message.

§How to Implement Durable Token Streaming?

The full working example is in the repo. Below are the pieces that matter.

The Worker

To cut latency, open a direct websocket connection from the worker to Inngest. Import and set up the inngest/connect package.

typescript

01import { connect } from "inngest/connect";
02import { inngest } from "../inngest/client";
03import { chatFn } from "./chat-function";
04
05// Establish an outbound persistent connection to Inngest.
06const connection = await connect({
07  apps: [{ client: inngest, functions: [chatFn] }],
08  // Identifies this worker instance for horizontal scaling and rolling
09  // deploys. Defaults to hostname if unset; in containers set this to the
10  // container id.
11  instanceId: process.env.INNGEST_INSTANCE_ID,
12});
13
14console.log("Worker: connected", connection.state);
15
16// Block until the connect socket gracefully closes (SIGTERM/SIGINT), then exit.
17await connection.closed;
18console.log("Worker: shut down");
19process.exit(0);

Next, a channel to publish the stream, with two topics: status and tokens.

typescript

01import { channel, staticSchema } from "inngest/realtime";
02
03// Shared contract between the worker (publisher) and the browser (subscriber).
04// Both sides import this file, so the topic shapes can never drift apart.
05
06// One `tokens` message is a small batch of streamed text for a single model
07// turn. `seq` is a per-turn counter starting at 0 — the UI groups by `turn`
08// and orders by `seq`, and treats a re-appearing `seq: 0` as a replay (the
09// turn's step was retried, so its buffer should be reset and re-built).
10export type TokenMessage = { turn: number; seq: number; delta: string };
11
12// Lifecycle/status events, published durably (`step.realtime.publish`) so
13// they survive worker restarts and are never duplicated or dropped.
14export type StatusMessage =
15  | { type: "run.started" }
16  | { type: "tool.called"; turn: number; name: string; input: unknown }
17  | { type: "tool.result"; turn: number; name: string; output: string }
18  | { type: "turn.completed"; turn: number; text: string }
19  | { type: "run.completed"; text: string }
20  | { type: "run.failed"; error: string };
21
22// One channel per chat session, with two topics: high-frequency token deltas
23// and low-frequency lifecycle status. Subscribers pick a session by calling
24// `chatChannel(sessionId)`.
25export const chatChannel = channel({
26  name: (sessionId: string) => `chat:${sessionId}`,
27  topics: {
28    tokens: { schema: staticSchema<TokenMessage>() },
29    status: { schema: staticSchema<StatusMessage>() },
30  },
31});
32
33// Client-owned chat history shape (see api/chat/route.ts) — this app has no
34// server-side session store, so the browser sends the full transcript on
35// every request.
36export type ChatMessage = { role: "user" | "assistant"; content: string };

status publishes durably, through step.realtime.publish. The publish is a step, so it's memoized and a completed turn's status never gets re-sent.
tokens publishes non-durably, through a plain inngest.realtime.publish call. Nothing is memoized, so when a step retries the client gets the text streamed again.

💡 One caveat: publishing every single token wastes bandwidth. The worker batches deltas and flushes every 40ms instead; frequent enough to read as live typing, without a message per token.

typescript

01// Runs one model turn as a single durable step. Streaming to the browser is a
02// side effect of that step (non-durable `inngest.realtime.publish` calls) —
03// if the step retries, it re-streams from `seq: 0` and the UI resets that
04// turn's buffer (see Chat.tsx); once the step completes, it's memoized and
05// never re-streams or re-calls the model again.
06const result = await step.run(`llm-turn-${turn}`, async () => {
07  let seq = 0;
08  let buffer = "";
09  let timer: ReturnType<typeof setTimeout> | null = null;
10  const pending: Promise<unknown>[] = [];
11
12  const flush = () => {
13    if (!buffer) return;
14    const delta = buffer;
15    buffer = "";
16    timer = null;
17    // Non-durable and non-blocking: this is a live UI nicety, not part of
18    // the durable record. `turn.completed` (below, durable) is the
19    // authoritative text the UI falls back to if any deltas are lost. Each
20    // publish is caught individually — a dropped batch is recoverable, but
21    // an unhandled rejection is fatal under Node's default settings — and
22    // tracked in `pending` so the step doesn't settle with publishes still
23    // in flight (a serverless runtime may freeze the instance as soon as
24    // the handler returns, silently dropping the tail of the stream).
25    pending.push(
26      inngest.realtime.publish(ch.tokens, { turn, seq: seq++, delta }).catch(() => {}),
27    );
28  };
29
30  const stream = client.messages.stream({
31    model,
32    // Both experiment variants (Haiku 4.5 and Opus 4.8) stream and support
33    // ≥64K max output tokens; 16K leaves ample room for ~20 parallel tool
34    // calls plus a long final answer in one turn while keeping worst-case
35    // cost bounded alongside MAX_TURNS.
36    max_tokens: 16384,
37    system: SYSTEM_PROMPT,
38    tools: toolDefinitions,
39    messages,
40  });
41
42  stream.on("text", (delta) => {
43    buffer += delta;
44    timer ??= setTimeout(flush, BATCH_MS);
45  });
46
47  try {
48    return await stream.finalMessage();
49  } finally {
50    // Always cancel a pending batch so no flush fires after the step
51    // settles — on failure, a stray timer would otherwise publish stale
52    // tokens into the retry's fresh stream.
53    if (timer) clearTimeout(timer);
54    flush();
55    // Never rejects — every promise in `pending` is already caught.
56    await Promise.all(pending);
57  }
58});

The Browser

The Next.js app imports the same ./inngest/channel module the worker uses, then subscribes with the inngest/react package.

typescript

01import { getClientSubscriptionToken } from "inngest/react";
02import { inngest } from "../../../inngest/client";
03import { chatChannel } from "../../../inngest/channel";
04
05// Mints a short-lived subscription token scoped to one session's channel and
06// its two topics. The browser fetches this right before subscribing with
07// `useRealtime` (see components/Chat.tsx) — the token is never embedded in
08// page HTML or reused across sessions.
09export async function POST(req: Request) {
10  const { sessionId } = (await req.json()) as { sessionId: string };
11
12  if (!sessionId) {
13    return Response.json({ error: "Expected { sessionId }" }, { status: 400 });
14  }
15
16  const token = await getClientSubscriptionToken(inngest, {
17    channel: chatChannel(sessionId),
18    topics: ["tokens", "status"],
19  });
20
21  return Response.json(token);
22}

The token is scoped to one session's channel and its two topics, and the browser mints a fresh one right before it subscribes. It never lands in the page HTML and never gets reused across sessions.

typescript

01import { useRealtime } from "inngest/react";
02...
03
04
05export default function Chat() {
06  // One id per browser tab/session, generated once. There's no server-side
07  // session store — the transcript below is the only place history lives.
08  const [sessionId] = useState(() => crypto.randomUUID());
09  const [transcript, setTranscript] = useState<TranscriptEntry[]>([]);
10  const [input, setInput] = useState("");
11  const [running, setRunning] = useState(false);
12  // A failed run keeps whatever trace it accumulated too, so it isn't just
13  // silently discarded — it's shown alongside the error line.
14  const [errorBanner, setErrorBanner] = useState<{ message: string; trace: TraceItem[] } | null>(null);
15  const bottomRef = useRef<HTMLDivElement | null>(null);
16
17  const { messages, reset } = useRealtime({
18    channel: chatChannel(sessionId),
19    topics: ["tokens", "status"],
20    token: async () => {
21      const res = await fetch("/api/realtime-token", {
22        method: "POST",
23        headers: { "Content-Type": "application/json" },
24        body: JSON.stringify({ sessionId }),
25      });
26      return res.json();
27    },
28    // Only subscribe while a run is actually in flight — no point holding a
29    // connection open (and no realtime token to mint) between sends.
30    enabled: running,
31  });
32
33  ...

The client keeps no run state, which makes reconnecting cheap. It groups deltas by turn, orders them by seq, and reads a re-appearing seq: 0 as a replay: that turn's step retried, so the browser clears the bubble's buffer and rebuilds it. There's no cursor to persist and no gap to backfill. If deltas go missing entirely, the durable turn.completed still arrives with the authoritative text, so the worst case is a bubble that fills in all at once instead of typing.

Finally, the route that sends the event.

typescript

01// Stateless server: the browser owns the full transcript and sends it on
02// every request (see components/Chat.tsx). Simplest possible design for an
03// example — a real app would likely persist history server-side instead.
04export async function POST(req: Request) {
05  const { sessionId, messages } = (await req.json()) as { sessionId: string; messages: ChatMessage[] };
06
07  if (!sessionId || !Array.isArray(messages)) {
08    return Response.json({ error: "Expected { sessionId, messages }" }, { status: 400 });
09  }
10
11  const { ids } = await inngest.send({
12    name: "chat/message.sent",
13    data: { sessionId, messages },
14  });
15
16  return Response.json({ eventId: ids[0] });
17}

§Summary

Ask the agent for something long, like a two-page report on current weather conditions in Omaha, NE, and tokens start arriving right away instead of after the whole generation finishes.

On the Inngest dashboard, the model turns and tool-call steps are all tracked, ready to retry if one fails.

Clone the repo, swap in your own model and tools, then kill the worker mid-stream and watch the turn restream from seq: 0.

← PreviousReliably run critical workflows