API index and search · Build metadata
Source snapshot
packages/react/src/stream.ts
1// `useStreamedText` — render an LM response that is arriving on two planes2// (designs-implemented/LM-STREAM-CHECKPOINT-DESIGN.md).3//4// The durable plane already flows through IVM like any other data: the app's chat query carries the5// message row and its un-compacted chunk rows, and `assembleDurableText` turns those into the text6// the store holds. This hook adds the LIVE plane — the tail that has not been checkpointed yet — and7// merges the two with `spliceStreamText`.8//9// It exists because the merge is one line but the SUBSCRIPTION has five traps, and every app would10// otherwise have to find them independently. Each is marked TRAP below:11//12// 1. the join offset must be read WITHOUT making the durable text a dependency (it advances on13// every checkpoint, and a dependency would tear the connection down and rebuild it);14// 2. the accumulator must be SEEDED with the text it joined at (the splice compares lengths, so a15// tail carrying only the frames it received reads as shorter than the durable text and is16// discarded);17// 3. the reader must close its own `EventSource` on a terminal frame (`EventSource` reconnects on18// ANY close, including a clean one);19// 4. the accumulated tail must be keyed to its stream (a tail left over from the previous message20// is a PREFIX of nothing, and for one render it can be longer than the new durable text and win21// the splice);22// 5. a chunk must be spliced at ITS OWN offset, never blindly appended: an `EventSource` reconnect23// resumes at the last id it saw, and a `durable` frame's id sits at the STORE's position, which24// trails the chunk progress — so a resumed replay can start BEHIND the accumulator, and an25// append would duplicate the overlap into the rendered text.26//27// Losing the live plane is not an error: `absent` and `stale` mean "the store is the whole truth28// now", and the hook simply stops accumulating. The rendered text stays correct — it just advances at29// checkpoint granularity, which is what a reader on the wrong instance or an old runtime gets anyway.3031import { useEffect, useRef, useState } from "react";32import { spliceStreamText } from "@rindle/client";33import type { StreamFrame } from "@rindle/client";3435/** Where {@link useStreamedText} subscribes by default — `DEFAULT_RINDLE_API_ROUTES.stream`, mirrored36 * here rather than imported so the browser never pulls `@rindle/api-server`. */37export const DEFAULT_STREAM_ENDPOINT = "/api/rindle/stream";3839/**40 * How the live plane is reached. The default ({@link eventSourceTransport}) is SSE, which is what the41 * api-server's `streamFramesToSse` serves and what gets `Last-Event-ID` resume for free. Supply your42 * own for a WebSocket, a fetch-stream, or a test.43 */44export interface StreamTransport {45 /** Attach at `url` and call `onFrame` per decoded frame. MUST return a detach function; it may be46 * called more than once and must tolerate that. `onFrame` may be called synchronously. */47 subscribe(url: string, onFrame: (frame: StreamFrame) => void): () => void;48}4950export interface UseStreamedTextInput {51 /** The stream's id — the message row's key. Changing it drops the old tail and rejoins. */52 streamId: string;53 /** What the IVM view shows: `assembleDurableText(message, message.chunks)`. Read from a ref54 * internally (TRAP 1), so it may change every checkpoint without disturbing the connection. */55 durable: string;56 /** Whether a producer is still running — the app's own read of its status column (typically57 * `status === "streaming" || status === "pending"`). The live leg attaches only while true. */58 live: boolean;59}6061export interface UseStreamedTextOptions {62 /** Default {@link DEFAULT_STREAM_ENDPOINT}. `streamId` and `from` are appended as query params. */63 endpoint?: string;64 /** Default {@link eventSourceTransport}. Read at subscribe time, NOT a dependency — an inline65 * literal would otherwise reconnect on every render. */66 transport?: StreamTransport;67 /** A frame that could not be decoded, or a transport-level error. The durable plane is unaffected,68 * so this is a diagnostic, not a failure. */69 onError?: (err: unknown) => void;70}7172/** `<endpoint>?streamId=…&from=…`. `from` is the join offset; a reconnecting `EventSource` overrides73 * it with its own `Last-Event-ID` header, which the server prefers. */74export function streamSubscribeUrl(endpoint: string, streamId: string, from: number): string {75 const sep = endpoint.includes("?") ? "&" : "?";76 return `${endpoint}${sep}streamId=${encodeURIComponent(streamId)}&from=${from}`;77}7879/** The default SSE transport. Absent `EventSource` (SSR, an older runtime, a test without jsdom) it80 * attaches nothing and the reader stays on the durable plane — correct, just chunkier. */81export function eventSourceTransport(onError?: (err: unknown) => void): StreamTransport {82 return {83 subscribe(url, onFrame) {84 const Ctor = (globalThis as { EventSource?: new (url: string) => EventSourceLike }).EventSource;85 if (!Ctor) return () => {};86 let es: EventSourceLike;87 try {88 es = new Ctor(url);89 } catch (err) {90 onError?.(err);91 return () => {};92 }93 es.onmessage = (event: { data: string }) => {94 try {95 onFrame(JSON.parse(event.data) as StreamFrame);96 } catch (err) {97 onError?.(err);98 }99 };100 // A transport-level error is not a stream error: `EventSource` retries on its own, and the101 // durable plane keeps the reader correct meanwhile.102 es.onerror = (event: unknown) => onError?.(event);103 return () => es.close();104 },105 };106}107108interface EventSourceLike {109 onmessage: ((event: { data: string }) => void) | null;110 onerror: ((event: unknown) => void) | null;111 close(): void;112}113114/**115 * The response text as it should be rendered right now: the durable prefix spliced with the live116 * tail.117 *118 * ```tsx119 * const data = useFragment(MessageFragment, message);120 * const text = useStreamedText({121 * streamId: data.id,122 * durable: assembleDurableText(data, data.chunks),123 * live: data.status === "streaming" || data.status === "pending",124 * });125 * ```126 *127 * The value is monotone in practice — it only grows while a stream runs — and when the closing128 * checkpoint compacts the chunks into the body it returns the identical string, so there is no129 * flicker at the handoff.130 */131export function useStreamedText(132 { streamId, durable, live }: UseStreamedTextInput,133 options: UseStreamedTextOptions = {},134): string {135 const { endpoint = DEFAULT_STREAM_ENDPOINT } = options;136137 // TRAP 1: the join offset is read from a ref at subscribe time. Making `durable` a dependency would138 // reconnect on every checkpoint — a fresh HTTP request every ~512 characters.139 const durableRef = useRef(durable);140 durableRef.current = durable;141 // Read at subscribe time for the same reason: callers pass these inline.142 const optionsRef = useRef(options);143 optionsRef.current = options;144145 // TRAP 4: the tail is keyed to its stream and compared during RENDER, not reset in an effect.146 // Resetting in an effect would leave one render where the previous message's (longer) tail wins the147 // splice and briefly renders the wrong message's text.148 const [tail, setTail] = useState<{ streamId: string; text: string }>({ streamId, text: "" });149 const produced = tail.streamId === streamId ? tail.text : "";150151 useEffect(() => {152 if (!live) return;153 const { transport, onError } = optionsRef.current;154 // TRAP 2: seed the accumulator with the text we are joining at, so its LENGTH is comparable to155 // the durable text's.156 let acc = durableRef.current;157 // `detach` is assigned by `subscribe` itself, and a transport may deliver frames (even terminal158 // ones) synchronously from inside that call — so completion is latched and applied after.159 let detach: (() => void) | undefined;160 let done = false;161 const finish = (): void => {162 done = true;163 detach?.();164 };165166 const active = transport ?? eventSourceTransport(onError);167 detach = active.subscribe(streamSubscribeUrl(endpoint, streamId, acc.length), (frame) => {168 if (done) return;169 switch (frame.type) {170 case "chunk":171 // TRAP 5: splice at the frame's own offset. A resumed transport can replay a span that172 // starts BEHIND the accumulator (both are prefixes of one response, so cutting at `from`173 // and appending is exact); a span that starts AHEAD would be a gap — detach and let the174 // durable plane carry the reader, which stays correct at checkpoint granularity.175 if (frame.from > acc.length) {176 optionsRef.current.onError?.(177 new Error(`stream ${streamId}: chunk at ${frame.from} leaves a gap after ${acc.length}`),178 );179 finish();180 break;181 }182 acc = acc.slice(0, frame.from) + frame.text;183 setTail({ streamId, text: acc });184 break;185 case "end":186 case "stale":187 case "absent":188 // TRAP 3. On `stale`/`absent` the durable plane already carries everything, and on `end` the189 // stream is sealed — in all three cases another connection would be pure waste.190 finish();191 break;192 default:193 // `open` and `durable` carry no text. `durable` is a durability signal the renderer does194 // not need: the splice is length-based, so the handoff needs no announcement.195 break;196 }197 });198 if (done) detach();199200 return finish;201 // `durable` is deliberately absent (TRAP 1); `transport`/`onError` are read from the ref.202 }, [streamId, live, endpoint]);203204 return spliceStreamText(durable, produced);205}206