fix(core): keep the cached text stream observable across a room disconnect

.changeset/olive-jokes-shave.md

@@ -0,0 +1,5 @@
+---
+"@livekit/components-core": patch
+---
+
+fix(core): key the text stream observable cache on the room instance so it survives a disconnect

packages/core/src/components/textStream.test.ts

@@ -0,0 +1,109 @@
+import { RoomEvent, type Room } from 'livekit-client';
+import { describe, expect, it, vi } from 'vitest';
+import { setupTextStream } from './textStream';
+
+/**
+ * Minimal stand-in for `Room` that mirrors livekit-client's one-handler-per-topic
+ * contract: registering the same topic twice throws.
+ */
+function createFakeRoom() {
+  const handlers = new Map<string, (reader: unknown, participantInfo: unknown) => Promise<void>>();
+  const disconnectListeners: Array<() => void> = [];
+
+  const registerTextStreamHandler = vi.fn(
+    (topic: string, handler: (reader: unknown, participantInfo: unknown) => Promise<void>) => {
+      if (handlers.has(topic)) {
+        throw new Error(`A text stream handler for topic "${topic}" has already been set.`);
+      }
+      handlers.set(topic, handler);
+    },
+  );
+  const unregisterTextStreamHandler = vi.fn((topic: string) => {
+    handlers.delete(topic);
+  });
+
+  const room = {
+    registerTextStreamHandler,
+    unregisterTextStreamHandler,
+    on: (event: RoomEvent, listener: () => void) => {
+      if (event === RoomEvent.Disconnected) disconnectListeners.push(listener);
+      return room;
+    },
+  } as unknown as Room;
+
+  return {
+    room,
+    registerTextStreamHandler,
+    disconnect: () => disconnectListeners.forEach((listener) => listener()),
+    push: (topic: string, reader: unknown) => handlers.get(topic)?.(reader, { identity: 'agent' }),
+  };
+}
+
+function fakeReader(id: string, chunks: string[]) {
+  return {
+    info: { id, attributes: {} },
+    async *[Symbol.asyncIterator]() {
+      for (const chunk of chunks) yield chunk;
+    },
+  };
+}
+
+/** `from(reader)` walks the async iterator, so emissions land a few microtasks later. */
+const flush = () => new Promise((resolve) => setTimeout(resolve, 0));
+
+const TOPIC = 'lk.transcription';
+
+describe('setupTextStream', () => {
+  it('reuses one observable per room and topic across disconnect/reconnect', () => {
+    const { room, registerTextStreamHandler, disconnect } = createFakeRoom();
+
+    // A consumer subscribes while connected, then the room disconnects and
+    // `useTextStream` drops the subscription.
+    const first = setupTextStream(room, TOPIC);
+    first.subscribe().unsubscribe();
+    disconnect();
+    registerTextStreamHandler.mockClear();
+
+    // A consumer that mounts *while disconnected* must get the same observable.
+    // Two observables for one topic both register it on reconnect, and the
+    // second call throws `A text stream handler ... has already been set`.
+    const second = setupTextStream(room, TOPIC);
+
+    const subA = first.subscribe();
+    const subB = second.subscribe();
+
+    expect(registerTextStreamHandler).toHaveBeenCalledTimes(1);
+    expect(second).toBe(first);
+
+    subA.unsubscribe();
+    subB.unsubscribe();
+  });
+
+  it('caches per room instance, so a second room gets its own observable', () => {
+    const a = createFakeRoom();
+    const b = createFakeRoom();
+
+    expect(setupTextStream(a.room, TOPIC)).not.toBe(setupTextStream(b.room, TOPIC));
+  });
+
+  it('starts each subscription window with an empty buffer', async () => {
+    const { room, push } = createFakeRoom();
+
+    const stream = setupTextStream(room, TOPIC);
+    const before: number[] = [];
+    const firstSub = stream.subscribe((streams) => before.push(streams.length));
+    await push(TOPIC, fakeReader('stream-1', ['hello']));
+    await flush();
+    expect(before.at(-1)).toBe(1);
+    firstSub.unsubscribe();
+
+    // Reconnect: the buffer from the previous window must not leak into this one.
+    const after: number[] = [];
+    const secondSub = stream.subscribe((streams) => after.push(streams.length));
+    await push(TOPIC, fakeReader('stream-2', ['world']));
+    await flush();
+
+    expect(after).toEqual([1]);
+    secondSub.unsubscribe();
+  });
+});

