Initial implementation for transcription deduplication

.changeset/transcription-back-conversion.md

@@ -0,0 +1,20 @@
+---
+'livekit-client': minor
+---
+
+Rebuild transcription events from `lk.transcription` data streams and advertise client protocol 3
+
+`RoomEvent.TranscriptionReceived` (and the matching `ParticipantEvent` / `TrackEvent`) are now
+sourced from the `lk.transcription` text stream channel rather than legacy `Transcription` data
+packets, which are ignored from this release on. The event signature is unchanged.
+
+**This is a behavior change that takes effect immediately, not once agents adopt protocol 3.**
+Agents currently publish both channels; this release reads the stream channel and drops the legacy
+one. Advertising client protocol 3 additionally lets agents stop publishing the legacy copy
+altogether, roughly halving reliable-channel traffic for transcriptions — a significant improvement
+on constrained uplinks.
+
+Applications reading `lk.transcription` directly via `registerTextStreamHandler` are unaffected: the
+SDK observes the topic internally without taking it over. Any non-agent publisher of legacy
+`Transcription` packets (a bespoke service calling `publish_transcription`, or a pre-1.0 agents
+framework) no longer surfaces.

src/index.ts

@@ -110,6 +110,7 @@ export type {
   ByteStreamInfo,
 } from './room/types';
 export * from './version';
+export { ParticipantAgentAttributes } from './room/participant/attributes';
 export {
   /** @internal */
   attributes,

src/room/Room.test.ts

@@ -1,18 +1,27 @@
 import {
   ClientInfo_Capability,
+  DataPacket,
+  DataStream_Chunk,
+  DataStream_Header,
+  DataStream_TextHeader,
+  DataStream_Trailer,
+  Encryption_Type,
   JoinResponse,
   StreamState as ProtoStreamState,
   StreamStateUpdate,
   SubscriptionError,
   SubscriptionResponse,
   TrackInfo,
+  TrackSource,
   TrackType,
+  Transcription,
+  TranscriptionSegment as TranscriptionSegmentModel,
 } from '@livekit/protocol';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 import MockMediaStreamTrack from '../test/MockMediaStreamTrack';
 import Room, { ConnectionState } from './Room';
 import { roomConnectOptionDefaults, roomOptionDefaults } from './defaults';
-import { EngineEvent, ParticipantEvent, RoomEvent } from './events';
+import { EngineEvent, ParticipantEvent, RoomEvent, TrackEvent } from './events';
 import RemoteParticipant from './participant/RemoteParticipant';
 import RemoteTrackPublication from './track/RemoteTrackPublication';
 import RemoteVideoTrack from './track/RemoteVideoTrack';
@@ -321,3 +330,173 @@ describe('stream state updates', () => {
     expect(participantEvents).not.toHaveBeenCalled();
   });
 });
