feat(realtime): land LiveKit dead code + LK-based subscribe (PR 2 of 3)

packages/sdk/package.json

@@ -56,6 +56,7 @@
     "vitest": "^4.0.18"
   },
   "dependencies": {
+    "livekit-client": "^2.0.0",
     "mitt": "^3.0.1",
     "p-retry": "^6.2.1",
     "zod": "^4.0.17"

packages/sdk/src/index.ts

@@ -2,6 +2,7 @@ import { z } from "zod";
 import { createProcessClient } from "./process/client";
 import { createQueueClient } from "./queue/client";
 import { createRealTimeClient } from "./realtime/client";
+import { createRealTimeSubscribeClient } from "./realtime/subscribe-client";
 import { createTokensClient } from "./tokens/client";
 import { readEnv } from "./utils/env";
 import { createInvalidApiKeyError, createInvalidBaseUrlError } from "./utils/errors";
@@ -45,7 +46,7 @@ export type {
   SubscribeEvents,
   SubscribeOptions,
 } from "./realtime/subscribe-client";
-export type { ConnectionState } from "./realtime/types";
+export type { ConnectionState, GenerationEndedMessage, QueuePosition, QueuePositionMessage } from "./realtime/types";
 export {
   type CanonicalModel,
   type CustomModelDefinition,
@@ -194,13 +195,21 @@ export const createDecartClient = (options: DecartClientOptions = {}) => {
   // Proxy mode is only for HTTP endpoints (process, queue, tokens)
   // Note: Realtime will fail at connection time if no API key is provided
   const wsBaseUrl = parsedOptions.data.realtimeBaseUrl || "wss://api3.decart.ai";
-  const realtime = createRealTimeClient({
+  const realtimePublish = createRealTimeClient({
     baseUrl: wsBaseUrl,
     apiKey: apiKey || "",
     integration,
     logger,
     telemetryEnabled,
   });
+  const hasExplicitBaseUrl = isProxyMode || parsedOptions.data.baseUrl !== undefined;
+  const subscribeBaseUrl = hasExplicitBaseUrl ? baseUrl : wsBaseUrl.replace(/^wss?:\/\//i, "https://");
+  const realtimeSubscribe = createRealTimeSubscribeClient({
+    baseUrl: subscribeBaseUrl,
+    apiKey: apiKey || "",
+    integration,
+    logger,
+  });
 
   const process = createProcessClient({
     baseUrl,
@@ -221,7 +230,10 @@ export const createDecartClient = (options: DecartClientOptions = {}) => {
   });
 
   return {
-    realtime,
+    realtime: {
+      connect: realtimePublish.connect,
+      subscribe: realtimeSubscribe.subscribe,
+    },
     /**
      * Client for synchronous image generation.
      * Only image models support the sync/process API.

packages/sdk/src/realtime/client.ts

@@ -9,16 +9,38 @@ import { createMirroredStream, type MirroredStream, shouldMirrorTrack } from "./
 import type { DiagnosticEvent } from "./observability/diagnostics";
 import { RealtimeObservability } from "./observability/realtime-observability";
 import type { WebRTCStats } from "./observability/webrtc-stats";
-import {
-  decodeSubscribeToken,
-  encodeSubscribeToken,
-  type RealTimeSubscribeClient,
-  type SubscribeEvents,
-  type SubscribeOptions,
-} from "./subscribe-client";
+import type { RealTimeSubscribeClient, SubscribeEvents, SubscribeOptions } from "./subscribe-client";
 import type { ConnectionState, GenerationTickMessage, SessionIdMessage } from "./types";
 import { WebRTCManager } from "./webrtc-manager";
 
+type AiortcTokenPayload = {
+  sid: string;
+  ip: string;
+  port: number;
+};
+
+function encodeSubscribeToken(sid: string, ip: string, port: number): string {
+  return btoa(JSON.stringify({ sid, ip, port }));
+}
+
+function decodeSubscribeToken(token: string): AiortcTokenPayload {
+  try {
+    const payload = JSON.parse(atob(token)) as Partial<AiortcTokenPayload>;
+    if (
+      !payload.sid ||
+      typeof payload.sid !== "string" ||
+      !payload.ip ||
+      typeof payload.ip !== "string" ||
+      typeof payload.port !== "number"
+    ) {
+      throw new Error("Invalid subscribe token format");
+    }
+    return { sid: payload.sid, ip: payload.ip, port: payload.port };
+  } catch {
+    throw new Error("Invalid subscribe token");
+  }
+}
+
 async function blobToBase64(blob: Blob): Promise<string> {
   return new Promise((resolve, reject) => {
     const reader = new FileReader();
@@ -285,7 +307,6 @@ export const createRealTimeClient = (opts: RealTimeClientOptions) => {
       integration,
       logger,
       onDiagnostic: (event) => emitOrBuffer("diagnostic", event as SubscribeEvents["diagnostic"]),
-      onStats: opts.telemetryEnabled ? (stats) => emitOrBuffer("stats", stats) : undefined,
     });
     observability.sessionStarted(sid);
 

packages/sdk/src/realtime/config-realtime.ts

@@ -0,0 +1,46 @@
+export const REALTIME_CONFIG = {
+  signaling: {
+    connectTimeoutMs: 60_000,
+    handshakeTimeoutMs: 15_000,
+    requestTimeoutMs: 30_000,
+  },
+  session: {
+    connectionTimeoutMs: 60_000 * 5,
+    retry: {
+      retries: 5,
+      factor: 2,
+      minTimeout: 1_000,
+      maxTimeout: 10_000,
+    },
+    permanentErrorSubstrings: [
+      "permission denied",
+      "not allowed",
+      "invalid session",
+      "401",
+      "invalid api key",
+      "unauthorized",
+    ],
+  },
+  methods: {
+    promptTimeoutMs: 15_000,
+    updateTimeoutMs: 30_000,
+  },
+  livekit: {
+    inferenceServerIdentityPrefix: "inference-server-",
+    roomOptions: {
+      adaptiveStream: false,
+      dynacast: false,
+    },
+    defaultVideoCodec: "h264",
+    defaultMaxVideoBitrateBps: 3_500_000,
+    defaultPublishFps: 30,
+  },
+  observability: {
+    stallFpsThreshold: 0.5,
+    statsDefaultIntervalMs: 1_000,
+    statsMinIntervalMs: 500,
+    telemetryReportIntervalMs: 10_000,
+    telemetryUrl: "https://platform.decart.ai/api/v1/telemetry",
+    telemetryMaxItemsPerReport: 120,
+  },
+} as const;

packages/sdk/src/realtime/media-channel.ts

@@ -0,0 +1,124 @@
+import {
+  type DisconnectReason,
+  type RemoteParticipant,
+  type RemoteTrack,
+  Room,
+  RoomEvent,
+  Track,
+  TrackEvent,
+  type TrackPublishOptions,
+} from "livekit-client";
+import mitt, { type Emitter } from "mitt";
+
+import { createConsoleLogger, type Logger } from "../utils/logger";
+import { REALTIME_CONFIG } from "./config-realtime";
+import type { RealtimeObservability } from "./observability/realtime-observability";
+
+export function getDefaultVideoPublishOptions(): TrackPublishOptions {
+  const videoEncoding = {
+    maxBitrate: REALTIME_CONFIG.livekit.defaultMaxVideoBitrateBps,
+    maxFramerate: REALTIME_CONFIG.livekit.defaultPublishFps,
+  };
+
+  return {
+    source: Track.Source.Camera,
+    videoCodec: REALTIME_CONFIG.livekit.defaultVideoCodec,
+    simulcast: true,
+    videoEncoding,
+  };
+}
+
+export type MediaChannelEvents = {
+  remoteStream: MediaStream;
+  firstFrame: undefined;
+  disconnected: { reason?: DisconnectReason };
+};
+
+export interface MediaChannelConfig {
+  observability?: RealtimeObservability;
+  localStream: MediaStream | null;
+  logger?: Logger;
+}
+
+export type MediaConnectOptions = {
+  url: string;
+  token: string;
+};
+
+export class MediaChannel {
+  private room: Room | null = null;
+  private remoteStream: MediaStream | null = null;
+  private events: Emitter<MediaChannelEvents> = mitt();
+  private readonly logger: Logger;
+
+  constructor(private readonly config: MediaChannelConfig) {
+    this.logger = config.logger ?? createConsoleLogger("warn");
+  }
+
+  get localStream(): MediaStream | null {
+    return this.config.localStream;
+  }
+
+  on<E extends keyof MediaChannelEvents>(event: E, handler: (data: MediaChannelEvents[E]) => void): void {
+    this.events.on(event, handler);
+  }
+
+  off<E extends keyof MediaChannelEvents>(event: E, handler: (data: MediaChannelEvents[E]) => void): void {
+    this.events.off(event, handler);
+  }
+
+  async connect(opts: MediaConnectOptions): Promise<void> {
+    this.room ??= new Room(REALTIME_CONFIG.livekit.roomOptions);
+    const room = this.room;
+
+    room.on(RoomEvent.TrackSubscribed, (track: RemoteTrack, _pub, participant: RemoteParticipant) => {
+      if (!participant.identity.startsWith(REALTIME_CONFIG.livekit.inferenceServerIdentityPrefix)) return;
+      if (track.kind !== Track.Kind.Video && track.kind !== Track.Kind.Audio) return;
+
+      track.attach();
+      const mediaStreamTrack = track.mediaStreamTrack;
+      if (mediaStreamTrack) {
+        this.remoteStream ??= new MediaStream();
+        if (!this.remoteStream.getTracks().includes(mediaStreamTrack)) {
+          this.remoteStream.addTrack(mediaStreamTrack);
+        }
+        this.events.emit("remoteStream", this.remoteStream);
+      }
+      track.on(TrackEvent.VideoPlaybackStarted, () => {
+        this.events.emit("firstFrame");
+      });
+    });
+
+    room.on(RoomEvent.Disconnected, (reason?: DisconnectReason) => {
+      this.logger.warn("livekit: room disconnected", { reason });
+      this.events.emit("disconnected", { reason });
+    });
+
+    await room.connect(opts.url, opts.token);
+    if (this.config.localStream) {
+      await this.publishLocalTracks(this.config.localStream);
+    }
+    this.config.observability?.setLiveKitRoom(room);
+  }
+
+  disconnect(): void {
+    const room = this.room;
+    this.room = null;
+    this.remoteStream = null;
+    this.config.observability?.setLiveKitRoom(null);
+    if (room) {
+      room.disconnect().catch(() => {});
+    }
+  }
+
+  private async publishLocalTracks(stream: MediaStream): Promise<void> {
+    if (!this.room) return;
+    for (const track of stream.getTracks()) {
+      if (track.kind === "video") {
+        await this.room.localParticipant.publishTrack(track, getDefaultVideoPublishOptions());
+      } else {
+        await this.room.localParticipant.publishTrack(track);
+      }
+    }
+  }
+}

packages/sdk/src/realtime/observability/livekit-stats-provider.ts

@@ -0,0 +1,41 @@
+import type { LocalTrack, RemoteTrack, Room } from "livekit-client";
+import type { StatsProvider } from "./webrtc-stats";
+
+export function createLiveKitStatsProvider(room: Room): StatsProvider {
+  let uid = 0;
+
+  const collectFromTrack = async (
+    track: LocalTrack | RemoteTrack | undefined,
+    entries: Array<[string, unknown]>,
+  ): Promise<void> => {
+    if (!track) return;
+    let report: RTCStatsReport | undefined;
+    try {
+      report = await track.getRTCStatsReport();
+    } catch {
+      return;
+    }
+    if (!report) return;
+    report.forEach((stat, id) => {
+      entries.push([`${id}#${uid++}`, stat]);
+    });
+  };
+
+  return {
+    async getStats(): Promise<RTCStatsReport> {
+      const entries: Array<[string, unknown]> = [];
+
+      for (const pub of room.localParticipant.trackPublications.values()) {
+        await collectFromTrack(pub.track, entries);
+      }
+
+      for (const participant of room.remoteParticipants.values()) {
+        for (const pub of participant.trackPublications.values()) {
+          await collectFromTrack(pub.track as RemoteTrack | undefined, entries);
+        }
+      }
+
+      return new Map(entries) as unknown as RTCStatsReport;
+    },
+  };
+}

packages/sdk/src/realtime/observability/realtime-observability.ts

@@ -1,10 +1,11 @@
+import type { Room } from "livekit-client";
 import type { Logger } from "../../utils/logger";
+import { REALTIME_CONFIG } from "../config-realtime";
 import type { DiagnosticEvent, DiagnosticEventName, DiagnosticEvents } from "./diagnostics";
+import { createLiveKitStatsProvider } from "./livekit-stats-provider";
 import { type ITelemetryReporter, NullTelemetryReporter, TelemetryReporter } from "./telemetry-reporter";
 import { type StatsProvider, type WebRTCStats, WebRTCStatsCollector } from "./webrtc-stats";
 
-const STALL_FPS_THRESHOLD = 0.5;
-
 export type RealtimeObservabilityOptions = {
   telemetryEnabled: boolean;
   apiKey: string;
@@ -27,6 +28,7 @@ export class RealtimeObservability {
   private pendingTelemetryDiagnostics: PendingTelemetryDiagnostic[] = [];
   private statsCollector: WebRTCStatsCollector | null = null;
   private statsCollectorSource: StatsProvider | null = null;
+  private liveKitRoom: Room | null = null;
   private videoStalled = false;
   private stallStartMs = 0;
 
@@ -75,6 +77,7 @@ export class RealtimeObservability {
     }
 
     this.stopStats();
+    this.resetStallDetection();
     this.statsCollectorSource = source;
 
     if (!this.options.telemetryEnabled && !this.options.onStats) {
@@ -85,10 +88,26 @@ export class RealtimeObservability {
     this.statsCollector.start(source, (stats) => this.handleStats(stats));
   }
 
+  setLiveKitRoom(room: Room | null): void {
+    if (!room) {
+      this.liveKitRoom = null;
+      this.setStatsProvider(null);
+      return;
+    }
+
+    if (room === this.liveKitRoom) {
+      return;
+    }
+
+    this.liveKitRoom = room;
+    this.setStatsProvider(createLiveKitStatsProvider(room));
+  }
+
   stopStats(): void {
     this.statsCollector?.stop();
     this.statsCollector = null;
     this.statsCollectorSource = null;
+    this.liveKitRoom = null;
     this.resetStallDetection();
   }
 
@@ -108,11 +127,11 @@ export class RealtimeObservability {
 
   private detectVideoStall(stats: WebRTCStats): void {
     const fps = stats.video?.framesPerSecond ?? 0;
-    if (!this.videoStalled && stats.video && fps < STALL_FPS_THRESHOLD) {
+    if (!this.videoStalled && stats.video && fps < REALTIME_CONFIG.observability.stallFpsThreshold) {
       this.videoStalled = true;
       this.stallStartMs = Date.now();
       this.diagnostic("videoStall", { stalled: true, durationMs: 0 }, this.stallStartMs);
-    } else if (this.videoStalled && fps >= STALL_FPS_THRESHOLD) {
+    } else if (this.videoStalled && fps >= REALTIME_CONFIG.observability.stallFpsThreshold) {
       const durationMs = Date.now() - this.stallStartMs;
       this.videoStalled = false;
       this.diagnostic("videoStall", { stalled: false, durationMs });