Streaming

Streaming delivers a session's activity as it happens, instead of polling for a status. The SDK owns the hard parts (frame parsing, reconnect with backoff, lossless resume, heartbeat liveness) and you supply the socket, because the right way to open one differs between a browser, a terminal, and a server.

import { streamSession } from '@ellipsis-dev/sdk/stream';

Open a socket

openSocket receives the session id and a prebuilt handshake query, and returns an object with four listeners and a close. On a server or in a terminal, authenticate with a bearer header:

import WebSocket from 'ws';
import type { OpenSocket, StreamSocket } from '@ellipsis-dev/sdk/stream';

const token = process.env.ELLIPSIS_API_TOKEN!;

const openSocket: OpenSocket = ({ sessionId, query }): StreamSocket => {
  const url = `wss://api.ellipsis.dev/sessions/${encodeURIComponent(sessionId)}/stream?${query}`;
  const ws = new WebSocket(url, {
    headers: { authorization: `Bearer ${token}` },
  });
  return {
    onOpen: (cb) => ws.on('open', cb),
    onMessage: (cb) => ws.on('message', (raw) => cb(raw.toString())),
    onClose: (cb) => ws.on('close', (code: number) => cb(code)),
    onError: (cb) =>
      ws.on('error', (err: unknown) =>
        cb(err instanceof Error ? err : new Error(String(err)))
      ),
    close: () => ws.close(),
  };
};

Append query to the URL verbatim. It carries the protocol version, which is required, and the resume cursor when reconnecting. Building it yourself means a protocol bump silently stops working.

Stream a session

import { streamSession } from '@ellipsis-dev/sdk/stream';

const outcome = await streamSession({
  sessionId: handle.id,
  openSocket,
  onFrame: (frame) => console.log(frame.type),
});
console.log(outcome.type);

onFrame is called for every frame. streamSession resolves when the session finishes, and the outcome says how: done with the session's final status and exitStatus, error with a message, or aborted if you cancelled. Cancel by passing an AbortSignal:

const controller = new AbortController();
setTimeout(() => controller.abort(), 60_000);

const outcome = await streamSession({
  sessionId: handle.id,
  openSocket,
  onFrame: (frame) => console.log(frame.type),
  signal: controller.signal,
});

Frame types

Seven frame types arrive, typed as StreamFrame. That union deliberately admits an unknown frame shape as well as the seven known ones, so a server that adds a frame type does not break the build:

Ignore frame types you do not recognize, and unknown values inside them. New frames and new values are additive, not a protocol break, so a client that rejects the unfamiliar breaks on a future release.

Switch on frame.type to narrow the union:

onFrame: (frame) => {
  if (frame.type === 'delta') process.stdout.write(frame.text ?? '');
  else if (frame.type === 'session') console.log(frame.session.status);
};

Reconnect and resume

A dropped socket reconnects with capped backoff, up to maxReconnects consecutive failures (5 by default). Resume is cursored on records_append only: the client tracks the highest feed_seq it has seen and asks for everything after it, so no record is delivered twice or lost. snapshot and session frames are whole-state and are resent in full, which is why they never advance the cursor.

Any frame arriving resets the failure count, so a long session that drops occasionally keeps a full reconnect budget each time.

When streaming is not available

Two errors, with different responses:

• StreamUnavailableError means streaming cannot be used here: the endpoint is missing, the server does not support the protocol version, or reconnects are exhausted. Poll with handle.wait() instead.
• StreamAuthError means the credential was rejected for this session. Polling would fail the same way, so this is not a fallback case; fix the credential.

import {
  StreamAuthError,
  StreamUnavailableError,
} from '@ellipsis-dev/sdk/stream';

try {
  await streamSession({ sessionId: handle.id, openSocket, onFrame });
} catch (error) {
  if (error instanceof StreamUnavailableError) await handle.wait();
  else throw error;
}

The stream client speaks protocol version 3 and sends it on every handshake.

Rendering a transcript

Feeding frames into a UI is its own problem: records arrive out of order relative to what you want to draw, and deltas overlay text that a later record replaces. @ellipsis-dev/sdk/store solves it with SessionTranscriptStore, which consumes every frame and maintains renderable state, grouped into chat turns:

import { SessionTranscriptStore } from '@ellipsis-dev/sdk/store';

const store = new SessionTranscriptStore();
await streamSession({
  sessionId: handle.id,
  openSocket,
  onFrame: store.ingest,
});

subscribe and getSnapshot are shaped for React's useSyncExternalStore, but the store is framework-free: any renderer can poll the snapshot. chatTurns() returns the record log grouped into turns, which is what a chat-style transcript renders.