{"version":3,"sources":["../../src/voice/console_io.ts"],"sourcesContent":["// SPDX-FileCopyrightText: 2026 LiveKit, Inc.\n//\n// SPDX-License-Identifier: Apache-2.0\nimport { AgentSession as pb } from '@livekit/protocol';\nimport { AudioFrame, AudioResampler } from '@livekit/rtc-node';\nimport * as path from 'node:path';\nimport { log } from '../log.js';\nimport { createStreamChannel } from '../stream/stream_channel.js';\nimport { Future, Task } from '../utils.js';\nimport type { AgentSession } from './agent_session.js';\nimport { AudioInput, AudioOutput } from './io.js';\nimport type { SessionTransport } from './remote_session.js';\n\n// The Go CLI / browser broker speaks 48 kHz on the wire; the agent pipeline\n// runs at 24 kHz. Resample on the way in and out.\nconst WIRE_SAMPLE_RATE = 48000;\nconst AGENT_SAMPLE_RATE = 24000;\n\nfunction consoleFrameToRtc(frame: pb.AgentSessionMessage_ConsoleIO_AudioFrame): AudioFrame {\n  // Copy into a fresh, 2-byte-aligned buffer; the protobuf bytes may start at an\n  // odd offset, which would make an Int16Array view throw.\n  const aligned = new ArrayBuffer(frame.data.byteLength);\n  new Uint8Array(aligned).set(frame.data);\n  return new AudioFrame(\n    new Int16Array(aligned),\n    frame.sampleRate,\n    frame.numChannels,\n    frame.samplesPerChannel,\n  );\n}\n\nfunction rtcFrameToConsole(frame: AudioFrame): pb.AgentSessionMessage_ConsoleIO_AudioFrame {\n  return new pb.AgentSessionMessage_ConsoleIO_AudioFrame({\n    data: new Uint8Array(frame.data.buffer, frame.data.byteOffset, frame.data.byteLength),\n    sampleRate: frame.sampleRate,\n    numChannels: frame.channels,\n    samplesPerChannel: frame.samplesPerChannel,\n  });\n}\n\n/**\n * Audio input fed by inbound `audio_input` console messages. Frames arrive at\n * the wire rate, are resampled to the agent rate, and pushed into the base\n * {@link AudioInput} stream that the STT pipeline reads from.\n *\n * Unlike the python port, no cross-thread queue is needed: the JS console job\n * runs in-process on a single event loop, so a stream channel is sufficient.\n *\n * @experimental\n */\nexport class TcpAudioInput extends AudioInput {\n  private readonly channel = createStreamChannel<AudioFrame>();\n  private readonly resampler = new AudioResampler(WIRE_SAMPLE_RATE, AGENT_SAMPLE_RATE, 1);\n  private closed = false;\n\n  constructor() {\n    super();\n    this.multiStream.addInputStream(this.channel.stream());\n  }\n\n  pushFrame(frame: pb.AgentSessionMessage_ConsoleIO_AudioFrame): void {\n    if (this.closed) return;\n    for (const resampled of this.resampler.push(consoleFrameToRtc(frame))) {\n      void this.channel.write(resampled);\n    }\n  }\n\n  override async close(): Promise<void> {\n    this.closed = true;\n    await this.channel.close();\n    await super.close();\n  }\n}\n\n/**\n * Audio output that streams the agent's TTS frames to the broker as\n * `audio_output` console messages (resampled to the wire rate) and drives the\n * flush/clear playout handshake. A flush blocks the agent turn until the broker\n * reports `audio_playback_finished` (or the buffer is cleared on interruption).\n *\n * @experimental\n */\nexport class TcpAudioOutput extends AudioOutput {\n  private readonly transport: SessionTransport;\n  private readonly resampler = new AudioResampler(AGENT_SAMPLE_RATE, WIRE_SAMPLE_RATE, 1);\n\n  private pushedDurationMs = 0;\n  private captureStartedAt = 0;\n  private flushTask?: Task<void>;\n  private playoutDone = new Future();\n  private interrupted = new Future();\n\n  constructor(transport: SessionTransport) {\n    super(AGENT_SAMPLE_RATE, undefined, { pause: true });\n    this.transport = transport;\n  }\n\n  override async captureFrame(frame: AudioFrame): Promise<void> {\n    await super.captureFrame(frame);\n\n    if (this.flushTask && !this.flushTask.done) {\n      log().error('captureFrame called while previous flush is in progress');\n      await this.flushTask.result;\n    }\n\n    if (this.pushedDurationMs === 0) {\n      this.captureStartedAt = Date.now();\n      this.onPlaybackStarted(Date.now());\n    }\n\n    this.pushedDurationMs += (frame.samplesPerChannel / frame.sampleRate) * 1000;\n\n    for (const resampled of this.resampler.push(frame)) {\n      await this.transport.sendMessage(\n        new pb.AgentSessionMessage({\n          message: { case: 'audioOutput', value: rtcFrameToConsole(resampled) },\n        }),\n      );\n    }\n  }\n\n  override flush(): void {\n    super.flush();\n    void this.transport.sendMessage(\n      new pb.AgentSessionMessage({\n        message: {\n          case: 'audioPlaybackFlush',\n          value: new pb.AgentSessionMessage_ConsoleIO_AudioPlaybackFlush(),\n        },\n      }),\n    );\n\n    if (this.pushedDurationMs > 0) {\n      if (this.flushTask && !this.flushTask.done) {\n        log().error('flush called while previous flush is in progress');\n        // TODO(follow-up PR): cancel() is currently a no-op here. runPlayoutHandshake\n        // awaits Futures that are never rejected and ignores the AbortController, so\n        // the previous handshake leaks instead of being cancelled (unlike Python's\n        // asyncio.Task.cancel). Make the handshake abort-aware before relying on this.\n        this.flushTask.cancel();\n      }\n      this.playoutDone = new Future();\n      this.interrupted = new Future();\n      this.flushTask = Task.from(() => this.runPlayoutHandshake());\n    }\n  }\n\n  override clearBuffer(): void {\n    void this.transport.sendMessage(\n      new pb.AgentSessionMessage({\n        message: {\n          case: 'audioPlaybackClear',\n          value: new pb.AgentSessionMessage_ConsoleIO_AudioPlaybackClear(),\n        },\n      }),\n    );\n\n    if (this.pushedDurationMs > 0) {\n      this.interrupted.resolve();\n    }\n  }\n\n  /** Called by {@link SessionHost} when the broker reports playout finished. */\n  notifyPlayoutFinished(): void {\n    if (!this.playoutDone.done) {\n      this.playoutDone.resolve();\n    }\n  }\n\n  private async runPlayoutHandshake(): Promise<void> {\n    try {\n      await Promise.race([this.playoutDone.await, this.interrupted.await]);\n    } catch {\n      // TODO(follow-up PR): unreachable today — the raced Futures are only ever\n      // resolved, never rejected. Becomes the cancellation path once the handshake\n      // is made abort-aware (see flush()).\n      return;\n    }\n    const interrupted = this.interrupted.done && !this.playoutDone.done;\n\n    let playedMs: number;\n    if (interrupted) {\n      const elapsed = Date.now() - this.captureStartedAt;\n      playedMs = Math.min(Math.max(0, elapsed), this.pushedDurationMs);\n    } else {\n      playedMs = this.pushedDurationMs;\n    }\n\n    this.onPlaybackFinished({ playbackPosition: playedMs / 1000, interrupted });\n\n    this.pushedDurationMs = 0;\n    this.interrupted = new Future();\n  }\n}\n\n/**\n * Process-wide singleton that carries the console transport (and audio IO) from\n * the `console` CLI runner into the {@link AgentSession} that starts inside the\n * agent entrypoint. Mirrors python `AgentsConsole`.\n *\n * The runner sets {@link enabled} and the transport; `AgentSession._startImpl`\n * then calls {@link acquireIo} to wire the session and build its `SessionHost`.\n *\n * @experimental\n */\nexport class AgentsConsole {\n  private static _instance?: AgentsConsole;\n\n  static getInstance(): AgentsConsole {\n    if (!AgentsConsole._instance) {\n      AgentsConsole._instance = new AgentsConsole();\n    }\n    return AgentsConsole._instance;\n  }\n\n  /** @internal — reset the process-wide singleton (test only). */\n  static _reset(): void {\n    AgentsConsole._instance = undefined;\n  }\n\n  enabled = false;\n  record = false;\n  transport?: SessionTransport;\n  audioInput?: TcpAudioInput;\n  audioOutput?: TcpAudioOutput;\n\n  /**\n   * Directory where console recordings (audio + session report) are written,\n   * relative to the working directory. Mirrors python's `AgentsConsole`.\n   */\n  readonly sessionDirectory: string;\n\n  private acquired = false;\n\n  private constructor() {\n    const now = new Date();\n    const pad = (n: number) => String(n).padStart(2, '0');\n    this.sessionDirectory = path.join(\n      'console-recordings',\n      `session-${pad(now.getMonth() + 1)}-${pad(now.getDate())}-${pad(now.getHours())}${pad(now.getMinutes())}${pad(now.getSeconds())}`,\n    );\n  }\n\n  get ioAcquired(): boolean {\n    return this.acquired;\n  }\n\n  /** Wire the session's IO for console mode. */\n  acquireIo(session: AgentSession): void {\n    if (this.acquired) {\n      throw new Error('the console IO was already acquired by another session');\n    }\n    this.acquired = true;\n\n    // Attach the audio bridges, enabled by default (voice mode). A text-mode\n    // driver disables them at runtime with an `update_io` request (audio off),\n    // so one wiring path serves both modes.\n    session.input.audio = this.audioInput ?? null;\n    session.output.audio = this.audioOutput ?? null;\n    // TODO(follow-up PR): Route audio through a TranscriptSynchronizer and set\n    // output.transcription to its text output, so transcripts stay in sync with\n    // playout. We leave transcription unset for now (the Go CLI owns the TUI).\n    session.output.transcription = null;\n  }\n}\n"],"mappings":";;;;;;;;;;;;;;;;;;;;;;;;;;;;AAAA;AAAA;AAAA;AAAA;AAAA;AAAA;AAAA;AAGA,sBAAmC;AACnC,sBAA2C;AAC3C,WAAsB;AACtB,iBAAoB;AACpB,4BAAoC;AACpC,mBAA6B;AAE7B,gBAAwC;AAKxC,MAAM,mBAAmB;AACzB,MAAM,oBAAoB;AAE1B,SAAS,kBAAkB,OAAgE;AAGzF,QAAM,UAAU,IAAI,YAAY,MAAM,KAAK,UAAU;AACrD,MAAI,WAAW,OAAO,EAAE,IAAI,MAAM,IAAI;AACtC,SAAO,IAAI;AAAA,IACT,IAAI,WAAW,OAAO;AAAA,IACtB,MAAM;AAAA,IACN,MAAM;AAAA,IACN,MAAM;AAAA,EACR;AACF;AAEA,SAAS,kBAAkB,OAAgE;AACzF,SAAO,IAAI,gBAAAA,aAAG,yCAAyC;AAAA,IACrD,MAAM,IAAI,WAAW,MAAM,KAAK,QAAQ,MAAM,KAAK,YAAY,MAAM,KAAK,UAAU;AAAA,IACpF,YAAY,MAAM;AAAA,IAClB,aAAa,MAAM;AAAA,IACnB,mBAAmB,MAAM;AAAA,EAC3B,CAAC;AACH;AAYO,MAAM,sBAAsB,qBAAW;AAAA,EAC3B,cAAU,2CAAgC;AAAA,EAC1C,YAAY,IAAI,+BAAe,kBAAkB,mBAAmB,CAAC;AAAA,EAC9E,SAAS;AAAA,EAEjB,cAAc;AACZ,UAAM;AACN,SAAK,YAAY,eAAe,KAAK,QAAQ,OAAO,CAAC;AAAA,EACvD;AAAA,EAEA,UAAU,OAA0D;AAClE,QAAI,KAAK,OAAQ;AACjB,eAAW,aAAa,KAAK,UAAU,KAAK,kBAAkB,KAAK,CAAC,GAAG;AACrE,WAAK,KAAK,QAAQ,MAAM,SAAS;AAAA,IACnC;AAAA,EACF;AAAA,EAEA,MAAe,QAAuB;AACpC,SAAK,SAAS;AACd,UAAM,KAAK,QAAQ,MAAM;AACzB,UAAM,MAAM,MAAM;AAAA,EACpB;AACF;AAUO,MAAM,uBAAuB,sBAAY;AAAA,EAC7B;AAAA,EACA,YAAY,IAAI,+BAAe,mBAAmB,kBAAkB,CAAC;AAAA,EAE9E,mBAAmB;AAAA,EACnB,mBAAmB;AAAA,EACnB;AAAA,EACA,cAAc,IAAI,oBAAO;AAAA,EACzB,cAAc,IAAI,oBAAO;AAAA,EAEjC,YAAY,WAA6B;AACvC,UAAM,mBAAmB,QAAW,EAAE,OAAO,KAAK,CAAC;AACnD,SAAK,YAAY;AAAA,EACnB;AAAA,EAEA,MAAe,aAAa,OAAkC;AAC5D,UAAM,MAAM,aAAa,KAAK;AAE9B,QAAI,KAAK,aAAa,CAAC,KAAK,UAAU,MAAM;AAC1C,0BAAI,EAAE,MAAM,yDAAyD;AACrE,YAAM,KAAK,UAAU;AAAA,IACvB;AAEA,QAAI,KAAK,qBAAqB,GAAG;AAC/B,WAAK,mBAAmB,KAAK,IAAI;AACjC,WAAK,kBAAkB,KAAK,IAAI,CAAC;AAAA,IACnC;AAEA,SAAK,oBAAqB,MAAM,oBAAoB,MAAM,aAAc;AAExE,eAAW,aAAa,KAAK,UAAU,KAAK,KAAK,GAAG;AAClD,YAAM,KAAK,UAAU;AAAA,QACnB,IAAI,gBAAAA,aAAG,oBAAoB;AAAA,UACzB,SAAS,EAAE,MAAM,eAAe,OAAO,kBAAkB,SAAS,EAAE;AAAA,QACtE,CAAC;AAAA,MACH;AAAA,IACF;AAAA,EACF;AAAA,EAES,QAAc;AACrB,UAAM,MAAM;AACZ,SAAK,KAAK,UAAU;AAAA,MAClB,IAAI,gBAAAA,aAAG,oBAAoB;AAAA,QACzB,SAAS;AAAA,UACP,MAAM;AAAA,UACN,OAAO,IAAI,gBAAAA,aAAG,iDAAiD;AAAA,QACjE;AAAA,MACF,CAAC;AAAA,IACH;AAEA,QAAI,KAAK,mBAAmB,GAAG;AAC7B,UAAI,KAAK,aAAa,CAAC,KAAK,UAAU,MAAM;AAC1C,4BAAI,EAAE,MAAM,kDAAkD;AAK9D,aAAK,UAAU,OAAO;AAAA,MACxB;AACA,WAAK,cAAc,IAAI,oBAAO;AAC9B,WAAK,cAAc,IAAI,oBAAO;AAC9B,WAAK,YAAY,kBAAK,KAAK,MAAM,KAAK,oBAAoB,CAAC;AAAA,IAC7D;AAAA,EACF;AAAA,EAES,cAAoB;AAC3B,SAAK,KAAK,UAAU;AAAA,MAClB,IAAI,gBAAAA,aAAG,oBAAoB;AAAA,QACzB,SAAS;AAAA,UACP,MAAM;AAAA,UACN,OAAO,IAAI,gBAAAA,aAAG,iDAAiD;AAAA,QACjE;AAAA,MACF,CAAC;AAAA,IACH;AAEA,QAAI,KAAK,mBAAmB,GAAG;AAC7B,WAAK,YAAY,QAAQ;AAAA,IAC3B;AAAA,EACF;AAAA;AAAA,EAGA,wBAA8B;AAC5B,QAAI,CAAC,KAAK,YAAY,MAAM;AAC1B,WAAK,YAAY,QAAQ;AAAA,IAC3B;AAAA,EACF;AAAA,EAEA,MAAc,sBAAqC;AACjD,QAAI;AACF,YAAM,QAAQ,KAAK,CAAC,KAAK,YAAY,OAAO,KAAK,YAAY,KAAK,CAAC;AAAA,IACrE,QAAQ;AAIN;AAAA,IACF;AACA,UAAM,cAAc,KAAK,YAAY,QAAQ,CAAC,KAAK,YAAY;AAE/D,QAAI;AACJ,QAAI,aAAa;AACf,YAAM,UAAU,KAAK,IAAI,IAAI,KAAK;AAClC,iBAAW,KAAK,IAAI,KAAK,IAAI,GAAG,OAAO,GAAG,KAAK,gBAAgB;AAAA,IACjE,OAAO;AACL,iBAAW,KAAK;AAAA,IAClB;AAEA,SAAK,mBAAmB,EAAE,kBAAkB,WAAW,KAAM,YAAY,CAAC;AAE1E,SAAK,mBAAmB;AACxB,SAAK,cAAc,IAAI,oBAAO;AAAA,EAChC;AACF;AAYO,MAAM,cAAc;AAAA,EACzB,OAAe;AAAA,EAEf,OAAO,cAA6B;AAClC,QAAI,CAAC,cAAc,WAAW;AAC5B,oBAAc,YAAY,IAAI,cAAc;AAAA,IAC9C;AACA,WAAO,cAAc;AAAA,EACvB;AAAA;AAAA,EAGA,OAAO,SAAe;AACpB,kBAAc,YAAY;AAAA,EAC5B;AAAA,EAEA,UAAU;AAAA,EACV,SAAS;AAAA,EACT;AAAA,EACA;AAAA,EACA;AAAA;AAAA;AAAA;AAAA;AAAA,EAMS;AAAA,EAED,WAAW;AAAA,EAEX,cAAc;AACpB,UAAM,MAAM,oBAAI,KAAK;AACrB,UAAM,MAAM,CAAC,MAAc,OAAO,CAAC,EAAE,SAAS,GAAG,GAAG;AACpD,SAAK,mBAAmB,KAAK;AAAA,MAC3B;AAAA,MACA,WAAW,IAAI,IAAI,SAAS,IAAI,CAAC,CAAC,IAAI,IAAI,IAAI,QAAQ,CAAC,CAAC,IAAI,IAAI,IAAI,SAAS,CAAC,CAAC,GAAG,IAAI,IAAI,WAAW,CAAC,CAAC,GAAG,IAAI,IAAI,WAAW,CAAC,CAAC;AAAA,IACjI;AAAA,EACF;AAAA,EAEA,IAAI,aAAsB;AACxB,WAAO,KAAK;AAAA,EACd;AAAA;AAAA,EAGA,UAAU,SAA6B;AACrC,QAAI,KAAK,UAAU;AACjB,YAAM,IAAI,MAAM,wDAAwD;AAAA,IAC1E;AACA,SAAK,WAAW;AAKhB,YAAQ,MAAM,QAAQ,KAAK,cAAc;AACzC,YAAQ,OAAO,QAAQ,KAAK,eAAe;AAI3C,YAAQ,OAAO,gBAAgB;AAAA,EACjC;AACF;","names":["pb"]}