{"version":3,"sources":["../../../src/voice/avatar/datastream_io.ts"],"sourcesContent":["// SPDX-FileCopyrightText: 2025 LiveKit, Inc.\n//\n// SPDX-License-Identifier: Apache-2.0\nimport { Mutex } from '@livekit/mutex';\nimport {\n  type AudioFrame,\n  type ByteStreamWriter,\n  type Room,\n  RoomEvent,\n  type RpcInvocationData,\n  type TrackKind,\n} from '@livekit/rtc-node';\nimport { log } from '../../log.js';\nimport {\n  Future,\n  Task,\n  shortuuid,\n  waitForParticipant,\n  waitForTrackPublication,\n} from '../../utils.js';\nimport { AudioOutput } from '../io.js';\nimport { parsePlaybackFinishedPayload } from './playback_payload.js';\n\nconst RPC_CLEAR_BUFFER = 'lk.clear_buffer';\nconst RPC_PLAYBACK_FINISHED = 'lk.playback_finished';\nconst RPC_PLAYBACK_STARTED = 'lk.playback_started';\nconst AUDIO_STREAM_TOPIC = 'lk.audio_stream';\n\nexport interface DataStreamAudioOutputOptions {\n  room: Room;\n  destinationIdentity: string;\n  sampleRate?: number;\n  waitRemoteTrack?: TrackKind;\n  /**\n   * When true, defer the `playbackStarted` event until the remote participant invokes the\n   * `lk.playback_started` RPC. When false (default), the event fires eagerly the moment the\n   * first audio frame is captured. Set to true when the remote avatar worker can notify\n   * actual playout start (e.g. an `AvatarRunner`).\n   */\n  waitPlaybackStart?: boolean;\n}\n\n/**\n * AudioOutput implementation that streams audio to a remote avatar worker using LiveKit DataStream.\n */\nexport class DataStreamAudioOutput extends AudioOutput {\n  static _playbackFinishedRpcRegistered: boolean = false;\n  static _playbackFinishedHandlers: Record<string, (data: RpcInvocationData) => string> = {};\n  static _playbackStartedRpcRegistered: boolean = false;\n  static _playbackStartedHandlers: Record<string, (data: RpcInvocationData) => string> = {};\n\n  private room: Room;\n  private destinationIdentity: string;\n  private roomConnectedFuture: Future<void>;\n  private waitRemoteTrack?: TrackKind;\n  private waitPlaybackStart: boolean;\n  private streamWriter?: ByteStreamWriter;\n  private pushedDuration: number = 0;\n  private started: boolean = false;\n  private lock = new Mutex();\n  private startTask?: Task<void>;\n  private firstFrameEmitted: boolean = false;\n\n  #logger = log();\n\n  constructor(opts: DataStreamAudioOutputOptions) {\n    super(opts.sampleRate, undefined, { pause: false });\n\n    const { room, destinationIdentity, sampleRate, waitRemoteTrack, waitPlaybackStart } = opts;\n    this.room = room;\n    this.destinationIdentity = destinationIdentity;\n    this.sampleRate = sampleRate;\n    this.waitRemoteTrack = waitRemoteTrack;\n    this.waitPlaybackStart = waitPlaybackStart ?? false;\n\n    const onRoomConnected = async () => {\n      if (this.startTask) return;\n\n      await this.roomConnectedFuture.await;\n\n      // register the rpc method right after the room is connected\n      DataStreamAudioOutput.registerPlaybackFinishedRpc({\n        room,\n        callerIdentity: this.destinationIdentity,\n        handler: (data) => this.handlePlaybackFinished(data),\n      });\n\n      if (this.waitPlaybackStart) {\n        DataStreamAudioOutput.registerPlaybackStartedRpc({\n          room,\n          callerIdentity: this.destinationIdentity,\n          handler: (data) => this.handlePlaybackStarted(data),\n        });\n      }\n\n      this.startTask = Task.from(({ signal }) => this._start(signal));\n    };\n\n    this.roomConnectedFuture = new Future<void>();\n\n    this.room.on(RoomEvent.ConnectionStateChanged, (_) => {\n      if (room.isConnected && !this.roomConnectedFuture.done) {\n        this.roomConnectedFuture.resolve(undefined);\n      }\n    });\n\n    if (this.room.isConnected) {\n      this.roomConnectedFuture.resolve(undefined);\n    }\n\n    onRoomConnected();\n  }\n\n  private async _start(_abortSignal: AbortSignal) {\n    const unlock = await this.lock.lock();\n\n    try {\n      if (this.started) return;\n\n      await this.roomConnectedFuture.await;\n\n      this.#logger.debug(\n        {\n          'lk.pii.destination_identity': this.destinationIdentity,\n        },\n        'waiting for the remote participant',\n      );\n\n      await waitForParticipant({\n        room: this.room,\n        identity: this.destinationIdentity,\n      });\n\n      if (this.waitRemoteTrack) {\n        this.#logger.debug(\n          {\n            'lk.pii.destination_identity': this.destinationIdentity,\n            kind: this.waitRemoteTrack,\n          },\n          'waiting for the remote track',\n        );\n\n        await waitForTrackPublication({\n          room: this.room,\n          identity: this.destinationIdentity,\n          kind: this.waitRemoteTrack,\n        });\n      }\n\n      this.#logger.debug(\n        {\n          'lk.pii.destination_identity': this.destinationIdentity,\n        },\n        'remote participant ready',\n      );\n\n      this.started = true;\n    } finally {\n      unlock();\n    }\n  }\n\n  async captureFrame(frame: AudioFrame): Promise<void> {\n    if (!this.startTask) {\n      this.startTask = Task.from(({ signal }) => this._start(signal));\n    }\n\n    await this.startTask.result;\n    await super.captureFrame(frame);\n\n    if (!this.firstFrameEmitted) {\n      this.firstFrameEmitted = true;\n      if (!this.waitPlaybackStart) {\n        // approximate the playback_started time; the frame isn't actually playing yet,\n        // used when the remote avatar doesn't send lk.playback_started notifications\n        this.onPlaybackStarted(Date.now());\n      }\n    }\n\n    if (!this.streamWriter) {\n      this.streamWriter = await this.room.localParticipant!.streamBytes({\n        name: shortuuid('AUDIO_'),\n        topic: AUDIO_STREAM_TOPIC,\n        destinationIdentities: [this.destinationIdentity],\n        attributes: {\n          sample_rate: frame.sampleRate.toString(),\n          num_channels: frame.channels.toString(),\n        },\n      });\n      this.pushedDuration = 0;\n    }\n\n    // frame.data is a Int16Array, write accepts a Uint8Array\n    await this.streamWriter.write(new Uint8Array(frame.data.buffer));\n    this.pushedDuration += frame.samplesPerChannel / frame.sampleRate;\n  }\n\n  flush(): void {\n    super.flush();\n\n    if (this.streamWriter === undefined || !this.started) {\n      return;\n    }\n\n    this.streamWriter.close().finally(() => {\n      this.streamWriter = undefined;\n    });\n\n    this.firstFrameEmitted = false;\n  }\n\n  clearBuffer(): void {\n    if (!this.started) return;\n\n    this.room.localParticipant!.performRpc({\n      destinationIdentity: this.destinationIdentity,\n      method: RPC_CLEAR_BUFFER,\n      payload: '',\n    });\n  }\n\n  private handlePlaybackFinished(data: RpcInvocationData): string {\n    if (data.callerIdentity !== this.destinationIdentity) {\n      this.#logger.warn(\n        {\n          'lk.pii.caller_identity': data.callerIdentity,\n          'lk.pii.destination_identity': this.destinationIdentity,\n        },\n        'playback finished event received from unexpected participant',\n      );\n      return 'reject';\n    }\n\n    this.#logger.info(\n      {\n        'lk.pii.caller_identity': data.callerIdentity,\n      },\n      'playback finished event received',\n    );\n\n    // The wire payload uses the protocol-canonical snake_case keys\n    // (`playback_position`, `synchronized_transcript`); parsePlaybackFinishedPayload\n    // normalizes them into the camelCase PlaybackFinishedEvent shape.\n    this.onPlaybackFinished(parsePlaybackFinishedPayload(data.payload));\n    return 'ok';\n  }\n\n  static registerPlaybackFinishedRpc({\n    room,\n    callerIdentity,\n    handler,\n  }: {\n    room: Room;\n    callerIdentity: string;\n    handler: (data: RpcInvocationData) => string;\n  }) {\n    DataStreamAudioOutput._playbackFinishedHandlers[callerIdentity] = handler;\n\n    if (DataStreamAudioOutput._playbackFinishedRpcRegistered) {\n      return;\n    }\n\n    const rpcHandler = async (data: RpcInvocationData): Promise<string> => {\n      const handler = DataStreamAudioOutput._playbackFinishedHandlers[data.callerIdentity];\n      if (!handler) {\n        log().warn(\n          {\n            'lk.pii.caller_identity': data.callerIdentity,\n            'lk.pii.expected_identities': Object.keys(\n              DataStreamAudioOutput._playbackFinishedHandlers,\n            ),\n          },\n          'playback finished event received from unexpected participant',\n        );\n\n        return 'reject';\n      }\n      return handler(data);\n    };\n\n    room.localParticipant?.registerRpcMethod(RPC_PLAYBACK_FINISHED, rpcHandler);\n    DataStreamAudioOutput._playbackFinishedRpcRegistered = true;\n  }\n\n  private handlePlaybackStarted(data: RpcInvocationData): string {\n    if (data.callerIdentity !== this.destinationIdentity) {\n      this.#logger.warn(\n        {\n          'lk.pii.caller_identity': data.callerIdentity,\n          'lk.pii.destination_identity': this.destinationIdentity,\n        },\n        'playback started event received from unexpected participant',\n      );\n      return 'reject';\n    }\n\n    this.onPlaybackStarted(Date.now());\n    return 'ok';\n  }\n\n  static registerPlaybackStartedRpc({\n    room,\n    callerIdentity,\n    handler,\n  }: {\n    room: Room;\n    callerIdentity: string;\n    handler: (data: RpcInvocationData) => string;\n  }) {\n    DataStreamAudioOutput._playbackStartedHandlers[callerIdentity] = handler;\n\n    if (DataStreamAudioOutput._playbackStartedRpcRegistered) {\n      return;\n    }\n\n    const rpcHandler = async (data: RpcInvocationData): Promise<string> => {\n      const handler = DataStreamAudioOutput._playbackStartedHandlers[data.callerIdentity];\n      if (!handler) {\n        log().warn(\n          {\n            'lk.pii.caller_identity': data.callerIdentity,\n            'lk.pii.expected_identities': Object.keys(\n              DataStreamAudioOutput._playbackStartedHandlers,\n            ),\n          },\n          'playback started event received from unexpected participant',\n        );\n\n        return 'reject';\n      }\n      return handler(data);\n    };\n\n    room.localParticipant?.registerRpcMethod(RPC_PLAYBACK_STARTED, rpcHandler);\n    DataStreamAudioOutput._playbackStartedRpcRegistered = true;\n  }\n}\n"],"mappings":";;;;;;;;;;;;;;;;;;AAAA;AAAA;AAAA;AAAA;AAAA;AAGA,mBAAsB;AACtB,sBAOO;AACP,iBAAoB;AACpB,mBAMO;AACP,gBAA4B;AAC5B,8BAA6C;AAE7C,MAAM,mBAAmB;AACzB,MAAM,wBAAwB;AAC9B,MAAM,uBAAuB;AAC7B,MAAM,qBAAqB;AAmBpB,MAAM,8BAA8B,sBAAY;AAAA,EACrD,OAAO,iCAA0C;AAAA,EACjD,OAAO,4BAAiF,CAAC;AAAA,EACzF,OAAO,gCAAyC;AAAA,EAChD,OAAO,2BAAgF,CAAC;AAAA,EAEhF;AAAA,EACA;AAAA,EACA;AAAA,EACA;AAAA,EACA;AAAA,EACA;AAAA,EACA,iBAAyB;AAAA,EACzB,UAAmB;AAAA,EACnB,OAAO,IAAI,mBAAM;AAAA,EACjB;AAAA,EACA,oBAA6B;AAAA,EAErC,cAAU,gBAAI;AAAA,EAEd,YAAY,MAAoC;AAC9C,UAAM,KAAK,YAAY,QAAW,EAAE,OAAO,MAAM,CAAC;AAElD,UAAM,EAAE,MAAM,qBAAqB,YAAY,iBAAiB,kBAAkB,IAAI;AACtF,SAAK,OAAO;AACZ,SAAK,sBAAsB;AAC3B,SAAK,aAAa;AAClB,SAAK,kBAAkB;AACvB,SAAK,oBAAoB,qBAAqB;AAE9C,UAAM,kBAAkB,YAAY;AAClC,UAAI,KAAK,UAAW;AAEpB,YAAM,KAAK,oBAAoB;AAG/B,4BAAsB,4BAA4B;AAAA,QAChD;AAAA,QACA,gBAAgB,KAAK;AAAA,QACrB,SAAS,CAAC,SAAS,KAAK,uBAAuB,IAAI;AAAA,MACrD,CAAC;AAED,UAAI,KAAK,mBAAmB;AAC1B,8BAAsB,2BAA2B;AAAA,UAC/C;AAAA,UACA,gBAAgB,KAAK;AAAA,UACrB,SAAS,CAAC,SAAS,KAAK,sBAAsB,IAAI;AAAA,QACpD,CAAC;AAAA,MACH;AAEA,WAAK,YAAY,kBAAK,KAAK,CAAC,EAAE,OAAO,MAAM,KAAK,OAAO,MAAM,CAAC;AAAA,IAChE;AAEA,SAAK,sBAAsB,IAAI,oBAAa;AAE5C,SAAK,KAAK,GAAG,0BAAU,wBAAwB,CAAC,MAAM;AACpD,UAAI,KAAK,eAAe,CAAC,KAAK,oBAAoB,MAAM;AACtD,aAAK,oBAAoB,QAAQ,MAAS;AAAA,MAC5C;AAAA,IACF,CAAC;AAED,QAAI,KAAK,KAAK,aAAa;AACzB,WAAK,oBAAoB,QAAQ,MAAS;AAAA,IAC5C;AAEA,oBAAgB;AAAA,EAClB;AAAA,EAEA,MAAc,OAAO,cAA2B;AAC9C,UAAM,SAAS,MAAM,KAAK,KAAK,KAAK;AAEpC,QAAI;AACF,UAAI,KAAK,QAAS;AAElB,YAAM,KAAK,oBAAoB;AAE/B,WAAK,QAAQ;AAAA,QACX;AAAA,UACE,+BAA+B,KAAK;AAAA,QACtC;AAAA,QACA;AAAA,MACF;AAEA,gBAAM,iCAAmB;AAAA,QACvB,MAAM,KAAK;AAAA,QACX,UAAU,KAAK;AAAA,MACjB,CAAC;AAED,UAAI,KAAK,iBAAiB;AACxB,aAAK,QAAQ;AAAA,UACX;AAAA,YACE,+BAA+B,KAAK;AAAA,YACpC,MAAM,KAAK;AAAA,UACb;AAAA,UACA;AAAA,QACF;AAEA,kBAAM,sCAAwB;AAAA,UAC5B,MAAM,KAAK;AAAA,UACX,UAAU,KAAK;AAAA,UACf,MAAM,KAAK;AAAA,QACb,CAAC;AAAA,MACH;AAEA,WAAK,QAAQ;AAAA,QACX;AAAA,UACE,+BAA+B,KAAK;AAAA,QACtC;AAAA,QACA;AAAA,MACF;AAEA,WAAK,UAAU;AAAA,IACjB,UAAE;AACA,aAAO;AAAA,IACT;AAAA,EACF;AAAA,EAEA,MAAM,aAAa,OAAkC;AACnD,QAAI,CAAC,KAAK,WAAW;AACnB,WAAK,YAAY,kBAAK,KAAK,CAAC,EAAE,OAAO,MAAM,KAAK,OAAO,MAAM,CAAC;AAAA,IAChE;AAEA,UAAM,KAAK,UAAU;AACrB,UAAM,MAAM,aAAa,KAAK;AAE9B,QAAI,CAAC,KAAK,mBAAmB;AAC3B,WAAK,oBAAoB;AACzB,UAAI,CAAC,KAAK,mBAAmB;AAG3B,aAAK,kBAAkB,KAAK,IAAI,CAAC;AAAA,MACnC;AAAA,IACF;AAEA,QAAI,CAAC,KAAK,cAAc;AACtB,WAAK,eAAe,MAAM,KAAK,KAAK,iBAAkB,YAAY;AAAA,QAChE,UAAM,wBAAU,QAAQ;AAAA,QACxB,OAAO;AAAA,QACP,uBAAuB,CAAC,KAAK,mBAAmB;AAAA,QAChD,YAAY;AAAA,UACV,aAAa,MAAM,WAAW,SAAS;AAAA,UACvC,cAAc,MAAM,SAAS,SAAS;AAAA,QACxC;AAAA,MACF,CAAC;AACD,WAAK,iBAAiB;AAAA,IACxB;AAGA,UAAM,KAAK,aAAa,MAAM,IAAI,WAAW,MAAM,KAAK,MAAM,CAAC;AAC/D,SAAK,kBAAkB,MAAM,oBAAoB,MAAM;AAAA,EACzD;AAAA,EAEA,QAAc;AACZ,UAAM,MAAM;AAEZ,QAAI,KAAK,iBAAiB,UAAa,CAAC,KAAK,SAAS;AACpD;AAAA,IACF;AAEA,SAAK,aAAa,MAAM,EAAE,QAAQ,MAAM;AACtC,WAAK,eAAe;AAAA,IACtB,CAAC;AAED,SAAK,oBAAoB;AAAA,EAC3B;AAAA,EAEA,cAAoB;AAClB,QAAI,CAAC,KAAK,QAAS;AAEnB,SAAK,KAAK,iBAAkB,WAAW;AAAA,MACrC,qBAAqB,KAAK;AAAA,MAC1B,QAAQ;AAAA,MACR,SAAS;AAAA,IACX,CAAC;AAAA,EACH;AAAA,EAEQ,uBAAuB,MAAiC;AAC9D,QAAI,KAAK,mBAAmB,KAAK,qBAAqB;AACpD,WAAK,QAAQ;AAAA,QACX;AAAA,UACE,0BAA0B,KAAK;AAAA,UAC/B,+BAA+B,KAAK;AAAA,QACtC;AAAA,QACA;AAAA,MACF;AACA,aAAO;AAAA,IACT;AAEA,SAAK,QAAQ;AAAA,MACX;AAAA,QACE,0BAA0B,KAAK;AAAA,MACjC;AAAA,MACA;AAAA,IACF;AAKA,SAAK,uBAAmB,sDAA6B,KAAK,OAAO,CAAC;AAClE,WAAO;AAAA,EACT;AAAA,EAEA,OAAO,4BAA4B;AAAA,IACjC;AAAA,IACA;AAAA,IACA;AAAA,EACF,GAIG;AA/PL;AAgQI,0BAAsB,0BAA0B,cAAc,IAAI;AAElE,QAAI,sBAAsB,gCAAgC;AACxD;AAAA,IACF;AAEA,UAAM,aAAa,OAAO,SAA6C;AACrE,YAAMA,WAAU,sBAAsB,0BAA0B,KAAK,cAAc;AACnF,UAAI,CAACA,UAAS;AACZ,4BAAI,EAAE;AAAA,UACJ;AAAA,YACE,0BAA0B,KAAK;AAAA,YAC/B,8BAA8B,OAAO;AAAA,cACnC,sBAAsB;AAAA,YACxB;AAAA,UACF;AAAA,UACA;AAAA,QACF;AAEA,eAAO;AAAA,MACT;AACA,aAAOA,SAAQ,IAAI;AAAA,IACrB;AAEA,eAAK,qBAAL,mBAAuB,kBAAkB,uBAAuB;AAChE,0BAAsB,iCAAiC;AAAA,EACzD;AAAA,EAEQ,sBAAsB,MAAiC;AAC7D,QAAI,KAAK,mBAAmB,KAAK,qBAAqB;AACpD,WAAK,QAAQ;AAAA,QACX;AAAA,UACE,0BAA0B,KAAK;AAAA,UAC/B,+BAA+B,KAAK;AAAA,QACtC;AAAA,QACA;AAAA,MACF;AACA,aAAO;AAAA,IACT;AAEA,SAAK,kBAAkB,KAAK,IAAI,CAAC;AACjC,WAAO;AAAA,EACT;AAAA,EAEA,OAAO,2BAA2B;AAAA,IAChC;AAAA,IACA;AAAA,IACA;AAAA,EACF,GAIG;AApTL;AAqTI,0BAAsB,yBAAyB,cAAc,IAAI;AAEjE,QAAI,sBAAsB,+BAA+B;AACvD;AAAA,IACF;AAEA,UAAM,aAAa,OAAO,SAA6C;AACrE,YAAMA,WAAU,sBAAsB,yBAAyB,KAAK,cAAc;AAClF,UAAI,CAACA,UAAS;AACZ,4BAAI,EAAE;AAAA,UACJ;AAAA,YACE,0BAA0B,KAAK;AAAA,YAC/B,8BAA8B,OAAO;AAAA,cACnC,sBAAsB;AAAA,YACxB;AAAA,UACF;AAAA,UACA;AAAA,QACF;AAEA,eAAO;AAAA,MACT;AACA,aAAOA,SAAQ,IAAI;AAAA,IACrB;AAEA,eAAK,qBAAL,mBAAuB,kBAAkB,sBAAsB;AAC/D,0BAAsB,gCAAgC;AAAA,EACxD;AACF;","names":["handler"]}