+
+describe('transcription back-conversion', () => {
+  const agentIdentity = 'agent-1';
+  const trackSid = 'TR_mic';
+
+  /** A connected room with one remote participant publishing a microphone track. */
+  function setupRoom() {
+    const room = new Room();
+    room.state = ConnectionState.Connected;
+    (
+      room as unknown as { incomingDataStreamManager: { setConnected: (c: boolean) => void } }
+    ).incomingDataStreamManager.setConnected(true);
+
+    const participant = new RemoteParticipant(room.engine.client, 'PA_agent', agentIdentity);
+    const publication = new RemoteTrackPublication(
+      Track.Kind.Audio,
+      new TrackInfo({
+        sid: trackSid,
+        type: TrackType.AUDIO,
+        name: 'roomio_audio',
+        source: TrackSource.MICROPHONE,
+      }),
+      true,
+    );
+    participant.trackPublications.set(trackSid, publication);
+    (
+      room as unknown as { remoteParticipants: Map<string, RemoteParticipant> }
+    ).remoteParticipants.set(agentIdentity, participant);
+
+    return { room, participant, publication };
+  }
+
+  function pushPacket(room: Room, packet: DataPacket) {
+    (
+      room as unknown as {
+        handleDataPacket: (packet: DataPacket, encryptionType: Encryption_Type) => void;
+      }
+    ).handleDataPacket(packet, Encryption_Type.NONE);
+  }
+
+  /** Publishes a complete single-chunk `lk.transcription` stream from `senderIdentity`. */
+  function pushTranscriptionStream(
+    room: Room,
+    senderIdentity: string,
+    text: string,
+    attributes: Record<string, string>,
+  ) {
+    const streamId = crypto.randomUUID();
+    pushPacket(
+      room,
+      new DataPacket({
+        participantIdentity: senderIdentity,
+        value: {
+          case: 'streamHeader',
+          value: new DataStream_Header({
+            streamId,
+            topic: 'lk.transcription',
+            mimeType: 'text/plain',
+            timestamp: 0n,
+            attributes,
+            contentHeader: { case: 'textHeader', value: new DataStream_TextHeader({}) },
+          }),
+        },
+      }),
+    );
+    pushPacket(
+      room,
+      new DataPacket({
+        participantIdentity: senderIdentity,
+        value: {
+          case: 'streamChunk',
+          value: new DataStream_Chunk({
+            streamId,
+            chunkIndex: 0n,
+            content: new TextEncoder().encode(text),
+          }),
+        },
+      }),
+    );
+    pushPacket(
+      room,
+      new DataPacket({
+        participantIdentity: senderIdentity,
+        value: {
+          case: 'streamTrailer',
+          value: new DataStream_Trailer({ streamId }),
+        },
+      }),
+    );
+  }
+
+  it('ignores legacy Transcription data packets', () => {
+    const { room } = setupRoom();
+    const received: Array<unknown> = [];
+    room.on(RoomEvent.TranscriptionReceived, (segments) => received.push(segments));
+
+    pushPacket(
+      room,
+      new DataPacket({
+        participantIdentity: agentIdentity,
+        value: {
+          case: 'transcription',
+          value: new Transcription({
+            transcribedParticipantIdentity: agentIdentity,
+            trackId: trackSid,
+            segments: [
+              new TranscriptionSegmentModel({ id: 'SG_legacy', text: 'legacy', final: true }),
+            ],
+          }),
+        },
+      }),
+    );
+
+    expect(received).toHaveLength(0);
+  });
+
+  it('emits TranscriptionReceived from an lk.transcription stream', async () => {
+    const { room, participant, publication } = setupRoom();
+    const roomEvents: Array<{ segments: Array<{ text: string }>; identity?: string }> = [];
+    const trackEvents: Array<Array<{ text: string }>> = [];
+    room.on(RoomEvent.TranscriptionReceived, (segments, p) =>
+      roomEvents.push({ segments, identity: p?.identity }),
+    );
+    publication.on(TrackEvent.TranscriptionReceived, (segments) => trackEvents.push(segments));
+
+    pushTranscriptionStream(room, agentIdentity, 'Hello world', {
+      'lk.segment_id': 'SG_1',
+      'lk.transcribed_track_id': trackSid,
+      'lk.transcription_final': 'true',
+    });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    expect(roomEvents.length).toBeGreaterThan(0);
+    expect(roomEvents[0].segments[0].text).toBe('Hello world');
+    expect(roomEvents[0].identity).toBe(participant.identity);
+    expect(trackEvents.length).toBeGreaterThan(0);
+    expect(trackEvents[0][0].text).toBe('Hello world');
+  });
+
+  it("resolves the publication from the speaker's mic track when the attribute is absent", async () => {
+    const { room, publication } = setupRoom();
+    const trackEvents: Array<Array<{ text: string }>> = [];
+    publication.on(TrackEvent.TranscriptionReceived, (segments) => trackEvents.push(segments));
+
+    pushTranscriptionStream(room, agentIdentity, 'No track attribute', {
+      'lk.segment_id': 'SG_1',
+      'lk.transcription_final': 'true',
+    });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    expect(trackEvents.length).toBeGreaterThan(0);
+    expect(trackEvents[0][0].text).toBe('No track attribute');
+  });
+
+  it('still delivers lk.transcription to an application text stream handler', async () => {
+    const { room } = setupRoom();
+    const appTexts: Array<string> = [];
+    room.registerTextStreamHandler('lk.transcription', async (reader) => {
+      appTexts.push(await reader.readAll());
+    });
+
+    pushTranscriptionStream(room, agentIdentity, 'Hello world', {
+      'lk.segment_id': 'SG_1',
+      'lk.transcription_final': 'true',
+    });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    expect(appTexts).toEqual(['Hello world']);
+  });
+});

src/room/Room.ts

@@ -83,6 +83,7 @@ import LocalParticipant from './participant/LocalParticipant';
 import Participant from './participant/Participant';
 import { type ConnectionQuality, ParticipantKind } from './participant/Participant';
 import RemoteParticipant from './participant/RemoteParticipant';