packages/core/src/components/textStream.ts

@@ -1,4 +1,4 @@
-import { RoomEvent, type Room, type TextStreamInfo } from 'livekit-client';
+import { type Room, type TextStreamInfo } from 'livekit-client';
 import { from, scan, Subject, type Observable } from 'rxjs';
 import { share, tap } from 'rxjs/operators';
 import { ParticipantAgentAttributes } from '../helper';
@@ -9,47 +9,25 @@ export interface TextStreamData {
   streamInfo: TextStreamInfo;
 }
 
-// Singleton getters for lazy initialization
-let observableCacheInstance: Map<string, Observable<TextStreamData[]>> | null = null;
-let roomInstanceMapInstance: WeakMap<Room, string> | null = null;
-let nextRoomId = 0;
+// One observable per room and topic. The outer map is weak so a room that is no
+// longer referenced takes its observables with it, while a reused room keeps
+// them across connect/disconnect cycles.
+const observableCache = new WeakMap<Room, Map<string, Observable<TextStreamData[]>>>();
 
-// Get or create the observable cache
-function getObservableCache(): Map<string, Observable<TextStreamData[]>> {
-  if (!observableCacheInstance) {
-    observableCacheInstance = new Map<string, Observable<TextStreamData[]>>();
+function getTopicCache(room: Room): Map<string, Observable<TextStreamData[]>> {
+  let topicCache = observableCache.get(room);
+  if (!topicCache) {
+    topicCache = new Map<string, Observable<TextStreamData[]>>();
+    observableCache.set(room, topicCache);
   }
-  return observableCacheInstance;
-}
-
-// Get or create the room instance map
-function getRoomInstanceMap(): WeakMap<Room, string> {
-  if (!roomInstanceMapInstance) {
-    roomInstanceMapInstance = new WeakMap<Room, string>();
-  }
-  return roomInstanceMapInstance;
-}
-
-// Helper to generate cache key
-function getCacheKey(room: Room, topic: string): string {
-  const instanceMap = getRoomInstanceMap();
-
-  // Get or create a unique ID for this room instance
-  let roomId = instanceMap.get(room);
-  if (!roomId) {
-    roomId = `room_${nextRoomId++}`;
-    instanceMap.set(room, roomId);
-  }
-
-  return `${roomId}:${topic}`;
+  return topicCache;
 }
 
 export function setupTextStream(room: Room, topic: string): Observable<TextStreamData[]> {
-  const cacheKey = getCacheKey(room, topic);
-  const observableCache = getObservableCache();
+  const topicCache = getTopicCache(room);
 
   // Check if we already have an observable for this room and topic
-  const existingObservable = observableCache.get(cacheKey);
+  const existingObservable = topicCache.get(topic);
   if (existingObservable) {
     return existingObservable;
   }
@@ -63,6 +41,10 @@ export function setupTextStream(room: Room, topic: string): Observable<TextStrea
   const sharedObservable = textStreamsSubject.pipe(
     tap({
       subscribe: () => {
+        // `share()` resets on refcount zero, so this runs once per subscription
+        // window: on the first subscriber, and again after every reconnect on a
+        // reused room. Each window starts from an empty buffer.
+        textStreams = [];
         room.registerTextStreamHandler(topic, async (reader, participantInfo) => {
           // Create an observable from the reader
           const streamObservable = from(reader).pipe(
@@ -118,14 +100,7 @@ export function setupTextStream(room: Room, topic: string): Observable<TextStrea
     share(),
   );
 
-  observableCache.set(cacheKey, sharedObservable);
-
-  // Add cleanup when room is disconnected
-  room.on(RoomEvent.Disconnected, () => {
-    getObservableCache().delete(cacheKey);
-    textStreams = [];
-    textStreamsSubject.next([]);
-  });
+  topicCache.set(topic, sharedObservable);
 
   return sharedObservable;
 }