refactor(discord): centralize voice resource lifecycle (#130415)

* refactor(discord): centralize voice resource lifecycle

* test(discord): avoid returning from voice promise executor

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<typeof import("libopus-wasm")>()),
+  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<typeof import("libopus-wasm")>("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<OpusEncoderHandle>((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<void>((resolve) => {
+    stream.once("close", resolve);
+  });
+  stream.end(Buffer.alloc(960 * 2 * 2));
+  await closed;
+
+  expect(errors).toEqual([error]);
+  expect(stream.read()).toBeNull();
+});

extensions/discord/src/voice/audio.test.ts

@@ -80,37 +80,39 @@ async function collectBuffers(stream: Readable): Promise<Buffer[]> {
   return chunks;
 }
 
+const decodeModes = [
+  { mode: "buffered", decode: decodeOpusStream },
+  {
+    mode: "streaming",
+    decode: async (stream: Readable, params: Parameters<typeof decodeOpusStream>[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", () => {

extensions/discord/src/voice/audio.ts

@@ -38,9 +38,10 @@ const FFMPEG_PCM_ARGUMENTS = [
   String(CHANNELS),
 ];
 
-type OpusDecoder = {
-  decode: (buffer: Buffer) => Buffer | Promise<Buffer>;
-  free?: () => Promise<void> | 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<LibopusEncoder> | null = null;
+  #encoder!: LibopusEncoder;
 
   constructor() {
     super({ readableObjectMode: true });
   }
 
-  async #getEncoder(): Promise<LibopusEncoder> {
-    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<Buffer> {
-  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<void> {
-  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();
   }
 }
 

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);
+    });
+  },
+);

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);
     }