Realtime

Realtime

@cosella/realtime provides a typed WebSocket client that multiplexes binary audio frames and JSON text events on a single connection. It handles authentication, reconnection with session resume, backpressure, and liveness detection.

Installation

pnpm add @cosella/realtime

Quick Start

Using the React Hook

import { useCallStream } from '@cosella/realtime';

function LiveCall({ callId, wsUrl }: { callId: string; wsUrl: string }) {
  const {
    connectionState,
    connectionQuality,
    micTranscript,
    loopbackTranscript,
    activeSuggestion,
    suggestionHistory,
    sendAudioFrame,
    sendVadEvent,
    startCall,
    endCall,
  } = useCallStream({
    url: wsUrl,
    getToken: () => auth.getAccessToken(),
    callId,
    enabled: true,
  });

  // Wire audio pipeline output to transport
  useEffect(() => {
    return pipeline.on('frame', ({ channel, seq, captureMs, opus }) => {
      sendAudioFrame(channel, seq, captureMs, opus);
    });
  }, []);

  return (
    <div>
      <p>Connection: {connectionState.phase} ({connectionQuality})</p>
      {activeSuggestion && (
        <p>{activeSuggestion.tokens.join('')}</p>
      )}
    </div>
  );
}

Using the Client Directly

import { RealtimeClient } from '@cosella/realtime';

const client = new RealtimeClient({
  url: 'wss://api.cosella.com/ws',
  getToken: () => auth.getAccessToken(),
  reconnect: { maxRetries: 8, baseDelay: 150, maxDelay: 30_000 },
});

client.onStateChange((state) => console.log('Phase:', state.phase));

client.on('transcript.partial', (e) => {
  console.log(`[${e.channel}] ${e.text}`);
});

client.on('suggestion.partial', (e) => {
  process.stdout.write(e.token); // Typewriter effect
});

client.connect();

Connection States

type ConnectionState =
  | { phase: 'disconnected' }
  | { phase: 'connecting' }
  | { phase: 'authenticating' }
  | { phase: 'awaiting_ready' }
  | { phase: 'active'; callId: string }
  | { phase: 'reconnecting'; attempt: number; nextRetryMs: number }
  | { phase: 'failed'; reason: string };

Phase Gating

sendAudioFrame() and sendVadEvent() silently drop data unless phase === 'active'. This prevents sending audio before the server is ready, during reconnection, or after the call ends. Dropped frames are counted in droppedFrameCount.

Events

Server → Client

Event	Payload	Description
transcript.partial	{ channel, segmentId, text, startMs }	Streaming ASR result
transcript.final	{ channel, segmentId, text, startMs, endMs, confidence }	Finalized transcript
suggestion.partial	{ suggestionId, inReplyTo, token, index }	Streaming LLM token
suggestion.final	{ suggestionId, inReplyTo, text, generatedAt }	Complete suggestion
suggestion.aborted	{ suggestionId, reason }	Superseded/timed out
call.ready	{ callId, epochMs, serverCapabilities }	Server ready for audio
call.state	{ callId, status }	Call lifecycle change
error	{ code, message, recoverable }	Protocol error

Suggestion Lifecycle

The useCallStream hook manages suggestion state automatically:

• suggestion.partial → tokens accumulate in activeSuggestion
• suggestion.final → moves to suggestionHistory (capped at 50)
• suggestion.aborted → clears activeSuggestion
• New partial with different inReplyTo → previous suggestion superseded

Backpressure

An app-level send queue (25 frames per channel) buffers audio between the pipeline and the WebSocket:

• Drop policy: oldest frames dropped first (most recent audio is most useful)
• Flush gate: only sends when ws.bufferedAmount < 16KB
• Connection quality: rolling 5-second dropped-frame rate → 'good' / 'degraded' / 'poor'

Reconnection

On unexpected disconnect:

1. Exponential backoff: 150ms → 300ms → … → 30s cap, with ±50% jitter
2. getToken() called fresh per attempt (handles expired tokens)
3. call.resume sent with last sequence numbers
4. Server responds call.ready (session preserved) or SESSION_NOT_FOUND (fresh call.start)
5. Max 8 attempts before phase: 'failed'

endCall() sends call.end and suppresses reconnection (intentional disconnect).

Binary Frame Encoding

import { encodeAudioFrame, decodeFrameHeader } from '@cosella/realtime';

// Encode: 7-byte big-endian header + opus payload
const frame = encodeAudioFrame('mic', seq, deltaMs, opusData);

// Decode header (for debugging/testing)
const { channel, seq, deltaMs } = decodeFrameHeader(frame);

Testing

The client accepts a wsFactory for dependency injection — no real WebSocket needed:

const mockWs = { onopen: null, onmessage: null, onclose: null, send: vi.fn(), close: vi.fn() };
const client = new RealtimeClient({
  url: 'ws://test',
  getToken: async () => 'token',
  wsFactory: () => mockWs as unknown as WebSocket,
});