From 54db7d4f99e8286fa26367151db3f3a34fbe08b0 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 26 Aug 2026 14:42:03 -0700 Subject: [PATCH] refactor(discord): centralize voice resource lifecycle (#130415) * refactor(discord): centralize voice resource lifecycle * test(discord): avoid returning from voice promise executor --- .../discord/src/voice/audio.lifecycle.test.ts | 58 +++++ extensions/discord/src/voice/audio.test.ts | 97 +++++---- extensions/discord/src/voice/audio.ts | 199 ++++++------------ .../src/voice/voice-session.startup.test.ts | 117 ++++++++++ extensions/discord/src/voice/voice-session.ts | 129 +++++------- 5 files changed, 348 insertions(+), 252 deletions(-) create mode 100644 extensions/discord/src/voice/audio.lifecycle.test.ts create mode 100644 extensions/discord/src/voice/voice-session.startup.test.ts diff --git a/extensions/discord/src/voice/audio.lifecycle.test.ts b/extensions/discord/src/voice/audio.lifecycle.test.ts new file mode 100644 index 000000000000..b59041cd5ebd --- /dev/null +++ b/extensions/discord/src/voice/audio.lifecycle.test.ts @@ -0,0 +1,58 @@ +import { once } from "node:events"; +import type { OpusEncoderHandle } from "libopus-wasm"; +import { beforeEach, expect, it, vi } from "vitest"; + +const { createEncoderMock } = vi.hoisted(() => ({ createEncoderMock: vi.fn() })); +vi.mock("libopus-wasm", async (importOriginal) => ({ + ...(await importOriginal()), + createEncoder: createEncoderMock, +})); + +import { createDiscordOpusEncodeStream } from "./audio.js"; + +beforeEach(() => createEncoderMock.mockReset()); + +it("releases an encoder acquired after playback was destroyed without encoding queued audio", async () => { + const codec = await vi.importActual("libopus-wasm"); + const encoder = await codec.createEncoder({ channels: 2, sampleRate: 48_000 }); + const encode = vi.spyOn(encoder, "encode"); + const free = vi.spyOn(encoder, "free"); + let resolveEncoder!: (encoder: OpusEncoderHandle) => void; + createEncoderMock.mockReturnValueOnce( + new Promise((resolve) => { + resolveEncoder = resolve; + }), + ); + const stream = createDiscordOpusEncodeStream(); + try { + stream.write(Buffer.alloc(960 * 2 * 2)); + await vi.waitFor(() => expect(createEncoderMock).toHaveBeenCalledOnce()); + const closed = once(stream, "close"); + stream.destroy(); + resolveEncoder(encoder); + await closed; + + expect(free).toHaveBeenCalledOnce(); + expect(encode).not.toHaveBeenCalled(); + } finally { + stream.destroy(); + encoder.free(); + vi.restoreAllMocks(); + } +}); + +it("reports encoder initialization failures without producing queued audio", async () => { + const error = new Error("encoder initialization failed"); + createEncoderMock.mockRejectedValueOnce(error); + const stream = createDiscordOpusEncodeStream(); + const errors: Error[] = []; + stream.on("error", (err) => errors.push(err)); + const closed = new Promise((resolve) => { + stream.once("close", resolve); + }); + stream.end(Buffer.alloc(960 * 2 * 2)); + await closed; + + expect(errors).toEqual([error]); + expect(stream.read()).toBeNull(); +}); diff --git a/extensions/discord/src/voice/audio.test.ts b/extensions/discord/src/voice/audio.test.ts index e427cbbd9f7a..c2d30f3d8990 100644 --- a/extensions/discord/src/voice/audio.test.ts +++ b/extensions/discord/src/voice/audio.test.ts @@ -80,37 +80,39 @@ async function collectBuffers(stream: Readable): Promise { return chunks; } +const decodeModes = [ + { mode: "buffered", decode: decodeOpusStream }, + { + mode: "streaming", + decode: async (stream: Readable, params: Parameters[1]) => { + const chunks: Buffer[] = []; + await decodeOpusStreamChunks(stream, { ...params, onChunk: (chunk) => chunks.push(chunk) }); + return Buffer.concat(chunks); + }, + }, +]; + describe("discord voice opus codec", () => { - it("defaults to libopus-wasm for receive decoding", async () => { - const verbose: string[] = []; - const warnings: string[] = []; + it.each(decodeModes)( + "round-trips Discord PCM through $mode Opus decoding", + async ({ decode }) => { + const encoder = createDiscordOpusEncodeStream(); + const packetsPromise = collectBuffers(encoder); - const decoded = await decodeOpusStream(Readable.from([]), { - onVerbose: (message) => verbose.push(message), - onWarn: (message) => warnings.push(message), - }); + encoder.end(Buffer.alloc(960 * 2 * 2)); + const packets = await packetsPromise; - expect(decoded.length).toBe(0); - expect(verbose).toContain("opus decoder: libopus-wasm"); - expect(warnings).toEqual([]); - }); + expect(packets).toHaveLength(1); + expect(packets[0]?.length).toBeGreaterThan(0); - it("encodes raw Discord PCM into Opus packets for realtime playback", async () => { - const encoder = createDiscordOpusEncodeStream(); - const packetsPromise = collectBuffers(encoder); - - encoder.end(Buffer.alloc(960 * 2 * 2)); - const packets = await packetsPromise; - - expect(packets).toHaveLength(1); - expect(packets[0]?.length).toBeGreaterThan(0); - - const decoded = await decodeOpusStream(Readable.from(packets), { - onVerbose: vi.fn(), - onWarn: vi.fn(), - }); - expect(decoded.length).toBe(960 * 2 * 2); - }); + const onVerbose = vi.fn(); + const onWarn = vi.fn(); + const decoded = await decode(Readable.from(packets), { onVerbose, onWarn }); + expect(decoded.length).toBe(960 * 2 * 2); + expect(onVerbose).toHaveBeenCalledWith("opus decoder: libopus-wasm"); + expect(onWarn).not.toHaveBeenCalled(); + }, + ); it("pads final partial PCM frames before encoding", async () => { const encoder = createDiscordOpusEncodeStream(); @@ -120,26 +122,35 @@ describe("discord voice opus codec", () => { const packets = await packetsPromise; expect(packets).toHaveLength(1); - }); - - it("surfaces chunk decode stream failures to callers", async () => { - const err = new Error("memory access out of bounds"); - const onError = vi.fn(); - const stream = new Readable({ - read() { - this.destroy(err); - }, - }); - - await decodeOpusStreamChunks(stream, { - onChunk: vi.fn(), - onError, + const decoded = await decodeOpusStream(Readable.from(packets), { onVerbose: vi.fn(), onWarn: vi.fn(), }); - - expect(onError).toHaveBeenCalledWith(err); + expect(decoded).toHaveLength(960 * 2 * 2); }); + + it.each(decodeModes)( + "preserves decoded audio and reports $mode stream failures", + async ({ decode }) => { + const err = new Error("memory access out of bounds"); + const onError = vi.fn(); + const stream = Readable.from( + (async function* () { + yield Buffer.from([0xf8, 0xff, 0xfe]); + throw err; + })(), + ); + + const decoded = await decode(stream, { + onError, + onVerbose: vi.fn(), + onWarn: vi.fn(), + }); + + expect(onError).toHaveBeenCalledWith(err); + expect(decoded).toHaveLength(960 * 2 * 2); + }, + ); }); describe("createDiscordOpusPlaybackStream child stream errors", () => { diff --git a/extensions/discord/src/voice/audio.ts b/extensions/discord/src/voice/audio.ts index 9d66f6d860ea..be9453a71074 100644 --- a/extensions/discord/src/voice/audio.ts +++ b/extensions/discord/src/voice/audio.ts @@ -38,9 +38,10 @@ const FFMPEG_PCM_ARGUMENTS = [ String(CHANNELS), ]; -type OpusDecoder = { - decode: (buffer: Buffer) => Buffer | Promise; - free?: () => Promise | void; +type OpusDecodeCallbacks = { + onError?: (err: unknown) => void; + onVerbose: (message: string) => void; + onWarn: (message: string) => void; }; let warnedOpusMissing = false; @@ -65,39 +66,6 @@ function buildWavBuffer(pcm: Buffer): Buffer { return Buffer.concat([header, pcm]); } -async function createOpusDecoder(params: { - onWarn: (message: string) => void; -}): Promise<{ decoder: OpusDecoder; name: string } | null> { - let decoder: LibopusDecoder; - try { - decoder = await createLibopusDecoder({ - channels: CHANNELS, - sampleRate: SAMPLE_RATE, - }); - } catch (err) { - const failure = formatErrorMessage(err); - if (!warnedOpusMissing) { - warnedOpusMissing = true; - params.onWarn( - `discord voice: no usable opus decoder available (libopus-wasm: ${failure}); cannot decode voice audio`, - ); - } - return null; - } - return { - name: "libopus-wasm", - decoder: { - decode: (buffer) => - pcmInt16ToBuffer( - decoder.decode(buffer, { - maxFrameSize: DISCORD_OPUS_FRAME_SIZE, - }), - ), - free: () => decoder.free(), - }, - }; -} - export function createDiscordOpusEncodeStream(): Transform { return new DiscordOpusEncodeStream(); } @@ -175,83 +143,65 @@ export function createDiscordOpusPlaybackStream(input: Readable | string): Reada class DiscordOpusEncodeStream extends Transform { #buffer = Buffer.alloc(0); - #encoder: LibopusEncoder | null = null; - #encoderPromise: Promise | null = null; + #encoder!: LibopusEncoder; constructor() { super({ readableObjectMode: true }); } - async #getEncoder(): Promise { - if (!this.#encoderPromise) { - this.#encoderPromise = createLibopusEncoder({ - application: Application.Audio, - channels: CHANNELS, - sampleRate: SAMPLE_RATE, - }); - } - if (!this.#encoder) { - this.#encoder = await this.#encoderPromise; - } - return this.#encoder; + override _construct(done: (error?: Error | null) => void): void { + // Node defers transforms and destruction until construction settles, so a late + // encoder is released by _destroy without processing cancelled playback. + void createLibopusEncoder({ + application: Application.Audio, + channels: CHANNELS, + sampleRate: SAMPLE_RATE, + }).then( + (encoder) => { + this.#encoder = encoder; + done(); + }, + (err: unknown) => done(err instanceof Error ? err : new Error(formatErrorMessage(err))), + ); } override _transform(chunk: Buffer, _encoding: BufferEncoding, done: TransformCallback): void { - void (async () => { - try { - const encoder = await this.#getEncoder(); - this.#buffer = - this.#buffer.length > 0 ? Buffer.concat([this.#buffer, chunk]) : Buffer.from(chunk); - while (this.#buffer.length >= DISCORD_OPUS_FRAME_BYTES) { - const frame = this.#buffer.subarray(0, DISCORD_OPUS_FRAME_BYTES); - this.#buffer = this.#buffer.subarray(DISCORD_OPUS_FRAME_BYTES); - this.push( - Buffer.from( - encoder.encode(frame, { - frameSize: DISCORD_OPUS_FRAME_SIZE, - }), - ), - ); - } - done(); - } catch (err) { - done(err instanceof Error ? err : new Error(formatErrorMessage(err))); + try { + this.#buffer = + this.#buffer.length > 0 ? Buffer.concat([this.#buffer, chunk]) : Buffer.from(chunk); + while (this.#buffer.length >= DISCORD_OPUS_FRAME_BYTES) { + const frame = this.#buffer.subarray(0, DISCORD_OPUS_FRAME_BYTES); + this.#buffer = this.#buffer.subarray(DISCORD_OPUS_FRAME_BYTES); + this.#encodeFrame(frame); } - })(); + done(); + } catch (err) { + done(err instanceof Error ? err : new Error(formatErrorMessage(err))); + } } - override _final(done: TransformCallback): void { - void (async () => { - try { - if (this.#buffer.length > 0) { - const encoder = await this.#getEncoder(); - const frame = Buffer.alloc(DISCORD_OPUS_FRAME_BYTES); - this.#buffer.copy(frame); - this.#buffer = Buffer.alloc(0); - this.push( - Buffer.from( - encoder.encode(frame, { - frameSize: DISCORD_OPUS_FRAME_SIZE, - }), - ), - ); - } - this.#freeEncoder(); - done(); - } catch (err) { - done(err instanceof Error ? err : new Error(formatErrorMessage(err))); + override _flush(done: TransformCallback): void { + try { + if (this.#buffer.length > 0) { + const frame = Buffer.alloc(DISCORD_OPUS_FRAME_BYTES); + this.#buffer.copy(frame); + this.#buffer = Buffer.alloc(0); + this.#encodeFrame(frame); } - })(); + done(); + } catch (err) { + done(err instanceof Error ? err : new Error(formatErrorMessage(err))); + } } override _destroy(err: Error | null, done: (error?: Error | null) => void): void { - this.#freeEncoder(); + this.#encoder?.free(); + this.#buffer = Buffer.alloc(0); done(err); } - #freeEncoder(): void { - this.#encoder?.free(); - this.#encoder = null; + #encodeFrame(frame: Buffer): void { + this.push(Buffer.from(this.#encoder.encode(frame, { frameSize: DISCORD_OPUS_FRAME_SIZE }))); } } @@ -261,61 +211,40 @@ function pcmInt16ToBuffer(pcm: Int16Array): Buffer { export async function decodeOpusStream( stream: Readable, - params: { - onError?: (err: unknown) => void; - onVerbose: (message: string) => void; - onWarn: (message: string) => void; - }, + params: OpusDecodeCallbacks, ): Promise { - const selected = await createOpusDecoder({ onWarn: params.onWarn }); - if (!selected) { - return Buffer.alloc(0); - } - params.onVerbose(`opus decoder: ${selected.name}`); const chunks: Buffer[] = []; - try { - for await (const chunk of stream) { - if (!chunk || !(chunk instanceof Buffer) || chunk.length === 0) { - continue; - } - const decoded = await selected.decoder.decode(chunk); - if (decoded && decoded.length > 0) { - chunks.push(Buffer.from(decoded)); - } - } - } catch (err) { - params.onError?.(err); - if (shouldLogVerbose()) { - logVerbose(`discord voice: opus decode failed: ${formatErrorMessage(err)}`); - } - } finally { - await selected.decoder.free?.(); - } - return chunks.length > 0 ? Buffer.concat(chunks) : Buffer.alloc(0); + await decodeOpusStreamChunks(stream, { ...params, onChunk: (chunk) => chunks.push(chunk) }); + return Buffer.concat(chunks); } export async function decodeOpusStreamChunks( stream: Readable, - params: { + params: OpusDecodeCallbacks & { onChunk: (pcm48kStereo: Buffer) => void; - onError?: (err: unknown) => void; - onVerbose: (message: string) => void; - onWarn: (message: string) => void; }, ): Promise { - const selected = await createOpusDecoder({ onWarn: params.onWarn }); - if (!selected) { + let decoder: LibopusDecoder; + try { + decoder = await createLibopusDecoder({ channels: CHANNELS, sampleRate: SAMPLE_RATE }); + } catch (err) { + if (!warnedOpusMissing) { + warnedOpusMissing = true; + params.onWarn( + `discord voice: no usable opus decoder available (libopus-wasm: ${formatErrorMessage(err)}); cannot decode voice audio`, + ); + } return; } - params.onVerbose(`opus decoder: ${selected.name}`); + params.onVerbose("opus decoder: libopus-wasm"); try { for await (const chunk of stream) { if (!chunk || !(chunk instanceof Buffer) || chunk.length === 0) { continue; } - const decoded = await selected.decoder.decode(chunk); - if (decoded && decoded.length > 0) { - params.onChunk(Buffer.from(decoded)); + const decoded = decoder.decode(chunk, { maxFrameSize: DISCORD_OPUS_FRAME_SIZE }); + if (decoded.length > 0) { + params.onChunk(pcmInt16ToBuffer(decoded)); } } } catch (err) { @@ -324,7 +253,7 @@ export async function decodeOpusStreamChunks( logVerbose(`discord voice: opus decode failed: ${formatErrorMessage(err)}`); } } finally { - await selected.decoder.free?.(); + decoder.free(); } } diff --git a/extensions/discord/src/voice/voice-session.startup.test.ts b/extensions/discord/src/voice/voice-session.startup.test.ts new file mode 100644 index 000000000000..ffd9961a3b3b --- /dev/null +++ b/extensions/discord/src/voice/voice-session.startup.test.ts @@ -0,0 +1,117 @@ +import { defineDiscordVoiceTests } from "./voice-test-harness.test-support.js"; + +defineDiscordVoiceTests( + ({ + expect, + it, + createConnectionMock, + joinVoiceChannelMock, + realtimeSessionMock, + createAgentProxyManager, + getLastAudioPlayer, + lastRealtimeBridgeParams, + }) => { + it("releases an initial voice session when provider connect fails", async () => { + const connection = createConnectionMock(); + joinVoiceChannelMock.mockReturnValueOnce(connection); + realtimeSessionMock.connect.mockRejectedValueOnce(new Error("provider unavailable")); + const manager = createAgentProxyManager(); + + try { + await expect(manager.join({ guildId: "g1", channelId: "1001" })).resolves.toEqual({ + ok: false, + message: "Failed to start Discord realtime voice: provider unavailable", + guildId: "g1", + channelId: "1001", + }); + + expect(manager.status()).toEqual([]); + expect(connection.destroy).toHaveBeenCalledTimes(1); + expect(getLastAudioPlayer().stop).toHaveBeenCalledWith(true); + expect(realtimeSessionMock.close).toHaveBeenCalled(); + } finally { + await manager.destroy(); + } + expect(connection.destroy).toHaveBeenCalledTimes(1); + }); + + it("releases initial startup when the manager is destroyed during provider connect", async () => { + const connection = createConnectionMock(); + joinVoiceChannelMock.mockReturnValueOnce(connection); + const manager = createAgentProxyManager(); + realtimeSessionMock.connect.mockImplementationOnce(async () => { + await manager.destroy(); + }); + + try { + await expect(manager.join({ guildId: "g1", channelId: "1001" })).resolves.toEqual({ + ok: false, + message: "Discord realtime voice session stopped before startup completed.", + guildId: "g1", + channelId: "1001", + }); + + expect(manager.status()).toEqual([]); + expect(realtimeSessionMock.close).toHaveBeenCalled(); + expect(connection.destroy).toHaveBeenCalledTimes(1); + expect(getLastAudioPlayer().stop).toHaveBeenCalledWith(true); + } finally { + await manager.destroy(); + } + }); + + it("does not activate capture if Discord destroys the connection during initial startup", async () => { + const connection = createConnectionMock(); + joinVoiceChannelMock.mockReturnValueOnce(connection); + const manager = createAgentProxyManager(); + realtimeSessionMock.connect.mockImplementationOnce(async () => { + connection.state.status = "destroyed"; + connection.handlers.get("destroyed")?.(); + }); + + try { + await expect(manager.join({ guildId: "g1", channelId: "1001" })).resolves.toEqual({ + ok: false, + message: "Discord realtime voice session stopped before startup completed.", + guildId: "g1", + channelId: "1001", + }); + + expect(manager.status()).toEqual([]); + expect(connection.receiver.speaking.on).not.toHaveBeenCalled(); + expect(realtimeSessionMock.close).toHaveBeenCalled(); + expect(getLastAudioPlayer().stop).toHaveBeenCalledWith(true); + } finally { + await manager.destroy(); + } + expect(connection.destroy).not.toHaveBeenCalled(); + }); + + it("handles terminal provider output before the initial realtime connect finishes", async () => { + const connection = createConnectionMock(); + joinVoiceChannelMock.mockReturnValueOnce(connection); + const manager = createAgentProxyManager(); + realtimeSessionMock.connect.mockImplementationOnce(async () => { + // Provider output can arrive while connect is pending; exceed the two-minute PCM cap. + lastRealtimeBridgeParams().audioSink.sendAudio(Buffer.alloc(24_000 * 2 * 121)); + }); + + try { + await expect(manager.join({ guildId: "g1", channelId: "1001" })).resolves.toEqual({ + ok: false, + message: "Discord realtime voice session stopped before startup completed.", + guildId: "g1", + channelId: "1001", + }); + + expect(manager.status()).toEqual([]); + expect(realtimeSessionMock.close).toHaveBeenCalled(); + expect(connection.destroy).toHaveBeenCalledTimes(1); + expect(getLastAudioPlayer().stop).toHaveBeenCalledWith(true); + } finally { + await manager.destroy(); + } + expect(connection.destroy).toHaveBeenCalledTimes(1); + }); + }, +); diff --git a/extensions/discord/src/voice/voice-session.ts b/extensions/discord/src/voice/voice-session.ts index 87edab19b0c1..36ee698946de 100644 --- a/extensions/discord/src/voice/voice-session.ts +++ b/extensions/discord/src/voice/voice-session.ts @@ -259,7 +259,6 @@ export class DiscordVoiceSessions { const existingEntry = this.params.sessions.get(guildId); if (existingEntry) { existingEntry.stop(); - this.params.sessions.delete(guildId); } const voiceConnectionGroup = resolveVoiceConnectionGroup(this.params.accountId); const staleConnection = voiceSdk.getVoiceConnection(guildId, voiceConnectionGroup); @@ -384,37 +383,22 @@ export class DiscordVoiceSessions { }) : voiceSdk.createAudioPlayer(); connection.subscribe(player); - const clearSessionIfCurrent = () => { - const active = this.params.sessions.get(guildId); - if (active?.connection === connection) { - this.params.sessions.delete(guildId); - } - }; - const stopEntry = ( - entry: VoiceSessionEntry, - optionsLocal: { destroyConnection: boolean; reason: string }, - ) => { + const stopEntry = (optionsLocal: { destroyConnection: boolean; reason: string }) => { if (entry.sessionLifecycle.status === "stopped") { return; } entry.sessionLifecycle = { status: "stopped", reason: optionsLocal.reason }; + // A late callback from an old connection must not remove its replacement. + if (this.params.sessions.get(guildId) === entry) { + this.params.sessions.delete(guildId); + } this.params.membership.deactivate(entry); - if (speakingHandler) { - connection.receiver.speaking.off("start", speakingHandler); - } - if (speakingEndHandler) { - connection.receiver.speaking.off("end", speakingEndHandler); - } + connection.receiver.speaking.off("start", speakingHandler); + connection.receiver.speaking.off("end", speakingEndHandler); stopVoiceCaptureState(entry.capture); - if (disconnectedHandler) { - connection.off(voiceSdk.VoiceConnectionStatus.Disconnected, disconnectedHandler); - } - if (destroyedHandler) { - connection.off(voiceSdk.VoiceConnectionStatus.Destroyed, destroyedHandler); - } - if (playerErrorHandler) { - player.off("error", playerErrorHandler); - } + connection.off(voiceSdk.VoiceConnectionStatus.Disconnected, disconnectedHandler); + connection.off(voiceSdk.VoiceConnectionStatus.Destroyed, destroyedHandler); + player.off("error", playerErrorHandler); const realtimeLifecycle = entry.realtimeLifecycle; if (realtimeLifecycle.status === "starting" || realtimeLifecycle.status === "active") { realtimeLifecycle.instance.close(); @@ -466,57 +450,23 @@ export class DiscordVoiceSessions { receiveRecovery: createVoiceReceiveRecoveryState(), realtimeLifecycle: { status: "inactive", generation: 0 }, stop(reason) { - clearSessionIfCurrent(); - stopEntry(entry, { + stopEntry({ destroyConnection: true, reason: reason ?? `stop guild ${guildId} channel ${channelId}`, }); }, }; - if (!options?.transcripts && isDiscordRealtimeVoiceMode(voiceMode)) { - const realtimeResult = await this.attachRealtimeSession(entry, voiceMode, { - isCurrent: authority?.isCurrent, - }); - if (!realtimeResult.ok) { - destroyVoiceConnectionSafely({ - connection, - voiceSdk, - reason: `realtime setup failed guild ${guildId} channel ${channelId}`, - }); - return { - ok: false, - message: realtimeResult.message, - guildId, - channelId, - }; - } - } - if (this.params.destroyed() || (authority && !authority.isCurrent())) { - stopEntry(entry, { - destroyConnection: true, - reason: `${this.params.destroyed() ? "manager stopped" : "join cancelled"} during setup guild ${guildId} channel ${channelId}`, - }); - return { - ok: false, - message: this.params.destroyed() - ? "Discord voice manager is stopped." - : "Discord voice join was cancelled.", - guildId, - channelId, - }; - } - - const speakingHandler: ((userId: string) => void) | undefined = (userId: string) => { + const speakingHandler = (userId: string) => { void this.params.receive.handleSpeakingStart(entry, userId).catch((err: unknown) => { logger.warn(`discord voice: capture failed: ${formatErrorMessage(err)}`); }); }; - const speakingEndHandler: ((userId: string) => void) | undefined = (userId: string) => { + const speakingEndHandler = (userId: string) => { this.params.receive.scheduleCaptureFinalize(entry, userId, "speaker end"); }; - const disconnectedHandler: (() => void) | undefined = () => { + const disconnectedHandler = () => { void (async () => { try { logVoiceVerbose( @@ -539,25 +489,60 @@ export class DiscordVoiceSessions { logger.warn( `discord voice: disconnect recovery failed: guild ${guildId} channel ${channelId} timeout=${reconnectGraceMs}ms error=${formatErrorMessage(err)}; destroying connection`, ); - clearSessionIfCurrent(); - stopEntry(entry, { + stopEntry({ destroyConnection: true, reason: `disconnect recovery failed guild ${guildId} channel ${channelId}`, }); } })(); }; - const destroyedHandler: (() => void) | undefined = () => { - clearSessionIfCurrent(); - stopEntry(entry, { + const destroyedHandler = () => { + stopEntry({ destroyConnection: false, reason: `destroyed guild ${guildId} channel ${channelId}`, }); }; - const playerErrorHandler: ((err: Error) => void) | undefined = (err: Error) => { + const playerErrorHandler = (err: Error) => { logger.warn(`discord voice: playback error: ${formatErrorMessage(err)}`); }; + // Realtime callbacks can stop playback during connect. Initialize teardown and + // observe connection/player failure before startup, but admit capture only once ready. + connection.on(voiceSdk.VoiceConnectionStatus.Disconnected, disconnectedHandler); + connection.on(voiceSdk.VoiceConnectionStatus.Destroyed, destroyedHandler); + player.on("error", playerErrorHandler); + if (!options?.transcripts && isDiscordRealtimeVoiceMode(voiceMode)) { + const realtimeResult = await this.attachRealtimeSession(entry, voiceMode, { + isCurrent: authority?.isCurrent, + }); + if (!realtimeResult.ok) { + entry.stop(`realtime setup failed guild ${guildId} channel ${channelId}`); + return { + ok: false, + message: realtimeResult.message, + guildId, + channelId, + }; + } + } + if ( + isVoiceSessionStopped(entry) || + this.params.destroyed() || + (authority && !authority.isCurrent()) + ) { + entry.stop( + `${this.params.destroyed() ? "manager stopped" : "join cancelled"} during setup guild ${guildId} channel ${channelId}`, + ); + return { + ok: false, + message: this.params.destroyed() + ? "Discord voice manager is stopped." + : "Discord voice join was cancelled.", + guildId, + channelId, + }; + } + this.params.receive.enableDaveReceivePassthrough( entry, "post-join warmup", @@ -565,9 +550,6 @@ export class DiscordVoiceSessions { ); connection.receiver.speaking.on("start", speakingHandler); connection.receiver.speaking.on("end", speakingEndHandler); - connection.on(voiceSdk.VoiceConnectionStatus.Disconnected, disconnectedHandler); - connection.on(voiceSdk.VoiceConnectionStatus.Destroyed, destroyedHandler); - player.on("error", playerErrorHandler); this.params.sessions.set(guildId, entry); this.params.membership.activate(entry, this.params.botUserId()); @@ -618,7 +600,6 @@ export class DiscordVoiceSessions { } } entry.stop(); - this.params.sessions.delete(guildId); if (!entry.receiveRecovery.decryptRecoveryInFlight) { this.params.receive.deleteRecoveryAttempt(guildId); }