From 30972edaea784ccd501320b5b00f3a17e8e2ebda Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 1 Aug 2026 08:57:12 -0700 Subject: [PATCH] fix(discord): recover zero-valued DAVE transition (#117381) Co-authored-by: Peter Steinberger --- .../discord/src/voice/manager.e2e.test.ts | 635 +++++++++++++++++- extensions/discord/src/voice/manager.ts | 72 +- .../src/voice/receive-recovery.test.ts | 289 ++++++++ .../discord/src/voice/receive-recovery.ts | 40 +- 4 files changed, 1033 insertions(+), 3 deletions(-) diff --git a/extensions/discord/src/voice/manager.e2e.test.ts b/extensions/discord/src/voice/manager.e2e.test.ts index 204b91a2c9ea..3effe9bb3921 100644 --- a/extensions/discord/src/voice/manager.e2e.test.ts +++ b/extensions/discord/src/voice/manager.e2e.test.ts @@ -1,5 +1,7 @@ import { PassThrough, type Readable } from "node:stream"; +import { DAVESession } from "@discordjs/voice"; import { expectDefined } from "@openclaw/normalization-core"; +import { VoiceOpcodes, type VoiceSendPayload } from "discord-api-types/voice/v8"; import { createOpenClawCodingTools } from "openclaw/plugin-sdk/agent-harness"; import type { RealtimeVoiceAgentControlResult, @@ -19,7 +21,7 @@ import { type TestRealtimeBridgeParams, type TestRealtimeSessionEntry, } from "./manager.e2e.test-support.js"; -import { createVoiceReceiveRecoveryState } from "./receive-recovery.js"; +import { createVoiceReceiveRecoveryState, DECRYPT_FAILURE_WINDOW_MS } from "./receive-recovery.js"; const { createConnectionMock, @@ -64,6 +66,9 @@ const { state: { code: string; dave: { + lastTransitionId?: number; + reinitializing?: boolean; + recoverFromInvalidTransition?: ReturnType; session: { setPassthroughMode: ReturnType; }; @@ -632,6 +637,56 @@ describe("DiscordVoiceManager", () => { ); }; + const installFailingDaveSession = ( + connection: ReturnType, + failure: "invalidation" | "native" | "key-package", + beforeFailure?: () => void, + ) => { + const dave = new DAVESession(1, "bot", "1001", { decryptionFailureTolerance: 0 }); + const nativeSession = { + decrypt: vi.fn(() => { + throw new Error("UnencryptedWhenPassthroughDisabled"); + }), + getSerializedKeyPackage: vi.fn(() => Buffer.from("new-key-package")), + ready: true, + reinit: vi.fn(() => { + if (failure === "native") { + beforeFailure?.(); + throw new Error("native DAVE reinitialization failed"); + } + }), + setPassthroughMode: connection.daveSetPassthroughMode, + }; + dave.session = nativeSession as unknown as NonNullable; + dave.lastTransitionId = 0; + const gateway = { + sendPacket: vi.fn((_packet: VoiceSendPayload) => { + if (failure === "invalidation") { + beforeFailure?.(); + throw new Error("voice gateway invalidation failed"); + } + }), + sendBinaryMessage: vi.fn((_opcode: VoiceOpcodes, _keyPackage: Buffer) => { + if (failure === "key-package") { + beforeFailure?.(); + throw new Error("voice gateway key-package delivery failed"); + } + }), + }; + dave.on("invalidateTransition", (transitionId) => { + gateway.sendPacket({ + op: VoiceOpcodes.DaveMlsInvalidCommitWelcome, + d: { transition_id: transitionId }, + }); + }); + dave.on("keyPackage", (keyPackage) => { + gateway.sendBinaryMessage(VoiceOpcodes.DaveMlsKeyPackage, keyPackage); + }); + connection.state.networking.state.dave = + dave as unknown as typeof connection.state.networking.state.dave; + return { dave, gateway }; + }; + it("rejects joins when Discord voice config is absent", async () => { const manager = createManager({}); @@ -830,6 +885,9 @@ describe("DiscordVoiceManager", () => { onUtterance, }); expect(entry.realtime).toBeTruthy(); + const attempts = (manager as unknown as { daveRecoveryAttempts: Map }) + .daveRecoveryAttempts; + attempts.set("g1", Date.now()); const stopNotesResult = await manager.leave( { guildId: "g1", channelId: "1001" }, @@ -840,6 +898,7 @@ describe("DiscordVoiceManager", () => { expect(entry.transcripts).toBeUndefined(); expect(entry.realtime).toBeTruthy(); expect(realtimeSessionMock.close).not.toHaveBeenCalled(); + expect(attempts.has("g1")).toBe(true); expectConnectedStatus(manager, "1001"); }); @@ -4527,6 +4586,576 @@ describe("DiscordVoiceManager", () => { expect(connection.daveSetPassthroughMode).toHaveBeenCalledWith(true, 30); }); + it("invalidates transition zero before re-arming receive passthrough", async () => { + const connection = createConnectionMock(); + const dave = connection.state.networking.state.dave; + dave.lastTransitionId = 0; + dave.reinitializing = false; + dave.recoverFromInvalidTransition = vi.fn(); + joinVoiceChannelMock.mockReturnValueOnce(connection); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + connection.daveSetPassthroughMode.mockClear(); + + emitDecryptFailure(manager); + + expect(dave.recoverFromInvalidTransition).toHaveBeenCalledOnce(); + expect(dave.recoverFromInvalidTransition).toHaveBeenCalledWith(0); + expect(connection.daveSetPassthroughMode).toHaveBeenCalledWith(true, 15); + expect(dave.recoverFromInvalidTransition.mock.invocationCallOrder[0]).toBeLessThan( + connection.daveSetPassthroughMode.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY, + ); + }); + + it.each([ + { + label: "non-zero transitions", + lastTransitionId: 1, + reinitializing: false, + networkingStatus: "networking-ready", + }, + { + label: "missing transitions", + lastTransitionId: undefined, + reinitializing: false, + networkingStatus: "networking-ready", + }, + { + label: "transitions already reinitializing", + lastTransitionId: 0, + reinitializing: true, + networkingStatus: "networking-ready", + }, + { + label: "resuming networking", + lastTransitionId: 0, + reinitializing: false, + networkingStatus: "networking-resuming", + }, + ])( + "does not invalidate $label", + async ({ lastTransitionId, reinitializing, networkingStatus }) => { + const connection = createConnectionMock(); + const dave = connection.state.networking.state.dave; + dave.lastTransitionId = lastTransitionId; + dave.reinitializing = reinitializing; + dave.recoverFromInvalidTransition = vi.fn(); + joinVoiceChannelMock.mockReturnValueOnce(connection); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + connection.state.networking.state.code = networkingStatus; + + emitDecryptFailure(manager); + + expect(dave.recoverFromInvalidTransition).not.toHaveBeenCalled(); + }, + ); + + it("does not invalidate a stale voice-session transition", async () => { + const staleConnection = createConnectionMock(); + const staleDave = staleConnection.state.networking.state.dave; + staleDave.lastTransitionId = 0; + staleDave.reinitializing = false; + staleDave.recoverFromInvalidTransition = vi.fn(); + joinVoiceChannelMock + .mockReturnValueOnce(staleConnection) + .mockReturnValueOnce(createConnectionMock()); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + const staleEntry = getSessionEntry(manager); + await manager.join({ guildId: "g1", channelId: "1002" }); + + ( + manager as unknown as { handleReceiveError: (entry: unknown, err: unknown) => void } + ).handleReceiveError( + staleEntry, + new Error("Failed to decrypt: DecryptionFailed(UnencryptedWhenPassthroughDisabled)"), + ); + + expect(staleDave.recoverFromInvalidTransition).not.toHaveBeenCalled(); + }); + + it("does not invalidate a stopped voice-session transition", async () => { + const connection = createConnectionMock(); + const dave = connection.state.networking.state.dave; + dave.lastTransitionId = 0; + dave.reinitializing = false; + dave.recoverFromInvalidTransition = vi.fn(); + joinVoiceChannelMock.mockReturnValueOnce(connection); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + const entry = getSessionEntry(manager) as TestRealtimeSessionEntry & { + isStopped: () => boolean; + }; + entry.isStopped = () => true; + + emitDecryptFailure(manager); + + expect(dave.recoverFromInvalidTransition).not.toHaveBeenCalled(); + }); + + it("does not invalidate transition zero for unrelated receive failures", async () => { + const connection = createConnectionMock(); + const dave = connection.state.networking.state.dave; + dave.lastTransitionId = 0; + dave.reinitializing = false; + dave.recoverFromInvalidTransition = vi.fn(); + joinVoiceChannelMock.mockReturnValueOnce(connection); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + ( + manager as unknown as { handleReceiveError: (entry: unknown, err: unknown) => void } + ).handleReceiveError( + getSessionEntry(manager), + new Error("DecryptionFailed(InvalidCiphertext)"), + ); + + expect(dave.recoverFromInvalidTransition).not.toHaveBeenCalled(); + }); + + it("keeps passthrough and bounded rejoin when zero-transition recovery throws", async () => { + const connection = createConnectionMock(); + const dave = connection.state.networking.state.dave; + dave.lastTransitionId = 0; + dave.reinitializing = false; + dave.recoverFromInvalidTransition = vi.fn(() => { + throw new Error("voice gateway unavailable"); + }); + joinVoiceChannelMock + .mockReturnValueOnce(connection) + .mockReturnValueOnce(createConnectionMock()); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + connection.daveSetPassthroughMode.mockClear(); + + emitDecryptFailure(manager); + emitDecryptFailure(manager); + emitDecryptFailure(manager); + + await vi.waitFor(() => { + expect(connection.daveSetPassthroughMode).toHaveBeenCalledWith(true, 15); + expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2); + }); + }); + + it.each([ + { label: "gateway invalidation", failure: "invalidation" as const }, + { label: "native DAVE reinitialization", failure: "native" as const }, + { label: "MLS key-package delivery", failure: "key-package" as const }, + ])( + "immediately rejoins after $label leaves the real DAVE session poisoned", + async ({ failure }) => { + const connection = createConnectionMock(); + const { dave, gateway } = installFailingDaveSession(connection, failure); + joinVoiceChannelMock + .mockReturnValueOnce(connection) + .mockReturnValueOnce(createConnectionMock()); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + connection.daveSetPassthroughMode.mockClear(); + expect(() => dave.decrypt(Buffer.from("encrypted-audio"), "speaker")).toThrow( + "UnencryptedWhenPassthroughDisabled", + ); + + emitDecryptFailure(manager); + + expect(dave.reinitializing).toBe(true); + expect(gateway.sendPacket).toHaveBeenCalledWith({ + op: VoiceOpcodes.DaveMlsInvalidCommitWelcome, + d: { transition_id: 0 }, + }); + expect(gateway.sendBinaryMessage).toHaveBeenCalledTimes(failure === "key-package" ? 1 : 0); + expect(connection.daveSetPassthroughMode).not.toHaveBeenCalled(); + expect(dave.decrypt(Buffer.from("encrypted-audio"), "speaker")).toBeNull(); + expect(connection.destroy).toHaveBeenCalledOnce(); + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2)); + }, + ); + + it("does not duplicate an in-flight reconnect after a real DAVE recovery fails", async () => { + const connection = createConnectionMock(); + const { dave } = installFailingDaveSession(connection, "native"); + joinVoiceChannelMock.mockReturnValueOnce(connection); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + const entry = getSessionEntry(manager); + entry.receiveRecovery.decryptRecoveryInFlight = true; + connection.daveSetPassthroughMode.mockClear(); + + emitDecryptFailure(manager); + + expect(dave.reinitializing).toBe(true); + expect(entry.receiveRecovery.decryptRecoveryInFlight).toBe(true); + expect(connection.destroy).not.toHaveBeenCalled(); + expect(connection.daveSetPassthroughMode).not.toHaveBeenCalled(); + expect(joinVoiceChannelMock).toHaveBeenCalledOnce(); + }); + + it("does not rejoin a voice session stopped during real DAVE recovery", async () => { + const connection = createConnectionMock(); + const stopEntry: { current?: () => void } = {}; + const { dave } = installFailingDaveSession(connection, "native", () => stopEntry.current?.()); + joinVoiceChannelMock.mockReturnValueOnce(connection); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + const entry = getSessionEntry(manager); + stopEntry.current = () => entry.stop(); + connection.daveSetPassthroughMode.mockClear(); + + emitDecryptFailure(manager); + + expect(dave.reinitializing).toBe(true); + expect(connection.destroy).toHaveBeenCalledOnce(); + expect(connection.daveSetPassthroughMode).not.toHaveBeenCalled(); + expect(joinVoiceChannelMock).toHaveBeenCalledOnce(); + expect(entry.receiveRecovery.decryptRecoveryInFlight).toBe(false); + }); + + it("disconnects after repeated poisoned DAVE sessions without a reconnect loop", async () => { + const firstConnection = createConnectionMock(); + const secondConnection = createConnectionMock(); + installFailingDaveSession(firstConnection, "native"); + installFailingDaveSession(secondConnection, "key-package"); + joinVoiceChannelMock.mockReturnValueOnce(firstConnection).mockReturnValueOnce(secondConnection); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + emitDecryptFailure(manager); + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2)); + secondConnection.daveSetPassthroughMode.mockClear(); + + emitDecryptFailure(manager); + + expect(firstConnection.destroy).toHaveBeenCalledOnce(); + expect(secondConnection.destroy).toHaveBeenCalledOnce(); + expect(secondConnection.daveSetPassthroughMode).not.toHaveBeenCalled(); + expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2); + expect(manager.status()).toEqual([]); + }); + + it("suppresses followed-user reconciliation until the poisoned-DAVE cooldown expires", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + const firstConnection = createConnectionMock(); + const secondConnection = createConnectionMock(); + installFailingDaveSession(firstConnection, "native"); + installFailingDaveSession(secondConnection, "key-package"); + joinVoiceChannelMock + .mockReturnValueOnce(firstConnection) + .mockReturnValueOnce(secondConnection) + .mockReturnValueOnce(createConnectionMock()); + const client = createClient(); + client.rest.get.mockResolvedValue({ + guild_id: "g1", + user_id: "u-owner", + channel_id: "1001", + }); + const manager = createManager( + { + guilds: { g1: {} }, + voice: { + enabled: true, + mode: "stt-tts", + followUsers: ["u-owner"], + }, + }, + client, + ); + + try { + await manager.autoJoin(); + emitDecryptFailure(manager); + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2)); + emitDecryptFailure(manager); + expect(manager.status()).toEqual([]); + + await vi.advanceTimersByTimeAsync(10_000); + + expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2); + const followedUsers = ( + manager as unknown as { followedUserChannels: Map } + ).followedUserChannels; + expect(followedUsers.get("g1:u-owner")?.channelId).toBe("1001"); + + await vi.advanceTimersByTimeAsync(20_000); + + expect(joinVoiceChannelMock).toHaveBeenCalledTimes(3); + expectConnectedStatus(manager, "1001"); + } finally { + await manager.destroy(); + vi.useRealTimers(); + } + }); + + it("suppresses repeated same-channel voice-state updates during a DAVE cooldown", async () => { + const firstConnection = createConnectionMock(); + const secondConnection = createConnectionMock(); + installFailingDaveSession(firstConnection, "native"); + installFailingDaveSession(secondConnection, "key-package"); + joinVoiceChannelMock.mockReturnValueOnce(firstConnection).mockReturnValueOnce(secondConnection); + const manager = createManager({ + voice: { + enabled: true, + mode: "stt-tts", + followUsers: ["u-owner"], + }, + }); + + await updateVoiceState(manager, "u-owner", "1001"); + emitDecryptFailure(manager); + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2)); + emitDecryptFailure(manager); + const previousVoiceState = { + guild_id: "g1", + user_id: "u-owner", + channel_id: "1001", + }; + + await manager.handleVoiceStateUpdate( + { ...previousVoiceState, self_mute: true } as never, + previousVoiceState as never, + ); + await manager.handleVoiceStateUpdate( + { ...previousVoiceState, self_deaf: true } as never, + { ...previousVoiceState, self_mute: true } as never, + ); + + expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2); + expect(manager.status()).toEqual([]); + }); + + it("still follows real user movement to another channel during a DAVE cooldown", async () => { + const firstConnection = createConnectionMock(); + const secondConnection = createConnectionMock(); + installFailingDaveSession(firstConnection, "native"); + installFailingDaveSession(secondConnection, "key-package"); + joinVoiceChannelMock + .mockReturnValueOnce(firstConnection) + .mockReturnValueOnce(secondConnection) + .mockReturnValueOnce(createConnectionMock()); + const manager = createManager({ + voice: { + enabled: true, + mode: "stt-tts", + followUsers: ["u-owner"], + }, + }); + + await updateVoiceState(manager, "u-owner", "1001"); + emitDecryptFailure(manager); + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2)); + emitDecryptFailure(manager); + expect(manager.status()).toEqual([]); + + await updateVoiceState(manager, "u-owner", "1002"); + + expect(joinVoiceChannelMock).toHaveBeenCalledTimes(3); + expectConnectedStatus(manager, "1002"); + }); + + it("follows a user who leaves and rejoins the same channel during a DAVE cooldown", async () => { + const firstConnection = createConnectionMock(); + const secondConnection = createConnectionMock(); + installFailingDaveSession(firstConnection, "native"); + installFailingDaveSession(secondConnection, "key-package"); + joinVoiceChannelMock + .mockReturnValueOnce(firstConnection) + .mockReturnValueOnce(secondConnection) + .mockReturnValueOnce(createConnectionMock()); + const manager = createManager({ + voice: { + enabled: true, + mode: "stt-tts", + followUsers: ["u-owner"], + }, + }); + + await updateVoiceState(manager, "u-owner", "1001"); + emitDecryptFailure(manager); + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2)); + emitDecryptFailure(manager); + + await updateVoiceState(manager, "u-owner", null); + await updateVoiceState(manager, "u-owner", "1001"); + + expect(joinVoiceChannelMock).toHaveBeenCalledTimes(3); + expectConnectedStatus(manager, "1001"); + }); + + it("reconciles a followed-user move to another channel during a DAVE cooldown", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + const firstConnection = createConnectionMock(); + const secondConnection = createConnectionMock(); + installFailingDaveSession(firstConnection, "native"); + installFailingDaveSession(secondConnection, "key-package"); + joinVoiceChannelMock + .mockReturnValueOnce(firstConnection) + .mockReturnValueOnce(secondConnection) + .mockReturnValueOnce(createConnectionMock()); + const client = createClient(); + client.rest.get.mockResolvedValue({ + guild_id: "g1", + user_id: "u-owner", + channel_id: "1001", + }); + const manager = createManager( + { + guilds: { g1: {} }, + voice: { + enabled: true, + mode: "stt-tts", + followUsers: ["u-owner"], + }, + }, + client, + ); + + try { + await manager.autoJoin(); + emitDecryptFailure(manager); + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2)); + emitDecryptFailure(manager); + client.rest.get.mockResolvedValue({ + guild_id: "g1", + user_id: "u-owner", + channel_id: "1002", + }); + + await vi.advanceTimersByTimeAsync(10_000); + + expect(joinVoiceChannelMock).toHaveBeenCalledTimes(3); + expectConnectedStatus(manager, "1002"); + } finally { + await manager.destroy(); + vi.useRealTimers(); + } + }); + + it("allows explicit manual joins during a poisoned-DAVE cooldown", async () => { + const firstConnection = createConnectionMock(); + const secondConnection = createConnectionMock(); + installFailingDaveSession(firstConnection, "native"); + installFailingDaveSession(secondConnection, "key-package"); + joinVoiceChannelMock + .mockReturnValueOnce(firstConnection) + .mockReturnValueOnce(secondConnection) + .mockReturnValueOnce(createConnectionMock()); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + emitDecryptFailure(manager); + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2)); + emitDecryptFailure(manager); + expect(manager.status()).toEqual([]); + + expect((await manager.join({ guildId: "g1", channelId: "1001" })).ok).toBe(true); + expect(joinVoiceChannelMock).toHaveBeenCalledTimes(3); + }); + + it("clears the poisoned-DAVE recovery budget after an intentional full leave", async () => { + const firstConnection = createConnectionMock(); + const recoveredConnection = createConnectionMock(); + const manuallyJoinedConnection = createConnectionMock(); + const lastConnection = createConnectionMock(); + installFailingDaveSession(firstConnection, "native"); + installFailingDaveSession(manuallyJoinedConnection, "native"); + joinVoiceChannelMock + .mockReturnValueOnce(firstConnection) + .mockReturnValueOnce(recoveredConnection) + .mockReturnValueOnce(manuallyJoinedConnection) + .mockReturnValueOnce(lastConnection); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + emitDecryptFailure(manager); + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2)); + expect((await manager.leave({ guildId: "g1" })).ok).toBe(true); + + await manager.join({ guildId: "g1", channelId: "1001" }); + emitDecryptFailure(manager); + + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(4)); + expect(lastConnection.destroy).not.toHaveBeenCalled(); + }); + + it("allows a poisoned-DAVE reconnect after the existing failure window expires", async () => { + const firstConnection = createConnectionMock(); + installFailingDaveSession(firstConnection, "native"); + joinVoiceChannelMock + .mockReturnValueOnce(firstConnection) + .mockReturnValueOnce(createConnectionMock()); + const manager = createManager(); + + await manager.join({ guildId: "g1", channelId: "1001" }); + const attempts = (manager as unknown as { daveRecoveryAttempts: Map }) + .daveRecoveryAttempts; + attempts.set("g1", Date.now() - DECRYPT_FAILURE_WINDOW_MS); + attempts.set("other-guild", Date.now()); + + emitDecryptFailure(manager); + + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(2)); + expect(attempts.has("other-guild")).toBe(true); + }); + + it("keeps poisoned-DAVE reconnect budgets isolated between guilds", async () => { + const firstGuildConnection = createConnectionMock(); + const secondGuildConnection = createConnectionMock(); + installFailingDaveSession(firstGuildConnection, "native"); + installFailingDaveSession(secondGuildConnection, "key-package"); + joinVoiceChannelMock + .mockReturnValueOnce(firstGuildConnection) + .mockReturnValueOnce(secondGuildConnection) + .mockReturnValueOnce(createConnectionMock()) + .mockReturnValueOnce(createConnectionMock()); + const client = createClient(); + client.fetchChannel.mockImplementation(async (channelId: string) => { + const guildId = channelId === "2001" ? "g2" : "g1"; + return { + id: channelId, + guildId, + guild: { id: guildId, name: guildId }, + type: ChannelType.GuildVoice, + }; + }); + const manager = createManager(undefined, client); + + await manager.join({ guildId: "g1", channelId: "1001" }); + await manager.join({ guildId: "g2", channelId: "2001" }); + emitDecryptFailure(manager); + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(3)); + ( + manager as unknown as { handleReceiveError: (entry: unknown, err: unknown) => void } + ).handleReceiveError( + getSessionEntry(manager, "g2"), + new Error("Failed to decrypt: DecryptionFailed(UnencryptedWhenPassthroughDisabled)"), + ); + + await vi.waitFor(() => expect(joinVoiceChannelMock).toHaveBeenCalledTimes(4)); + expect(manager.status()).toHaveLength(2); + }); + + it("clears poisoned-DAVE reconnect budgets when the manager is destroyed", async () => { + const manager = createManager(); + const attempts = (manager as unknown as { daveRecoveryAttempts: Map }) + .daveRecoveryAttempts; + attempts.set("g1", Date.now()); + + await manager.destroy(); + + expect(attempts.size).toBe(0); + }); + it("re-arms passthrough but still rejoin-recovers after repeated decrypt failures", async () => { const connection = createConnectionMock(); joinVoiceChannelMock @@ -4601,6 +5230,9 @@ describe("DiscordVoiceManager", () => { emitDecryptFailure(manager); emitDecryptFailure(manager); const entry = getSessionEntry(manager); + const attempts = (manager as unknown as { daveRecoveryAttempts: Map }) + .daveRecoveryAttempts; + attempts.set("g1", Date.now()); expect(entry.receiveRecovery.decryptFailureCount).toBe(2); const stream = { on: vi.fn(), @@ -4618,6 +5250,7 @@ describe("DiscordVoiceManager", () => { expect(decodeOpusStreamChunksMock).toHaveBeenCalledTimes(1); expect(entry.receiveRecovery.decryptFailureCount).toBe(0); expect(entry.receiveRecovery.lastDecryptFailureAt).toBe(0); + expect(attempts.has("g1")).toBe(false); expect(joinVoiceChannelMock).toHaveBeenCalledTimes(1); }); diff --git a/extensions/discord/src/voice/manager.ts b/extensions/discord/src/voice/manager.ts index 7a7be6417163..a7fd5cbce372 100644 --- a/extensions/discord/src/voice/manager.ts +++ b/extensions/discord/src/voice/manager.ts @@ -47,9 +47,11 @@ import { createVoiceReceiveRecoveryState, DAVE_RECEIVE_PASSTHROUGH_INITIAL_EXPIRY_SECONDS, DAVE_RECEIVE_PASSTHROUGH_REARM_EXPIRY_SECONDS, + DECRYPT_FAILURE_WINDOW_MS, enableDaveReceivePassthrough as tryEnableDaveReceivePassthrough, finishVoiceDecryptRecovery, noteVoiceDecryptFailure, + recoverDaveZeroTransition as tryRecoverDaveZeroTransition, resetVoiceReceiveRecoveryState, } from "./receive-recovery.js"; import { loadDiscordVoiceSdk } from "./sdk-runtime.js"; @@ -258,6 +260,7 @@ function resolveDiscordVoiceAgentRoute(params: { export class DiscordVoiceManager { private sessions = new Map(); private readonly joinTasks = new Map>(); + private readonly daveRecoveryAttempts = new Map(); private botUserId?: string; private readonly voiceEnabled: boolean; private autoJoinTask: Promise | null = null; @@ -950,6 +953,9 @@ export class DiscordVoiceManager { } entry.stop(); this.sessions.delete(guildId); + if (!entry.receiveRecovery.decryptRecoveryInFlight) { + this.daveRecoveryAttempts.delete(guildId); + } if (!options?.preserveFollowState) { this.followedVoiceGuilds.delete(guildId); this.deleteFollowedUserChannelsForGuild(guildId); @@ -1047,6 +1053,7 @@ export class DiscordVoiceManager { } const { guildId, channelId, userId } = params; const followKey = this.formatFollowedUserKey({ guildId, userId }); + const previousFollowedChannelId = this.followedUserChannels.get(followKey)?.channelId; const existing = this.sessions.get(guildId); const wasFollowedVoiceSession = this.followedUserChannels.has(followKey) || this.followedVoiceGuilds.has(guildId); @@ -1082,6 +1089,16 @@ export class DiscordVoiceManager { this.followedVoiceGuilds.add(guildId); return; } + const recoveryAttemptAt = this.daveRecoveryAttempts.get(guildId); + if (!existing && previousFollowedChannelId === channelId && recoveryAttemptAt !== undefined) { + if (Date.now() - recoveryAttemptAt < DECRYPT_FAILURE_WINDOW_MS) { + logger.warn( + `discord voice: automatic follow suppressed during DAVE recovery cooldown guild=${guildId} channel=${channelId}; retry /vc join after the voice gateway recovers`, + ); + return; + } + this.daveRecoveryAttempts.delete(guildId); + } logger.info( `discord voice: following user guild=${guildId} user=${userId} channel=${channelId}`, ); @@ -1111,6 +1128,7 @@ export class DiscordVoiceManager { entry.stop(); } this.sessions.clear(); + this.daveRecoveryAttempts.clear(); this.followedUserChannels.clear(); this.followedVoiceGuilds.clear(); } @@ -1801,6 +1819,17 @@ export class DiscordVoiceManager { } logger.warn(`discord voice: receive error: ${analysis.message}`); if (analysis.shouldAttemptPassthrough) { + if (this.sessions.get(entry.guildId) === entry && !entry.isStopped()) { + const recovery = tryRecoverDaveZeroTransition({ + target: entry, + sdk: loadDiscordVoiceSdk(), + onWarn: (message) => logger.warn(message), + }); + if (recovery === "failed") { + this.startDecryptRecovery(entry, true); + return; + } + } this.enableDaveReceivePassthrough( entry, "receive decrypt error", @@ -1819,7 +1848,45 @@ export class DiscordVoiceManager { if (!decryptFailure.shouldRecover) { return; } - void this.recoverFromDecryptFailures(entry) + this.startDecryptRecovery(entry); + } + + private startDecryptRecovery(entry: VoiceSessionEntry, force = false): void { + let recovery: Promise; + if (force) { + if ( + this.sessions.get(entry.guildId) !== entry || + entry.isStopped() || + entry.receiveRecovery.decryptRecoveryInFlight + ) { + return; + } + const now = Date.now(); + for (const [guildId, attemptedAt] of this.daveRecoveryAttempts) { + if (now - attemptedAt >= DECRYPT_FAILURE_WINDOW_MS) { + this.daveRecoveryAttempts.delete(guildId); + } + } + resetVoiceReceiveRecoveryState(entry.receiveRecovery); + entry.receiveRecovery.decryptRecoveryInFlight = true; + if (this.daveRecoveryAttempts.has(entry.guildId)) { + const windowSeconds = DECRYPT_FAILURE_WINDOW_MS / 1_000; + logger.warn( + `discord voice: DAVE recovery failed again within ${windowSeconds} seconds; disconnecting guild=${entry.guildId} channel=${entry.channelId} to avoid a reconnect loop; retry /vc join after the voice gateway recovers`, + ); + recovery = this.leave( + { guildId: entry.guildId }, + { preserveFollowState: this.isFollowOwnedGuild(entry.guildId) }, + ); + } else { + // A partially invalidated DAVE session suppresses all later decrypt failures. + this.daveRecoveryAttempts.set(entry.guildId, now); + recovery = this.recoverFromDecryptFailures(entry); + } + } else { + recovery = this.recoverFromDecryptFailures(entry); + } + void recovery .catch((recoverErr: unknown) => logger.warn(`discord voice: decrypt recovery failed: ${formatErrorMessage(recoverErr)}`), ) @@ -1872,6 +1939,9 @@ export class DiscordVoiceManager { private resetDecryptFailureState(entry: VoiceSessionEntry) { resetVoiceReceiveRecoveryState(entry.receiveRecovery); + if (this.sessions.get(entry.guildId) === entry && !entry.isStopped()) { + this.daveRecoveryAttempts.delete(entry.guildId); + } } private async recoverFromDecryptFailures(entry: VoiceSessionEntry) { diff --git a/extensions/discord/src/voice/receive-recovery.test.ts b/extensions/discord/src/voice/receive-recovery.test.ts index d3a94cb0c219..dc049d16418b 100644 --- a/extensions/discord/src/voice/receive-recovery.test.ts +++ b/extensions/discord/src/voice/receive-recovery.test.ts @@ -1,4 +1,6 @@ // Discord tests cover receive recovery plugin behavior. +import { DAVESession, NetworkingStatusCode, VoiceConnectionStatus } from "@discordjs/voice"; +import { VoiceOpcodes, type VoiceSendPayload } from "discord-api-types/voice/v8"; import { OpusError } from "libopus-wasm"; import { describe, expect, it, vi } from "vitest"; import { @@ -6,6 +8,7 @@ import { createVoiceReceiveRecoveryState, enableDaveReceivePassthrough, noteVoiceDecryptFailure, + recoverDaveZeroTransition, } from "./receive-recovery.js"; const OPUS_INVALID_PACKET_CODE = -4; @@ -144,4 +147,290 @@ describe("voice receive recovery", () => { expect(onVerbose).toHaveBeenCalled(); expect(onWarn).not.toHaveBeenCalled(); }); + + it("recovers transition zero through the real DAVE invalidation and key-package events", () => { + const dave = new DAVESession(1, "bot", "c1", { decryptionFailureTolerance: 0 }); + const keyPackage = Buffer.from("new-key-package"); + const nativeSession = { + decrypt: vi.fn(() => { + throw new Error("UnencryptedWhenPassthroughDisabled"); + }), + getSerializedKeyPackage: vi.fn(() => keyPackage), + ready: true, + reinit: vi.fn(), + setPassthroughMode: vi.fn(), + }; + dave.session = nativeSession as unknown as NonNullable; + dave.lastTransitionId = 0; + const gateway = { + sendBinaryMessage: vi.fn(), + sendPacket: vi.fn(), + }; + dave.on("invalidateTransition", (transitionId) => { + gateway.sendPacket({ + op: VoiceOpcodes.DaveMlsInvalidCommitWelcome, + d: { transition_id: transitionId }, + }); + }); + dave.on("keyPackage", (nextKeyPackage) => { + gateway.sendBinaryMessage(VoiceOpcodes.DaveMlsKeyPackage, nextKeyPackage); + }); + const onWarn = vi.fn(); + + expect(() => dave.decrypt(Buffer.from("encrypted-audio"), "speaker")).toThrow( + "UnencryptedWhenPassthroughDisabled", + ); + + expect( + recoverDaveZeroTransition({ + target: { + guildId: "g1", + channelId: "c1", + connection: { + state: { + status: VoiceConnectionStatus.Ready, + networking: { + state: { + code: NetworkingStatusCode.Ready, + dave, + }, + }, + }, + }, + }, + sdk: { VoiceConnectionStatus, NetworkingStatusCode }, + onWarn, + }), + ).toBe("recovered"); + + expect(nativeSession.reinit).toHaveBeenCalledWith(1, "bot", "c1"); + expect(gateway.sendPacket).toHaveBeenCalledWith({ + op: VoiceOpcodes.DaveMlsInvalidCommitWelcome, + d: { transition_id: 0 }, + }); + expect(gateway.sendBinaryMessage).toHaveBeenCalledWith( + VoiceOpcodes.DaveMlsKeyPackage, + keyPackage, + ); + expect(gateway.sendPacket.mock.invocationCallOrder[0]).toBeLessThan( + gateway.sendBinaryMessage.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY, + ); + expect(dave.reinitializing).toBe(true); + expect(dave.decrypt(Buffer.from("encrypted-audio"), "speaker")).toBeNull(); + expect(onWarn).not.toHaveBeenCalled(); + }); + + it.each([ + { label: "gateway invalidation", failure: "invalidation" }, + { label: "native reinitialization", failure: "native" }, + { label: "gateway key-package delivery", failure: "key-package" }, + ])("reports the poisoned session when $label fails", ({ failure }) => { + const dave = new DAVESession(1, "bot", "c1", { decryptionFailureTolerance: 0 }); + const nativeSession = { + decrypt: vi.fn(() => { + throw new Error("UnencryptedWhenPassthroughDisabled"); + }), + getSerializedKeyPackage: vi.fn(() => Buffer.from("new-key-package")), + ready: true, + reinit: vi.fn(() => { + if (failure === "native") { + throw new Error("native DAVE reinitialization failed"); + } + }), + setPassthroughMode: vi.fn(), + }; + dave.session = nativeSession as unknown as NonNullable; + dave.lastTransitionId = 0; + const gateway = { + sendPacket: vi.fn((_packet: VoiceSendPayload) => { + if (failure === "invalidation") { + throw new Error("voice gateway invalidation failed"); + } + }), + sendBinaryMessage: vi.fn((_opcode: VoiceOpcodes, _keyPackage: Buffer) => { + if (failure === "key-package") { + throw new Error("voice gateway key-package delivery failed"); + } + }), + }; + dave.on("invalidateTransition", (transitionId) => { + gateway.sendPacket({ + op: VoiceOpcodes.DaveMlsInvalidCommitWelcome, + d: { transition_id: transitionId }, + }); + }); + dave.on("keyPackage", (keyPackage) => { + gateway.sendBinaryMessage(VoiceOpcodes.DaveMlsKeyPackage, keyPackage); + }); + const onWarn = vi.fn(); + + expect( + recoverDaveZeroTransition({ + target: { + guildId: "g1", + channelId: "c1", + connection: { + state: { + status: VoiceConnectionStatus.Ready, + networking: { + state: { code: NetworkingStatusCode.Ready, dave }, + }, + }, + }, + }, + sdk: { VoiceConnectionStatus, NetworkingStatusCode }, + onWarn, + }), + ).toBe("failed"); + + expect(dave.reinitializing).toBe(true); + expect(dave.decrypt(Buffer.from("encrypted-audio"), "speaker")).toBeNull(); + expect(gateway.sendPacket).toHaveBeenCalledWith({ + op: VoiceOpcodes.DaveMlsInvalidCommitWelcome, + d: { transition_id: 0 }, + }); + expect(gateway.sendBinaryMessage).toHaveBeenCalledTimes(failure === "key-package" ? 1 : 0); + expect(onWarn).toHaveBeenCalledWith( + expect.stringContaining("failed to recover DAVE transition 0"), + ); + }); + + it("keeps bounded recovery when a real DAVE debug listener fails before invalidation", () => { + const dave = new DAVESession(1, "bot", "c1", { decryptionFailureTolerance: 0 }); + dave.lastTransitionId = 0; + dave.on("debug", () => { + throw new Error("debug subscriber unavailable"); + }); + const onWarn = vi.fn(); + + expect( + recoverDaveZeroTransition({ + target: { + guildId: "g1", + channelId: "c1", + connection: { + state: { + status: VoiceConnectionStatus.Ready, + networking: { + state: { code: NetworkingStatusCode.Ready, dave }, + }, + }, + }, + }, + sdk: { VoiceConnectionStatus, NetworkingStatusCode }, + onWarn, + }), + ).toBe("not-attempted"); + + expect(dave.reinitializing).toBe(false); + expect(onWarn).toHaveBeenCalledWith( + expect.stringContaining("failed to recover DAVE transition 0"), + ); + }); + + it.each([ + { + label: "non-zero transitions", + lastTransitionId: 1, + reinitializing: false, + connectionStatus: VoiceConnectionStatus.Ready, + networkingStatus: NetworkingStatusCode.Ready, + }, + { + label: "missing transitions", + lastTransitionId: undefined, + reinitializing: false, + connectionStatus: VoiceConnectionStatus.Ready, + networkingStatus: NetworkingStatusCode.Ready, + }, + { + label: "reinitializing sessions", + lastTransitionId: 0, + reinitializing: true, + connectionStatus: VoiceConnectionStatus.Ready, + networkingStatus: NetworkingStatusCode.Ready, + }, + { + label: "resuming networking", + lastTransitionId: 0, + reinitializing: false, + connectionStatus: VoiceConnectionStatus.Ready, + networkingStatus: NetworkingStatusCode.Resuming, + }, + { + label: "disconnected voice connections", + lastTransitionId: 0, + reinitializing: false, + connectionStatus: VoiceConnectionStatus.Disconnected, + networkingStatus: NetworkingStatusCode.Ready, + }, + ])("does not invalidate $label", (scenario) => { + const { lastTransitionId, reinitializing, connectionStatus, networkingStatus } = scenario; + const recoverFromInvalidTransition = vi.fn(); + + expect( + recoverDaveZeroTransition({ + target: { + guildId: "g1", + channelId: "c1", + connection: { + state: { + status: connectionStatus, + networking: { + state: { + code: networkingStatus, + dave: { + lastTransitionId, + reinitializing, + recoverFromInvalidTransition, + }, + }, + }, + }, + }, + }, + sdk: { VoiceConnectionStatus, NetworkingStatusCode }, + onWarn: vi.fn(), + }), + ).toBe("not-attempted"); + + expect(recoverFromInvalidTransition).not.toHaveBeenCalled(); + }); + + it("keeps bounded recovery available when transition invalidation throws", () => { + const onWarn = vi.fn(); + const recoverFromInvalidTransition = vi.fn(() => { + throw new Error("voice gateway unavailable"); + }); + + expect( + recoverDaveZeroTransition({ + target: { + guildId: "g1", + channelId: "c1", + connection: { + state: { + status: VoiceConnectionStatus.Ready, + networking: { + state: { + code: NetworkingStatusCode.Ready, + dave: { + lastTransitionId: 0, + reinitializing: false, + recoverFromInvalidTransition, + }, + }, + }, + }, + }, + }, + sdk: { VoiceConnectionStatus, NetworkingStatusCode }, + onWarn, + }), + ).toBe("not-attempted"); + + expect(onWarn).toHaveBeenCalledWith( + expect.stringContaining("failed to recover DAVE transition 0"), + ); + }); }); diff --git a/extensions/discord/src/voice/receive-recovery.ts b/extensions/discord/src/voice/receive-recovery.ts index f477da66880c..4b1f2a83279b 100644 --- a/extensions/discord/src/voice/receive-recovery.ts +++ b/extensions/discord/src/voice/receive-recovery.ts @@ -2,7 +2,7 @@ import { OpusError } from "libopus-wasm"; import { formatErrorMessage } from "openclaw/plugin-sdk/ssrf-runtime"; -const DECRYPT_FAILURE_WINDOW_MS = 30_000; +export const DECRYPT_FAILURE_WINDOW_MS = 30_000; const DECRYPT_FAILURE_RECONNECT_THRESHOLD = 3; const DECRYPT_FAILURE_MARKER = "DecryptionFailed("; const DAVE_PASSTHROUGH_DISABLED_MARKER = "UnencryptedWhenPassthroughDisabled"; @@ -36,6 +36,9 @@ type DavePassthroughTarget = { state?: { code?: unknown; dave?: { + lastTransitionId?: number; + reinitializing?: boolean; + recoverFromInvalidTransition?: (transitionId: number) => void; session?: { setPassthroughMode: (passthrough: boolean, expirySeconds: number) => void; }; @@ -155,6 +158,41 @@ export function finishVoiceDecryptRecovery(state: VoiceReceiveRecoveryState): vo state.decryptRecoveryInFlight = false; } +function isDaveReinitializing(session: { reinitializing?: boolean }): boolean { + return session.reinitializing === true; +} + +export function recoverDaveZeroTransition(params: { + target: DavePassthroughTarget; + sdk: DavePassthroughSdk; + onWarn: (message: string) => void; +}): "not-attempted" | "recovered" | "failed" { + const { target, sdk, onWarn } = params; + const networkingState = target.connection.state.networking?.state; + const daveSession = networkingState?.dave; + if ( + target.connection.state.status !== sdk.VoiceConnectionStatus.Ready || + networkingState?.code !== sdk.NetworkingStatusCode.Ready || + daveSession?.lastTransitionId !== 0 || + daveSession.reinitializing !== false || + typeof daveSession.recoverFromInvalidTransition !== "function" + ) { + return "not-attempted"; + } + + try { + // The upstream DAVE recovery guard treats transition zero as falsy. + daveSession.recoverFromInvalidTransition(0); + return "recovered"; + } catch (err) { + onWarn( + `discord voice: failed to recover DAVE transition 0 guild=${target.guildId} channel=${target.channelId}: ${formatErrorMessage(err)}`, + ); + // Upstream marks the session reinitializing before gateway/native work can fail. + return isDaveReinitializing(daveSession) ? "failed" : "not-attempted"; + } +} + export function enableDaveReceivePassthrough(params: { target: DavePassthroughTarget; sdk: DavePassthroughSdk;