+import { ParticipantAgentAttributes } from './participant/attributes';
 import {
   RPC_REQUEST_DATA_STREAM_TOPIC,
   RPC_RESPONSE_DATA_STREAM_TOPIC,
@@ -104,6 +105,7 @@ import type { TrackPublication } from './track/TrackPublication';
 import type { TrackProcessor } from './track/processor/types';
 import type { AdaptiveStreamSettings } from './track/types';
 import { getNewAudioContext, kindToSource, sourceToKind } from './track/utils';
+import TranscriptionStreamConverter from './transcription/TranscriptionStreamConverter';
 import {
   type ChatMessage,
   type SimulationOptions,
@@ -234,6 +236,8 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
 
   private incomingDataStreamManager: IncomingDataStreamManager;
 
+  private transcriptionStreamConverter: TranscriptionStreamConverter;
+
   private outgoingDataStreamManager: OutgoingDataStreamManager;
 
   private incomingDataTrackManager: IncomingDataTrackManager;
@@ -283,6 +287,17 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     this.incomingDataStreamManager = new IncomingDataStreamManager(
       this.options.dataStream?.maxPayloadByteLength,
     );
+    this.transcriptionStreamConverter = new TranscriptionStreamConverter({
+      onTranscription: (transcription) => this.handleTranscription(transcription),
+      getMicrophoneTrackSid: this.getMicrophoneTrackSid,
+      getDelegatingPublisherIdentity: this.getDelegatingPublisherIdentity,
+    });
+    this.incomingDataStreamManager.on(
+      'transcriptionStreamArrived',
+      ({ reader, participantIdentity }) => {
+        this.transcriptionStreamConverter.handleTextStream(reader, participantIdentity);
+      },
+    );
     this.outgoingDataStreamManager = new OutgoingDataStreamManager(
       this.engine,
       this.log,
@@ -1833,6 +1848,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     this.isResuming = false;
     this.bufferedEvents = [];
     this.transcriptionReceivedTimes.clear();
+    this.transcriptionStreamConverter.reset();
     this.incomingDataStreamManager.clearControllers();
     this.incomingDataTrackManager.reset();
     this.outgoingDataTrackManager.reset();
@@ -2099,7 +2115,11 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     if (packet.value.case === 'user') {
       this.handleUserPacket(participant, packet.value.value, packet.kind, encryptionType);
     } else if (packet.value.case === 'transcription') {
-      this.handleTranscription(participant, packet.value.value);
+      // Legacy `Transcription` packets are ignored: transcription events are rebuilt from the
+      // `lk.transcription` data stream channel instead, which this client advertises support for
+      // via client protocol 3. See
+      // docs/superpowers/specs/2026-09-04-transcription-back-conversion-design.md
+      this.log.debug('ignoring legacy transcription data packet', this.logContext);
     } else if (packet.value.case === 'sipDtmf') {
       this.handleSipDtmf(participant, packet.value.value);
     } else if (packet.value.case === 'chatMessage') {
@@ -2173,10 +2193,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     participant?.emit(ParticipantEvent.SipDTMFReceived, dtmf);
   };
 
-  private handleTranscription = (
-    _remoteParticipant: RemoteParticipant | undefined,
-    transcription: TranscriptionModel,
-  ) => {
+  private handleTranscription = (transcription: TranscriptionModel) => {
     // find the participant
     const participant =
       transcription.transcribedParticipantIdentity === this.localParticipant.identity
@@ -2603,6 +2620,28 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     );
   }
 
+  private getMicrophoneTrackSid = (identity: Participant['identity']): string | undefined => {
+    const participant = this.getParticipantByIdentity(identity);
+    for (const publication of participant?.trackPublications.values() ?? []) {
+      if (publication.source === Track.Source.Microphone) {
+        return publication.trackSid;
+      }
+    }
+    return undefined;
+  };
+
+  private getDelegatingPublisherIdentity = (
+    identity: Participant['identity'],
+  ): string | undefined => {
+    // An avatar worker carries `lk.publish_on_behalf` naming the agent it speaks for.
+    for (const participant of this.remoteParticipants.values()) {
+      if (participant.attributes[ParticipantAgentAttributes.PublishOnBehalf] === identity) {
+        return participant.identity;
+      }
+    }
+    return undefined;
+  };
+
   private setStatsLogging(enabled: boolean) {
     if (enabled) {
       if (!this.statsLogInterval) {

src/room/data-stream/constants.ts

@@ -16,3 +16,12 @@ export const STREAM_CHUNK_SIZE_BYTES = 15_000;
  * @internal
  */
 export const DEFAULT_MAX_PAYLOAD_BYTE_LENGTH = 5_000_000_000;
+
+/**
+ * Reserved topic carrying transcription text streams. `IncomingDataStreamManager` taps this topic
+ * with its `transcriptionStreamArrived` event so the SDK can rebuild transcription events while
+ * still delivering the stream to any application handler.
+ *
+ * @internal
+ */
+export const TRANSCRIPTION_TOPIC = 'lk.transcription';