Streaming LLM responses to the UI: SSE, chunking, abort, and resume?

// SSE over a real network arrives in arbitrary byte chunks: a chunk can end in
// the middle of a line, a JSON object, or a multi-byte UTF-8 character.
// Compare a naive per-chunk parser with a buffered, streaming-decoder parser.

const deltas = ["Caf", "é au lait ", "costs ", "€4", " ☕ ", "enjoy 🙂"];
let body = "";
deltas.forEach((d, i) => { body += "id: " + (i + 1) + "\ndata: " + JSON.stringify({ delta: d }) + "\n\n"; });
body += "data: [DONE]\n\n";
const bytes = new TextEncoder().encode(body);

// Simulated network: fixed 40-byte reads (real reads vary; any split must work).
const CHUNK = 40;
const chunks = [];
for (let i = 0; i < bytes.length; i += CHUNK) chunks.push(bytes.slice(i, i + CHUNK));
console.log("body: " + bytes.length + " bytes, " + deltas.length + " delta events, delivered in " + chunks.length + " chunks of " + CHUNK + " bytes");

// 1) Naive: decode each chunk on its own and parse whatever lines it contains.
function naive() {
  let text = "", parsed = 0, broken = 0;
  for (const c of chunks) {
    const s = new TextDecoder().decode(c);            // no stream: true
    for (const line of s.split("\n")) {
      if (!line.startsWith("data: ") || line === "data: [DONE]") continue;
      try { text += JSON.parse(line.slice(6)).delta; parsed++; } catch { broken++; }
    }
  }
  return { text, parsed, broken };
}

// 2) Correct: one streaming decoder, a carry-over buffer, dispatch on blank line.
function buffered() {
  const dec = new TextDecoder();
  let buf = "", text = "", parsed = 0, lastId = null, done = false;
  const dispatch = (frame) => {
    let data = "";
    for (const line of frame.split("\n")) {
      if (line.startsWith("id: ")) lastId = line.slice(4);
      if (line.startsWith("data: ")) data += line.slice(6);
    }
    if (data === "[DONE]") { done = true; return; }
    if (data) { text += JSON.parse(data).delta; parsed++; }
  };
  for (const c of chunks) {
    buf += dec.decode(c, { stream: true });           // keeps half a character for later
    let at;
    while ((at = buf.indexOf("\n\n")) !== -1) { dispatch(buf.slice(0, at)); buf = buf.slice(at + 2); }
  }
  buf += dec.decode();
  return { text, parsed, lastId, done, leftover: buf.length };
}

const expected = deltas.join("");
const a = naive();
console.log("");
console.log("naive per-chunk parser");
console.log("  events parsed:   " + a.parsed + " of " + deltas.length + " (" + a.broken + " fragments failed JSON.parse)");
console.log("  text:            " + JSON.stringify(a.text));
console.log("  correct:         " + (a.text === expected));

const b = buffered();
console.log("");
console.log("buffered parser with streaming UTF-8 decoder");
console.log("  events parsed:   " + b.parsed + " of " + deltas.length);
console.log("  text:            " + JSON.stringify(b.text));
console.log("  correct:         " + (b.text === expected));
console.log("  last event id:   " + b.lastId + " (send as Last-Event-ID to resume)");
console.log("  saw [DONE]:      " + b.done + ", bytes left in buffer: " + b.leftover);

// Zoom in on one character: the 4-byte emoji split 2 + 2 across two reads.
const emoji = new TextEncoder().encode("🙂");
const halves = [emoji.slice(0, 2), emoji.slice(2)];
const hex = (s) => [...s].map((ch) => "U+" + ch.codePointAt(0).toString(16).toUpperCase().padStart(4, "0")).join(" ");
const perChunk = halves.map((h) => new TextDecoder().decode(h)).join("");
const sd = new TextDecoder();
const streamed = halves.map((h) => sd.decode(h, { stream: true })).join("") + sd.decode();
console.log("");
console.log("one emoji split across two reads");
console.log("  decoded per chunk:    " + hex(perChunk));
console.log("  streaming decoder:    " + hex(streamed));

// Disconnects, Stop buttons and resume, on a fake clock (one token per tick).
// "Tied" = generation lives inside the HTTP request (request abort aborts it).
// "Resumable" = generation runs as a job that appends events to a log (in
// production: Redis via a package such as resumable-stream); clients read it.

const ANSWER = "Streams need ids so a client can resume after a network drop.".split(" ");
const N = ANSWER.length;

function makeJob() {
  return { log: [], tokens: 0, cancelled: false,
    step() { if (this.cancelled || this.log.length === N) return;
      this.tokens++; this.log.push({ id: this.log.length + 1, delta: ANSWER[this.log.length] }); } };
}

function makeClient() {
  return { lastId: 0, text: [], dupes: 0,
    receive(ev) { if (ev.id <= this.lastId) { this.dupes++; return; } // idempotent by id
      this.lastId = ev.id; this.text.push(ev.delta); } };
}

function report(name, job, client, extraTokens = 0) {
  const billed = job.tokens + extraTokens;
  console.log(name.padEnd(44) + ("billed " + billed).padEnd(11) +
    ("shown " + client.text.length + "/" + N).padEnd(12) + "dupes " + client.dupes);
}

console.log("Answer is " + N + " tokens. Columns: output tokens billed, tokens on screen, duplicates dropped.");
console.log("");
console.log("Case 1: network drops after tick 5, client is back at tick 8");

{ // Tied: the drop aborts generation; the only recovery is to ask again.
  const first = makeJob(), c = makeClient();
  for (let t = 1; t <= 5; t++) { first.step(); c.receive(first.log[first.log.length - 1]); }
  const retry = makeJob(), c2 = makeClient();
  while (retry.log.length < N) { retry.step(); c2.receive(retry.log[retry.log.length - 1]); }
  report("  tied, user re-sends the prompt", retry, c2, first.tokens);
}
{ // Resumable: the job keeps writing while the client is away.
  const job = makeJob(), c = makeClient();
  for (let t = 1; t <= 8; t++) { job.step(); if (t <= 5) c.receive(job.log[job.log.length - 1]); }
  console.log("  client reconnects with Last-Event-ID: " + c.lastId + "; log holds " + job.log.length + " events");
  for (const ev of job.log) c.receive(ev);          // store replays from the start
  while (job.log.length < N) { job.step(); c.receive(job.log[job.log.length - 1]); }
  report("  resumable, replay + live tail", job, c);
  console.log("  text: " + c.text.join(" "));
}

console.log("");
console.log("Case 2: user presses Stop after tick 4");
{
  const job = makeJob(), c = makeClient();
  for (let t = 1; t <= 4; t++) { job.step(); c.receive(job.log[job.log.length - 1]); }
  job.cancelled = true;                             // request abort reaches the provider
  for (let t = 5; t <= N; t++) job.step();
  report("  tied, abort propagated", job, c);
}
{
  const job = makeJob(), c = makeClient();
  for (let t = 1; t <= 4; t++) { job.step(); c.receive(job.log[job.log.length - 1]); }
  for (let t = 5; t <= N; t++) job.step();          // client abort = just a disconnect
  report("  resumable, client abort only", job, c);
}
{
  const job = makeJob(), c = makeClient();
  for (let t = 1; t <= 4; t++) { job.step(); c.receive(job.log[job.log.length - 1]); }
  job.cancelled = true;                             // explicit POST /stop cancels the job
  for (let t = 5; t <= N; t++) job.step();
  report("  resumable, plus a stop endpoint", job, c);
}