From f5c986bb59c25103b86defa635316d7796a141d8 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 31 Jul 2026 13:13:16 +0800 Subject: [PATCH 1/6] fix(gateway): close talk relays on disconnect --- src/gateway/server/ws-connection.test.ts | 29 +++++++ src/gateway/server/ws-connection.ts | 2 + src/gateway/talk-realtime-relay-operations.ts | 16 ++++ src/gateway/talk-realtime-relay.test.ts | 82 +++++++++++++++++++ src/gateway/talk-realtime-relay.ts | 1 + 5 files changed, 130 insertions(+) diff --git a/src/gateway/server/ws-connection.test.ts b/src/gateway/server/ws-connection.test.ts index 92cc4a021cd7..60f0a425951b 100644 --- a/src/gateway/server/ws-connection.test.ts +++ b/src/gateway/server/ws-connection.test.ts @@ -15,12 +15,14 @@ const { attachGatewayWsMessageHandlerMock, attachWorkerWsMessageHandlerMock, broadcastPresenceSnapshotMock, + closeTalkRealtimeRelaySessionsForConnectionMock, touchPresenceMock, upsertPresenceMock, } = vi.hoisted(() => ({ attachGatewayWsMessageHandlerMock: vi.fn(), attachWorkerWsMessageHandlerMock: vi.fn((_params: unknown) => vi.fn()), broadcastPresenceSnapshotMock: vi.fn(), + closeTalkRealtimeRelaySessionsForConnectionMock: vi.fn(), touchPresenceMock: vi.fn(), upsertPresenceMock: vi.fn(), })); @@ -38,6 +40,9 @@ vi.mock("../../infra/system-presence.js", () => ({ vi.mock("./presence-events.js", () => ({ broadcastPresenceSnapshot: broadcastPresenceSnapshotMock, })); +vi.mock("../talk-realtime-relay.js", () => ({ + closeTalkRealtimeRelaySessionsForConnection: closeTalkRealtimeRelaySessionsForConnectionMock, +})); import { attachGatewayWsConnectionHandler } from "./ws-connection.js"; import { resolveSharedGatewaySessionGeneration } from "./ws-shared-generation.js"; @@ -91,6 +96,7 @@ describe("attachGatewayWsConnectionHandler", () => { attachGatewayWsMessageHandlerMock.mockReset(); attachWorkerWsMessageHandlerMock.mockClear(); broadcastPresenceSnapshotMock.mockReset(); + closeTalkRealtimeRelaySessionsForConnectionMock.mockReset(); touchPresenceMock.mockReset(); upsertPresenceMock.mockReset(); }); @@ -262,6 +268,29 @@ describe("attachGatewayWsConnectionHandler", () => { expect(socket.ping).toHaveBeenCalledOnce(); }); + it("releases realtime Talk relays when a gateway connection closes", async () => { + const { passed, socket } = await connectTestWs(); + const handlerParams = passed as { + connId: string; + setClient: (client: unknown) => boolean; + }; + expect( + handlerParams.setClient({ + socket, + connect: { client: { id: "openclaw-control-ui", mode: "webchat" } }, + connId: handlerParams.connId, + usesSharedGatewayAuth: false, + }), + ).toBe(true); + + socket.emit("close", 1000, Buffer.from("done")); + + expect(closeTalkRealtimeRelaySessionsForConnectionMock).toHaveBeenCalledOnce(); + expect(closeTalkRealtimeRelaySessionsForConnectionMock).toHaveBeenCalledWith( + handlerParams.connId, + ); + }); + it("continues protocol pings after pong and stops when the connection closes", async () => { vi.useFakeTimers(); const socket = Object.assign(createGatewayWsTestSocket({ ping: true }), { diff --git a/src/gateway/server/ws-connection.ts b/src/gateway/server/ws-connection.ts index 6357d1b7a487..f0db7cb79dd6 100644 --- a/src/gateway/server/ws-connection.ts +++ b/src/gateway/server/ws-connection.ts @@ -28,6 +28,7 @@ import { } from "../server-constants.js"; import type { GatewayRequestContext, GatewayRequestHandlers } from "../server-methods/types.js"; import { formatError } from "../server-utils.js"; +import { closeTalkRealtimeRelaySessionsForConnection } from "../talk-realtime-relay.js"; import { formatForLog, logWs } from "../ws-log.js"; import { getHealthVersion, incrementPresenceVersion } from "./health-state.js"; import type { PreauthConnectionBudget } from "./preauth-connection-budget.js"; @@ -554,6 +555,7 @@ export function attachGatewayWsConnectionHandler(params: AttachGatewayWsConnecti } if (connectionKind === "gateway") { const context = buildRequestContext(); + closeTalkRealtimeRelaySessionsForConnection(connId); context.unsubscribeAllSessionEvents(connId); // Detach (or, with a zero grace period, kill) any PTY shells this // connection owned; detached sessions stay reattachable via diff --git a/src/gateway/talk-realtime-relay-operations.ts b/src/gateway/talk-realtime-relay-operations.ts index 97b0d977df1a..2b609a27f095 100644 --- a/src/gateway/talk-realtime-relay-operations.ts +++ b/src/gateway/talk-realtime-relay-operations.ts @@ -6,6 +6,7 @@ import { import { registerClientVoiceConsultRun } from "../talk/client-voice-session.js"; import type { RealtimeVoiceToolResultOptions } from "../talk/provider-types.js"; import { abortChatRunById } from "./chat-abort.js"; +import { formatError } from "./server-utils.js"; import { cancelForcedConsults, submitForcedTalkRealtimeRelayToolResult, @@ -107,6 +108,21 @@ export function closeRelaySession(session: RelaySession, reason: "completed" | " }); } +/** Releases every realtime relay session owned by a disconnected gateway connection. */ +export function closeTalkRealtimeRelaySessionsForConnection(connId: string): void { + for (const session of relaySessions.values()) { + if (session.connId === connId) { + try { + closeRelaySession(session, "completed"); + } catch (error) { + session.context.logGateway.warn( + `failed to close realtime relay session after connection disconnect: ${formatError(error)}`, + ); + } + } + } +} + function pruneExpiredRelaySessions(nowMs = Date.now()): void { closeExpiredTalkRelaySessions({ sessions: relaySessions.values(), diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index e213aec5205a..f8a28e77a5ff 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -25,6 +25,7 @@ import { createChatRunState } from "./server-chat-state.js"; import { acknowledgeTalkRealtimeRelayMark, cancelTalkRealtimeRelayTurn, + closeTalkRealtimeRelaySessionsForConnection, createTalkRealtimeRelaySession as createTalkRealtimeRelaySessionRaw, ensureTalkRealtimeRelayVoiceSession, flushTalkRealtimeRelayVoiceWrites, @@ -92,6 +93,87 @@ describe("talk realtime gateway relay", () => { }; } + it("closes only realtime relays owned by the disconnected connection", () => { + const bridgeCloses: Array> = []; + const bridgeAudioSends: Array> = []; + const provider = createIdleRelayProvider(); + provider.createBridge = () => { + const close = vi.fn(); + const sendAudio = vi.fn(); + bridgeCloses.push(close); + bridgeAudioSends.push(sendAudio); + return { + connect: vi.fn(async () => undefined), + sendAudio, + setMediaTimestamp: vi.fn(), + handleBargeIn: vi.fn(), + submitToolResult: vi.fn(), + acknowledgeMark: vi.fn(), + close, + isConnected: vi.fn(() => true), + }; + }; + const logGateway = { warn: vi.fn() }; + const context = { + broadcastToConnIds: vi.fn(), + chatAbortControllers: new Map(), + getRuntimeConfig: () => ({}), + logGateway, + } as never; + const createSession = (connId: string) => + createTalkRealtimeRelaySession({ + context, + connId, + provider, + providerConfig: {}, + instructions: "brief", + tools: [], + }); + const firstOwned = createSession("conn-owner"); + const secondOwned = createSession("conn-owner"); + const unrelated = createSession("conn-other"); + bridgeCloses[0]?.mockImplementationOnce(() => { + throw new Error("provider close failed"); + }); + + expect(() => closeTalkRealtimeRelaySessionsForConnection("conn-owner")).not.toThrow(); + closeTalkRealtimeRelaySessionsForConnection("conn-owner"); + + expect(bridgeCloses[0]).toHaveBeenCalledOnce(); + expect(bridgeCloses[1]).toHaveBeenCalledOnce(); + expect(bridgeCloses[2]).not.toHaveBeenCalled(); + expect(logGateway.warn).toHaveBeenCalledWith( + "failed to close realtime relay session after connection disconnect: provider close failed", + ); + expect(() => + sendTalkRealtimeRelayAudio({ + relaySessionId: firstOwned.relaySessionId, + connId: "conn-owner", + audioBase64: "AQI=", + }), + ).toThrow("Unknown realtime relay session"); + expect(() => + sendTalkRealtimeRelayAudio({ + relaySessionId: secondOwned.relaySessionId, + connId: "conn-owner", + audioBase64: "AQI=", + }), + ).toThrow("Unknown realtime relay session"); + + sendTalkRealtimeRelayAudio({ + relaySessionId: unrelated.relaySessionId, + connId: "conn-other", + audioBase64: "AQI=", + }); + expect(bridgeAudioSends[2]).toHaveBeenCalledOnce(); + stopTalkRealtimeRelaySession({ + relaySessionId: unrelated.relaySessionId, + connId: "conn-other", + }); + closeTalkRealtimeRelaySessionsForConnection("conn-other"); + expect(bridgeCloses[2]).toHaveBeenCalledOnce(); + }); + it("injects the host agent runner only into gateway-relay bridge creation", () => { let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; const provider = createIdleRelayProvider(); diff --git a/src/gateway/talk-realtime-relay.ts b/src/gateway/talk-realtime-relay.ts index 225a59f2b426..2ae9964ec36c 100644 --- a/src/gateway/talk-realtime-relay.ts +++ b/src/gateway/talk-realtime-relay.ts @@ -4,6 +4,7 @@ export { createTalkRealtimeRelaySession } from "./talk-realtime-relay-session-cr export { acknowledgeTalkRealtimeRelayMark, cancelTalkRealtimeRelayTurn, + closeTalkRealtimeRelaySessionsForConnection, ensureTalkRealtimeRelayVoiceSession, flushTalkRealtimeRelayVoiceWrites, registerTalkRealtimeRelayAgentRun, From c278eb0ba78596e39f0f1851e26dc7ba9d5c131c Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 31 Jul 2026 15:48:10 +0800 Subject: [PATCH 2/6] fix(gateway): finalize talk relay teardown after close errors --- src/gateway/talk-realtime-relay-operations.ts | 29 ++-- src/gateway/talk-realtime-relay.test.ts | 152 +++++++++++------- 2 files changed, 112 insertions(+), 69 deletions(-) diff --git a/src/gateway/talk-realtime-relay-operations.ts b/src/gateway/talk-realtime-relay-operations.ts index 2b609a27f095..e54b99717c06 100644 --- a/src/gateway/talk-realtime-relay-operations.ts +++ b/src/gateway/talk-realtime-relay-operations.ts @@ -94,18 +94,23 @@ export function closeRelaySession(session: RelaySession, reason: "completed" | " forgetUnifiedTalkSession(session.id); clearTimeout(session.cleanupTimer); abortRelayAgentRuns(session, reason === "error" ? "relay-error" : "relay-closed"); - session.bridge.close(); - closeRelayVoiceSession(session); - broadcastToOwner(session.context, session.connId, { - relaySessionId: session.id, - type: "close", - reason, - talkEvent: session.harness.talk.emit({ - type: "session.closed", - payload: { reason }, - final: true, - }), - }); + try { + session.bridge.close(); + } finally { + // Provider teardown may throw, but the relay must still reach its durable + // voice and owner-visible terminal state before that error is surfaced. + closeRelayVoiceSession(session); + broadcastToOwner(session.context, session.connId, { + relaySessionId: session.id, + type: "close", + reason, + talkEvent: session.harness.talk.emit({ + type: "session.closed", + payload: { reason }, + final: true, + }), + }); + } } /** Releases every realtime relay session owned by a disconnected gateway connection. */ diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index f8a28e77a5ff..4c87e0f23860 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -5,6 +5,7 @@ import path from "node:path"; * Tests talk realtime relay event forwarding and connection cleanup. */ import { afterEach, describe, expect, it, vi } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; import { setActiveEmbeddedRun } from "../agents/embedded-agent-runner/runs.js"; import { testing as embeddedRunTesting } from "../agents/embedded-agent-runner/runs.test-support.js"; import { @@ -37,6 +38,7 @@ import { } from "./talk-realtime-relay.js"; const activeRelaySessions = new Map(); +const tempDirs = useAutoCleanupTempDirTracker(afterEach); function createTalkRealtimeRelaySession( params: Parameters[0], @@ -93,7 +95,10 @@ describe("talk realtime gateway relay", () => { }; } - it("closes only realtime relays owned by the disconnected connection", () => { + it("closes only realtime relays owned by the disconnected connection", async () => { + const envSnapshot = captureEnv(["OPENCLAW_STATE_DIR"]); + const tempDir = await fs.realpath(tempDirs.make("openclaw-relay-disconnect-")); + setTestEnvValue("OPENCLAW_STATE_DIR", tempDir); const bridgeCloses: Array> = []; const bridgeAudioSends: Array> = []; const provider = createIdleRelayProvider(); @@ -113,65 +118,98 @@ describe("talk realtime gateway relay", () => { isConnected: vi.fn(() => true), }; }; - const logGateway = { warn: vi.fn() }; - const context = { - broadcastToConnIds: vi.fn(), - chatAbortControllers: new Map(), - getRuntimeConfig: () => ({}), - logGateway, - } as never; - const createSession = (connId: string) => - createTalkRealtimeRelaySession({ - context, - connId, - provider, - providerConfig: {}, - instructions: "brief", - tools: [], - }); - const firstOwned = createSession("conn-owner"); - const secondOwned = createSession("conn-owner"); - const unrelated = createSession("conn-other"); - bridgeCloses[0]?.mockImplementationOnce(() => { - throw new Error("provider close failed"); - }); - - expect(() => closeTalkRealtimeRelaySessionsForConnection("conn-owner")).not.toThrow(); - closeTalkRealtimeRelaySessionsForConnection("conn-owner"); - - expect(bridgeCloses[0]).toHaveBeenCalledOnce(); - expect(bridgeCloses[1]).toHaveBeenCalledOnce(); - expect(bridgeCloses[2]).not.toHaveBeenCalled(); - expect(logGateway.warn).toHaveBeenCalledWith( - "failed to close realtime relay session after connection disconnect: provider close failed", - ); - expect(() => - sendTalkRealtimeRelayAudio({ + try { + const logGateway = { warn: vi.fn() }; + const broadcastToConnIds = vi.fn(); + const context = { + broadcastToConnIds, + chatAbortControllers: new Map(), + getRuntimeConfig: () => ({}), + logGateway, + } as never; + const createSession = (connId: string) => + createTalkRealtimeRelaySession({ + context, + connId, + provider, + providerConfig: {}, + instructions: "brief", + tools: [], + }); + const firstOwned = createSession("conn-owner"); + const secondOwned = createSession("conn-owner"); + const unrelated = createSession("conn-other"); + ensureTalkRealtimeRelayVoiceSession({ relaySessionId: firstOwned.relaySessionId, connId: "conn-owner", - audioBase64: "AQI=", - }), - ).toThrow("Unknown realtime relay session"); - expect(() => - sendTalkRealtimeRelayAudio({ - relaySessionId: secondOwned.relaySessionId, - connId: "conn-owner", - audioBase64: "AQI=", - }), - ).toThrow("Unknown realtime relay session"); + sessionKey: "agent:main:main", + }); + expect(clientVoiceSessionTesting.readRecord("main", firstOwned.relaySessionId)).toMatchObject( + { + status: "open", + }, + ); + bridgeCloses[0]?.mockImplementationOnce(() => { + throw new Error("provider close failed"); + }); - sendTalkRealtimeRelayAudio({ - relaySessionId: unrelated.relaySessionId, - connId: "conn-other", - audioBase64: "AQI=", - }); - expect(bridgeAudioSends[2]).toHaveBeenCalledOnce(); - stopTalkRealtimeRelaySession({ - relaySessionId: unrelated.relaySessionId, - connId: "conn-other", - }); - closeTalkRealtimeRelaySessionsForConnection("conn-other"); - expect(bridgeCloses[2]).toHaveBeenCalledOnce(); + expect(() => closeTalkRealtimeRelaySessionsForConnection("conn-owner")).not.toThrow(); + closeTalkRealtimeRelaySessionsForConnection("conn-owner"); + + expect(bridgeCloses[0]).toHaveBeenCalledOnce(); + expect(bridgeCloses[1]).toHaveBeenCalledOnce(); + expect(bridgeCloses[2]).not.toHaveBeenCalled(); + expect(logGateway.warn).toHaveBeenCalledWith( + "failed to close realtime relay session after connection disconnect: provider close failed", + ); + await vi.waitFor(() => + expect( + clientVoiceSessionTesting.readRecord("main", firstOwned.relaySessionId)?.status, + ).toBe("closed"), + ); + expect( + broadcastToConnIds.mock.calls.some( + ([event, payload]) => + event === "talk.event" && + payload.relaySessionId === firstOwned.relaySessionId && + payload.type === "close" && + payload.talkEvent?.type === "session.closed" && + payload.talkEvent.final === true, + ), + ).toBe(true); + expect(() => + sendTalkRealtimeRelayAudio({ + relaySessionId: firstOwned.relaySessionId, + connId: "conn-owner", + audioBase64: "AQI=", + }), + ).toThrow("Unknown realtime relay session"); + expect(() => + sendTalkRealtimeRelayAudio({ + relaySessionId: secondOwned.relaySessionId, + connId: "conn-owner", + audioBase64: "AQI=", + }), + ).toThrow("Unknown realtime relay session"); + + sendTalkRealtimeRelayAudio({ + relaySessionId: unrelated.relaySessionId, + connId: "conn-other", + audioBase64: "AQI=", + }); + expect(bridgeAudioSends[2]).toHaveBeenCalledOnce(); + stopTalkRealtimeRelaySession({ + relaySessionId: unrelated.relaySessionId, + connId: "conn-other", + }); + closeTalkRealtimeRelaySessionsForConnection("conn-other"); + expect(bridgeCloses[2]).toHaveBeenCalledOnce(); + } finally { + clientVoiceSessionTesting.reset(); + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + envSnapshot.restore(); + } }); it("injects the host agent runner only into gateway-relay bridge creation", () => { From 743b04d1f8b0eddbace8edafb09dd79ee96c8666 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 31 Jul 2026 20:29:45 +0800 Subject: [PATCH 3/6] fix(gateway): close transcription relays on disconnect --- src/gateway/server/ws-connection.test.ts | 13 ++- src/gateway/server/ws-connection.ts | 2 + src/gateway/talk-realtime-relay-operations.ts | 22 ++--- .../talk-realtime-relay-session-create.ts | 9 +- src/gateway/talk-realtime-relay.test.ts | 17 +++- src/gateway/talk-relay-session-lifecycle.ts | 21 +++++ src/gateway/talk-transcription-relay.test.ts | 84 ++++++++++++++++++- src/gateway/talk-transcription-relay.ts | 53 ++++++++---- 8 files changed, 189 insertions(+), 32 deletions(-) diff --git a/src/gateway/server/ws-connection.test.ts b/src/gateway/server/ws-connection.test.ts index 60f0a425951b..2085b27570cb 100644 --- a/src/gateway/server/ws-connection.test.ts +++ b/src/gateway/server/ws-connection.test.ts @@ -16,6 +16,7 @@ const { attachWorkerWsMessageHandlerMock, broadcastPresenceSnapshotMock, closeTalkRealtimeRelaySessionsForConnectionMock, + closeTalkTranscriptionRelaySessionsForConnectionMock, touchPresenceMock, upsertPresenceMock, } = vi.hoisted(() => ({ @@ -23,6 +24,7 @@ const { attachWorkerWsMessageHandlerMock: vi.fn((_params: unknown) => vi.fn()), broadcastPresenceSnapshotMock: vi.fn(), closeTalkRealtimeRelaySessionsForConnectionMock: vi.fn(), + closeTalkTranscriptionRelaySessionsForConnectionMock: vi.fn(), touchPresenceMock: vi.fn(), upsertPresenceMock: vi.fn(), })); @@ -43,6 +45,10 @@ vi.mock("./presence-events.js", () => ({ vi.mock("../talk-realtime-relay.js", () => ({ closeTalkRealtimeRelaySessionsForConnection: closeTalkRealtimeRelaySessionsForConnectionMock, })); +vi.mock("../talk-transcription-relay.js", () => ({ + closeTalkTranscriptionRelaySessionsForConnection: + closeTalkTranscriptionRelaySessionsForConnectionMock, +})); import { attachGatewayWsConnectionHandler } from "./ws-connection.js"; import { resolveSharedGatewaySessionGeneration } from "./ws-shared-generation.js"; @@ -97,6 +103,7 @@ describe("attachGatewayWsConnectionHandler", () => { attachWorkerWsMessageHandlerMock.mockClear(); broadcastPresenceSnapshotMock.mockReset(); closeTalkRealtimeRelaySessionsForConnectionMock.mockReset(); + closeTalkTranscriptionRelaySessionsForConnectionMock.mockReset(); touchPresenceMock.mockReset(); upsertPresenceMock.mockReset(); }); @@ -268,7 +275,7 @@ describe("attachGatewayWsConnectionHandler", () => { expect(socket.ping).toHaveBeenCalledOnce(); }); - it("releases realtime Talk relays when a gateway connection closes", async () => { + it("releases connection-owned Talk relays when a gateway connection closes", async () => { const { passed, socket } = await connectTestWs(); const handlerParams = passed as { connId: string; @@ -289,6 +296,10 @@ describe("attachGatewayWsConnectionHandler", () => { expect(closeTalkRealtimeRelaySessionsForConnectionMock).toHaveBeenCalledWith( handlerParams.connId, ); + expect(closeTalkTranscriptionRelaySessionsForConnectionMock).toHaveBeenCalledOnce(); + expect(closeTalkTranscriptionRelaySessionsForConnectionMock).toHaveBeenCalledWith( + handlerParams.connId, + ); }); it("continues protocol pings after pong and stops when the connection closes", async () => { diff --git a/src/gateway/server/ws-connection.ts b/src/gateway/server/ws-connection.ts index f0db7cb79dd6..201ac8fec0ee 100644 --- a/src/gateway/server/ws-connection.ts +++ b/src/gateway/server/ws-connection.ts @@ -29,6 +29,7 @@ import { import type { GatewayRequestContext, GatewayRequestHandlers } from "../server-methods/types.js"; import { formatError } from "../server-utils.js"; import { closeTalkRealtimeRelaySessionsForConnection } from "../talk-realtime-relay.js"; +import { closeTalkTranscriptionRelaySessionsForConnection } from "../talk-transcription-relay.js"; import { formatForLog, logWs } from "../ws-log.js"; import { getHealthVersion, incrementPresenceVersion } from "./health-state.js"; import type { PreauthConnectionBudget } from "./preauth-connection-budget.js"; @@ -556,6 +557,7 @@ export function attachGatewayWsConnectionHandler(params: AttachGatewayWsConnecti if (connectionKind === "gateway") { const context = buildRequestContext(); closeTalkRealtimeRelaySessionsForConnection(connId); + closeTalkTranscriptionRelaySessionsForConnection(connId); context.unsubscribeAllSessionEvents(connId); // Detach (or, with a zero grace period, kill) any PTY shells this // connection owned; detached sessions stay reattachable via diff --git a/src/gateway/talk-realtime-relay-operations.ts b/src/gateway/talk-realtime-relay-operations.ts index e54b99717c06..8f791f39608b 100644 --- a/src/gateway/talk-realtime-relay-operations.ts +++ b/src/gateway/talk-realtime-relay-operations.ts @@ -41,6 +41,7 @@ import { import { decodeTalkRelayAudioBase64 } from "./talk-relay-audio-base64.js"; import { closeExpiredTalkRelaySessions, + closeTalkRelaySessionsForConnection, requireActiveTalkRelaySession, } from "./talk-relay-session-lifecycle.js"; import { forgetUnifiedTalkSession } from "./talk-session-registry.js"; @@ -115,17 +116,16 @@ export function closeRelaySession(session: RelaySession, reason: "completed" | " /** Releases every realtime relay session owned by a disconnected gateway connection. */ export function closeTalkRealtimeRelaySessionsForConnection(connId: string): void { - for (const session of relaySessions.values()) { - if (session.connId === connId) { - try { - closeRelaySession(session, "completed"); - } catch (error) { - session.context.logGateway.warn( - `failed to close realtime relay session after connection disconnect: ${formatError(error)}`, - ); - } - } - } + closeTalkRelaySessionsForConnection({ + sessions: relaySessions.values(), + connId, + closeSession: (session) => closeRelaySession(session, "completed"), + onCloseError: (error, session) => { + session.context.logGateway.warn( + `failed to close realtime relay session after connection disconnect: ${formatError(error)}`, + ); + }, + }); } function pruneExpiredRelaySessions(nowMs = Date.now()): void { diff --git a/src/gateway/talk-realtime-relay-session-create.ts b/src/gateway/talk-realtime-relay-session-create.ts index b148805c3a44..5b2b02be6b29 100644 --- a/src/gateway/talk-realtime-relay-session-create.ts +++ b/src/gateway/talk-realtime-relay-session-create.ts @@ -465,6 +465,10 @@ export function createTalkRealtimeRelaySession( relay.cleanupTimer.unref?.(); relaySessions.set(relaySessionId, relay); bridge.connect().catch((error: unknown) => { + const active = relaySessions.get(relaySessionId); + if (active !== relay) { + return; + } const issue = realtimeRelayIssue({ message: formatErrorMessage(error), provider: params.provider.id, @@ -477,10 +481,7 @@ export function createTalkRealtimeRelaySession( payload: issue, final: true, }); - const active = relaySessions.get(relaySessionId); - if (active) { - closeRelaySession(active, "error"); - } + closeRelaySession(active, "error"); }); return { diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index 4c87e0f23860..6609090e9d5c 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -103,12 +103,17 @@ describe("talk realtime gateway relay", () => { const bridgeAudioSends: Array> = []; const provider = createIdleRelayProvider(); provider.createBridge = () => { + const bridgeIndex = bridgeCloses.length; const close = vi.fn(); const sendAudio = vi.fn(); bridgeCloses.push(close); bridgeAudioSends.push(sendAudio); return { - connect: vi.fn(async () => undefined), + connect: vi.fn(async () => { + if (bridgeIndex === 1) { + throw new Error("late connect failure"); + } + }), sendAudio, setMediaTimestamp: vi.fn(), handleBargeIn: vi.fn(), @@ -155,6 +160,8 @@ describe("talk realtime gateway relay", () => { expect(() => closeTalkRealtimeRelaySessionsForConnection("conn-owner")).not.toThrow(); closeTalkRealtimeRelaySessionsForConnection("conn-owner"); + await Promise.resolve(); + await Promise.resolve(); expect(bridgeCloses[0]).toHaveBeenCalledOnce(); expect(bridgeCloses[1]).toHaveBeenCalledOnce(); @@ -177,6 +184,14 @@ describe("talk realtime gateway relay", () => { payload.talkEvent.final === true, ), ).toBe(true); + expect( + broadcastToConnIds.mock.calls.some( + ([event, payload]) => + event === "talk.event" && + payload.relaySessionId === secondOwned.relaySessionId && + payload.type === "error", + ), + ).toBe(false); expect(() => sendTalkRealtimeRelayAudio({ relaySessionId: firstOwned.relaySessionId, diff --git a/src/gateway/talk-relay-session-lifecycle.ts b/src/gateway/talk-relay-session-lifecycle.ts index aebb9fd68614..4d37aff98976 100644 --- a/src/gateway/talk-relay-session-lifecycle.ts +++ b/src/gateway/talk-relay-session-lifecycle.ts @@ -39,6 +39,27 @@ export function closeExpiredTalkRelaySessions(params: { + sessions: Iterable; + connId: string; + closeSession: CloseTalkRelaySession; + onCloseError: (error: unknown, session: TSession) => void; +}): void { + for (const session of params.sessions) { + if (session.connId !== params.connId) { + continue; + } + try { + params.closeSession(session); + } catch (error) { + params.onCloseError(error, session); + } + } +} + /** Returns the active session only when it belongs to the current connection. */ export function requireActiveTalkRelaySession(params: { sessions: ReadonlyMap; diff --git a/src/gateway/talk-transcription-relay.test.ts b/src/gateway/talk-transcription-relay.test.ts index 9eb57d7ba278..5614d3fe7f91 100644 --- a/src/gateway/talk-transcription-relay.test.ts +++ b/src/gateway/talk-transcription-relay.test.ts @@ -6,6 +6,7 @@ import type { RealtimeTranscriptionProviderPlugin } from "../plugins/types.js"; import type { RealtimeTranscriptionSessionCreateRequest } from "../realtime-transcription/provider-types.js"; import { cancelTalkTranscriptionRelayTurn, + closeTalkTranscriptionRelaySessionsForConnection, createTalkTranscriptionRelaySession, sendTalkTranscriptionRelayAudio, stopTalkTranscriptionRelaySession, @@ -40,13 +41,15 @@ function createTranscriptionProvider( function createBroadcastContext() { const events: BroadcastEvent[] = []; + const logGateway = { warn: vi.fn() }; const context = { getRuntimeConfig: () => ({}), + logGateway, broadcastToConnIds: (event: string, payload: unknown, connIds: ReadonlySet) => { events.push({ event, payload, connIds: [...connIds] }); }, } as never; - return { context, events }; + return { context, events, logGateway }; } async function createStartedRelaySession( @@ -212,6 +215,85 @@ describe("talk transcription gateway relay", () => { }); }); + it("closes only transcription relays owned by the disconnected connection", async () => { + const firstOwned = createSttSessionMock(); + const secondOwned = createSttSessionMock(async () => { + throw new Error("late connect failure"); + }); + const unrelated = createSttSessionMock(); + const { context, events, logGateway } = createBroadcastContext(); + const createSession = (connId: string, sttSession: ReturnType) => + createTalkTranscriptionRelaySession({ + context, + connId, + provider: createTranscriptionProvider(sttSession), + providerConfig: {}, + }); + const firstSession = createSession("conn-owner", firstOwned); + const secondSession = createSession("conn-owner", secondOwned); + const unrelatedSession = createSession("conn-other", unrelated); + firstOwned.close.mockImplementationOnce(() => { + throw new Error("provider close failed"); + }); + + expect(() => closeTalkTranscriptionRelaySessionsForConnection("conn-owner")).not.toThrow(); + closeTalkTranscriptionRelaySessionsForConnection("conn-owner"); + await Promise.resolve(); + await Promise.resolve(); + + expect(firstOwned.close).toHaveBeenCalledOnce(); + expect(secondOwned.close).toHaveBeenCalledOnce(); + expect(unrelated.close).not.toHaveBeenCalled(); + expect(logGateway.warn).toHaveBeenCalledWith( + "failed to close transcription relay session after connection disconnect: provider close failed", + ); + for (const transcriptionSessionId of [ + firstSession.transcriptionSessionId, + secondSession.transcriptionSessionId, + ]) { + expect( + events.some( + (event) => + isRecord(event.payload) && + event.payload.transcriptionSessionId === transcriptionSessionId && + event.payload.type === "close" && + isRecord(event.payload.talkEvent) && + event.payload.talkEvent.type === "session.closed" && + event.payload.talkEvent.final === true, + ), + ).toBe(true); + expect(() => + sendTalkTranscriptionRelayAudio({ + transcriptionSessionId, + connId: "conn-owner", + audioBase64: "AQI=", + }), + ).toThrow("Unknown transcription Talk session"); + } + expect( + events.some( + (event) => + isRecord(event.payload) && + (event.payload.transcriptionSessionId === firstSession.transcriptionSessionId || + event.payload.transcriptionSessionId === secondSession.transcriptionSessionId) && + (event.payload.type === "ready" || event.payload.type === "error"), + ), + ).toBe(false); + + sendTalkTranscriptionRelayAudio({ + transcriptionSessionId: unrelatedSession.transcriptionSessionId, + connId: "conn-other", + audioBase64: "AQI=", + }); + expect(unrelated.sendAudio).toHaveBeenCalledOnce(); + stopTalkTranscriptionRelaySession({ + transcriptionSessionId: unrelatedSession.transcriptionSessionId, + connId: "conn-other", + }); + closeTalkTranscriptionRelaySessionsForConnection("conn-other"); + expect(unrelated.close).toHaveBeenCalledOnce(); + }); + it("rejects provider configs that do not match relay audio input", () => { const provider = createTranscriptionProvider(createSttSessionMock()); const { context } = createBroadcastContext(); diff --git a/src/gateway/talk-transcription-relay.ts b/src/gateway/talk-transcription-relay.ts index a5d0acb3ba18..ceb2e431cedf 100644 --- a/src/gateway/talk-transcription-relay.ts +++ b/src/gateway/talk-transcription-relay.ts @@ -15,9 +15,11 @@ import { createTalkSessionController, } from "../talk/talk-session-controller.js"; import type { GatewayRequestContext } from "./server-methods/shared-types.js"; +import { formatError } from "./server-utils.js"; import { decodeTalkRelayAudioBase64 } from "./talk-relay-audio-base64.js"; import { closeExpiredTalkRelaySessions, + closeTalkRelaySessionsForConnection, requireActiveTalkRelaySession, } from "./talk-relay-session-lifecycle.js"; @@ -182,16 +184,35 @@ function closeTranscriptionSession( session.closed = true; transcriptionSessions.delete(session.id); clearTimeout(session.cleanupTimer); - session.sttSession.close(); - broadcastToOwner(session.context, session.connId, { - transcriptionSessionId: session.id, - type: "close", - reason, - talkEvent: session.talk.emit({ - type: "session.closed", - payload: { reason }, - final: true, - }), + try { + session.sttSession.close(); + } finally { + // Provider teardown may throw, but the owner-visible terminal event must + // still complete so disconnect cleanup cannot leave ambiguous state. + broadcastToOwner(session.context, session.connId, { + transcriptionSessionId: session.id, + type: "close", + reason, + talkEvent: session.talk.emit({ + type: "session.closed", + payload: { reason }, + final: true, + }), + }); + } +} + +/** Releases every transcription relay owned by a disconnected gateway connection. */ +export function closeTalkTranscriptionRelaySessionsForConnection(connId: string): void { + closeTalkRelaySessionsForConnection({ + sessions: transcriptionSessions.values(), + connId, + closeSession: (session) => closeTranscriptionSession(session, "completed"), + onCloseError: (error, session) => { + session.context.logGateway.warn( + `failed to close transcription relay session after connection disconnect: ${formatError(error)}`, + ); + }, }); } @@ -334,9 +355,16 @@ export function createTalkTranscriptionRelaySession( sttSession .connect() .then(() => { + if (transcriptionSessions.get(transcriptionSessionId) !== relay) { + return; + } emit({ transcriptionSessionId, type: "ready" }, { type: "session.ready", payload: null }); }) .catch((error: unknown) => { + const active = transcriptionSessions.get(transcriptionSessionId); + if (active !== relay) { + return; + } emit( { transcriptionSessionId, @@ -349,10 +377,7 @@ export function createTalkTranscriptionRelaySession( final: true, }, ); - const active = transcriptionSessions.get(transcriptionSessionId); - if (active) { - closeTranscriptionSession(active, "error"); - } + closeTranscriptionSession(active, "error"); }); return { From 796dbacdfc261681c57826dc8a9c5854c926ece0 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 31 Jul 2026 20:35:22 +0800 Subject: [PATCH 4/6] fix(gateway): ignore late talk relay callbacks --- .../talk-realtime-relay-session-create.ts | 65 +++++++++++++++---- src/gateway/talk-realtime-relay.test.ts | 34 +++++++++- src/gateway/talk-transcription-relay.test.ts | 19 +++++- src/gateway/talk-transcription-relay.ts | 54 +++++++++------ 4 files changed, 135 insertions(+), 37 deletions(-) diff --git a/src/gateway/talk-realtime-relay-session-create.ts b/src/gateway/talk-realtime-relay-session-create.ts index 5b2b02be6b29..aa9e9dec5283 100644 --- a/src/gateway/talk-realtime-relay-session-create.ts +++ b/src/gateway/talk-realtime-relay-session-create.ts @@ -103,12 +103,19 @@ export function createTalkRealtimeRelaySession( let ready = false; let failureEmitted = false; const relayRef: { current?: RelaySession } = {}; + const getActiveRelay = (): RelaySession | undefined => { + const relay = relayRef.current; + return relay && relaySessions.get(relay.id) === relay ? relay : undefined; + }; let consultAgentRuntime: ReturnType["agent"] | undefined; const relaySessionKey = params.sessionKey?.trim(); const relayAgentId = relaySessionKey ? resolveTalkSessionAgentId(params.cfg ?? params.context.getRuntimeConfig(), relaySessionKey) : undefined; const runAgentConsult = async ({ prompt, signal }: { prompt: string; signal?: AbortSignal }) => { + if (!getActiveRelay()) { + throw new Error("Realtime gateway-relay session is closed"); + } const runtimeConfig = params.cfg ?? params.context.getRuntimeConfig(); const sessionKey = relaySessionKey; if (!sessionKey) { @@ -182,9 +189,13 @@ export function createTalkRealtimeRelaySession( tools: params.tools, markStrategy: "transport", audioSink: { - isOpen: () => Boolean(relayRef.current && relaySessions.has(relayRef.current.id)), + isOpen: () => Boolean(getActiveRelay()), sendAudio: (audio) => { - const turnId = relayRef.current ? ensureRelayTurn(relayRef.current) : undefined; + const relay = getActiveRelay(); + if (!relay) { + return; + } + const turnId = ensureRelayTurn(relay); emit( { relaySessionId, @@ -201,7 +212,11 @@ export function createTalkRealtimeRelaySession( ); }, clearAudio: (reason) => { - const turnId = relayRef.current ? ensureRelayTurn(relayRef.current) : undefined; + const relay = getActiveRelay(); + if (!relay) { + return; + } + const turnId = ensureRelayTurn(relay); emit( { relaySessionId, type: "clear", ...(reason ? { reason } : {}) }, { @@ -213,7 +228,11 @@ export function createTalkRealtimeRelaySession( ); }, sendMark: (markName) => { - const turnId = relayRef.current ? ensureRelayTurn(relayRef.current) : undefined; + const relay = getActiveRelay(); + if (!relay) { + return; + } + const turnId = ensureRelayTurn(relay); emit( { relaySessionId, type: "mark", markName }, { @@ -226,6 +245,9 @@ export function createTalkRealtimeRelaySession( }, }, onEvent: (event) => { + if (!getActiveRelay()) { + return; + } if (event.direction !== "server") { return; } @@ -260,9 +282,12 @@ export function createTalkRealtimeRelaySession( } }, onTranscript: (role, text, final) => { - const relay = relayRef.current; - const turnId = relay ? ensureRelayTurn(relay) : undefined; - if (final && relay) { + const relay = getActiveRelay(); + if (!relay) { + return; + } + const turnId = ensureRelayTurn(relay); + if (final) { enqueueRelayVoiceTranscript(relay, role, text); } const eventType = @@ -289,7 +314,6 @@ export function createTalkRealtimeRelaySession( return; } if ( - relay && pruneInactiveRelayAgentRuns(relay) > 0 && shouldAutoControlRealtimeVoiceAgentText(question) ) { @@ -301,11 +325,17 @@ export function createTalkRealtimeRelaySession( text: question, }) .then((result) => { + if (!getActiveRelay()) { + return; + } if (result.speak && !result.suppress && result.message.trim()) { bridge.sendUserMessage(buildRealtimeVoiceAgentControlSpeechMessage(result.message)); } }) .catch((error: unknown) => { + if (!getActiveRelay()) { + return; + } emit( { relaySessionId, type: "error", message: formatErrorMessage(error) }, { @@ -323,9 +353,12 @@ export function createTalkRealtimeRelaySession( } }, onToolCall: (toolCall) => { - const relay = relayRef.current; + const relay = getActiveRelay(); + if (!relay) { + return; + } let shouldSubmitWorkingResult = false; - if (relay && toolCall.name === REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME) { + if (toolCall.name === REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME) { const forcedConsult = relay.harness.forcedConsults.recordNativeConsult( toolCall.args, toolCall.callId, @@ -351,7 +384,7 @@ export function createTalkRealtimeRelaySession( } shouldSubmitWorkingResult = true; } - const turnId = relay ? ensureRelayTurn(relay) : undefined; + const turnId = ensureRelayTurn(relay); emit( { relaySessionId, @@ -369,15 +402,21 @@ export function createTalkRealtimeRelaySession( payload: { name: toolCall.name, args: toolCall.args }, }, ); - if (relay && shouldSubmitWorkingResult) { + if (shouldSubmitWorkingResult) { return submitRealtimeAgentConsultWorkingResponse(relay, toolCall.callId, turnId); } }, onReady: () => { + if (!getActiveRelay()) { + return; + } ready = true; emit({ relaySessionId, type: "ready" }, { type: "session.ready", payload: null }); }, onError: (error) => { + if (!getActiveRelay()) { + return; + } const issue = realtimeRelayIssue({ message: formatErrorMessage(error), provider: params.provider.id, @@ -393,7 +432,7 @@ export function createTalkRealtimeRelaySession( }, onClose: (reason) => { const active = relaySessions.get(relaySessionId); - if (!active) { + if (active !== relayRef.current) { return; } active.harness.close(); diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index 6609090e9d5c..d6559482d752 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -101,13 +101,18 @@ describe("talk realtime gateway relay", () => { setTestEnvValue("OPENCLAW_STATE_DIR", tempDir); const bridgeCloses: Array> = []; const bridgeAudioSends: Array> = []; + const bridgeRequests: RealtimeVoiceBridgeCreateRequest[] = []; + const bridgeToolResults: Array> = []; const provider = createIdleRelayProvider(); - provider.createBridge = () => { + provider.createBridge = (request) => { const bridgeIndex = bridgeCloses.length; const close = vi.fn(); const sendAudio = vi.fn(); + const submitToolResult = vi.fn(); + bridgeRequests.push(request); bridgeCloses.push(close); bridgeAudioSends.push(sendAudio); + bridgeToolResults.push(submitToolResult); return { connect: vi.fn(async () => { if (bridgeIndex === 1) { @@ -117,7 +122,7 @@ describe("talk realtime gateway relay", () => { sendAudio, setMediaTimestamp: vi.fn(), handleBargeIn: vi.fn(), - submitToolResult: vi.fn(), + submitToolResult, acknowledgeMark: vi.fn(), close, isConnected: vi.fn(() => true), @@ -192,6 +197,31 @@ describe("talk realtime gateway relay", () => { payload.type === "error", ), ).toBe(false); + const eventCountAfterClose = broadcastToConnIds.mock.calls.length; + const lateRequest = bridgeRequests[0]; + if (!lateRequest?.runAgentConsult) { + throw new Error("expected relay provider request to include the agent consult runner"); + } + lateRequest.onReady?.(); + lateRequest.onError?.(new Error("late provider error")); + lateRequest.onEvent?.({ direction: "server", type: "response.done" }); + lateRequest.onAudio(Buffer.from("late audio")); + lateRequest.onClearAudio("barge-in"); + lateRequest.onMark?.("late-mark"); + lateRequest.onTranscript?.("user", "late transcript", true); + lateRequest.onToolCall?.({ + itemId: "late-item", + callId: "late-call", + name: "openclaw_agent_consult", + args: { question: "late consult" }, + }); + lateRequest.onClose?.("error"); + await expect(lateRequest.runAgentConsult({ prompt: "late direct consult" })).rejects.toThrow( + "Realtime gateway-relay session is closed", + ); + await Promise.resolve(); + expect(broadcastToConnIds).toHaveBeenCalledTimes(eventCountAfterClose); + expect(bridgeToolResults[0]).not.toHaveBeenCalled(); expect(() => sendTalkRealtimeRelayAudio({ relaySessionId: firstOwned.relaySessionId, diff --git a/src/gateway/talk-transcription-relay.test.ts b/src/gateway/talk-transcription-relay.test.ts index 5614d3fe7f91..2d7456d6a323 100644 --- a/src/gateway/talk-transcription-relay.test.ts +++ b/src/gateway/talk-transcription-relay.test.ts @@ -4,6 +4,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import type { RealtimeTranscriptionProviderPlugin } from "../plugins/types.js"; import type { RealtimeTranscriptionSessionCreateRequest } from "../realtime-transcription/provider-types.js"; +import { getUnifiedTalkSession, rememberUnifiedTalkSession } from "./talk-session-registry.js"; import { cancelTalkTranscriptionRelayTurn, closeTalkTranscriptionRelaySessionsForConnection, @@ -221,17 +222,25 @@ describe("talk transcription gateway relay", () => { throw new Error("late connect failure"); }); const unrelated = createSttSessionMock(); + const requests: RealtimeTranscriptionSessionCreateRequest[] = []; const { context, events, logGateway } = createBroadcastContext(); const createSession = (connId: string, sttSession: ReturnType) => createTalkTranscriptionRelaySession({ context, connId, - provider: createTranscriptionProvider(sttSession), + provider: createTranscriptionProvider(sttSession, (request) => requests.push(request)), providerConfig: {}, }); const firstSession = createSession("conn-owner", firstOwned); const secondSession = createSession("conn-owner", secondOwned); const unrelatedSession = createSession("conn-other", unrelated); + for (const session of [firstSession, secondSession]) { + rememberUnifiedTalkSession(session.transcriptionSessionId, { + kind: "transcription-relay", + connId: "conn-owner", + transcriptionSessionId: session.transcriptionSessionId, + }); + } firstOwned.close.mockImplementationOnce(() => { throw new Error("provider close failed"); }); @@ -269,6 +278,7 @@ describe("talk transcription gateway relay", () => { audioBase64: "AQI=", }), ).toThrow("Unknown transcription Talk session"); + expect(() => getUnifiedTalkSession(transcriptionSessionId)).toThrow("Unknown Talk session"); } expect( events.some( @@ -279,6 +289,13 @@ describe("talk transcription gateway relay", () => { (event.payload.type === "ready" || event.payload.type === "error"), ), ).toBe(false); + const eventCountAfterClose = events.length; + requests[0]?.onSpeechStart?.(); + requests[0]?.onPartial?.("late partial"); + requests[0]?.onTranscript?.("late transcript"); + requests[0]?.onError?.(new Error("late provider error")); + await Promise.resolve(); + expect(events).toHaveLength(eventCountAfterClose); sendTalkTranscriptionRelayAudio({ transcriptionSessionId: unrelatedSession.transcriptionSessionId, diff --git a/src/gateway/talk-transcription-relay.ts b/src/gateway/talk-transcription-relay.ts index ceb2e431cedf..2107f979891e 100644 --- a/src/gateway/talk-transcription-relay.ts +++ b/src/gateway/talk-transcription-relay.ts @@ -22,6 +22,7 @@ import { closeTalkRelaySessionsForConnection, requireActiveTalkRelaySession, } from "./talk-relay-session-lifecycle.js"; +import { forgetUnifiedTalkSession } from "./talk-session-registry.js"; /** * Gateway-owned relay for streaming speech-to-text providers used by Talk. @@ -183,6 +184,7 @@ function closeTranscriptionSession( } session.closed = true; transcriptionSessions.delete(session.id); + forgetUnifiedTalkSession(session.id); clearTimeout(session.cleanupTimer); try { session.sttSession.close(); @@ -272,18 +274,26 @@ export function createTalkTranscriptionRelaySession( }); }; const relayRef: { current?: TranscriptionRelaySession } = {}; - const ensureTurnId = (): string => { + const getActiveRelay = (): TranscriptionRelaySession | undefined => { const relay = relayRef.current; - return relay ? ensureTranscriptionTurn(relay) : "turn-1"; + return relay && transcriptionSessions.get(relay.id) === relay ? relay : undefined; }; const sttSession = params.provider.createSession({ cfg: params.context.getRuntimeConfig(), providerConfig: params.providerConfig, onSpeechStart: () => { - ensureTurnId(); + const relay = getActiveRelay(); + if (!relay) { + return; + } + ensureTranscriptionTurn(relay); }, onPartial: (text) => { - const turnId = ensureTurnId(); + const relay = getActiveRelay(); + if (!relay) { + return; + } + const turnId = ensureTranscriptionTurn(relay); emit( { transcriptionSessionId, type: "partial", text }, { @@ -294,7 +304,11 @@ export function createTalkTranscriptionRelaySession( ); }, onTranscript: (text) => { - const turnId = ensureTurnId(); + const relay = getActiveRelay(); + if (!relay) { + return; + } + const turnId = ensureTranscriptionTurn(relay); emit( { transcriptionSessionId, type: "transcript", text, final: true }, { @@ -304,21 +318,22 @@ export function createTalkTranscriptionRelaySession( final: true, }, ); - const relay = relayRef.current; - if (relay) { - const ended = relay.talk.endTurn({ turnId, payload: {} }); - if (ended.ok) { - broadcastToOwner(relay.context, relay.connId, { - transcriptionSessionId, - type: "transcript", - text: "", - final: true, - talkEvent: ended.event, - }); - } + const ended = relay.talk.endTurn({ turnId, payload: {} }); + if (ended.ok) { + broadcastToOwner(relay.context, relay.connId, { + transcriptionSessionId, + type: "transcript", + text: "", + final: true, + talkEvent: ended.event, + }); } }, onError: (error) => { + const relay = getActiveRelay(); + if (!relay) { + return; + } emit( { transcriptionSessionId, type: "error", message: error.message }, { @@ -327,10 +342,7 @@ export function createTalkTranscriptionRelaySession( final: true, }, ); - const relay = relayRef.current; - if (relay) { - closeTranscriptionSession(relay, "error"); - } + closeTranscriptionSession(relay, "error"); }, }); const relay: TranscriptionRelaySession = { From bec00de4a802e025eb9ca631b7edca521a5df7f4 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 31 Jul 2026 20:36:28 +0800 Subject: [PATCH 5/6] fix(gateway): ignore early realtime relay close --- .../talk-realtime-relay-session-create.ts | 2 +- src/gateway/talk-realtime-relay.test.ts | 27 +++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/src/gateway/talk-realtime-relay-session-create.ts b/src/gateway/talk-realtime-relay-session-create.ts index aa9e9dec5283..250d950d3a97 100644 --- a/src/gateway/talk-realtime-relay-session-create.ts +++ b/src/gateway/talk-realtime-relay-session-create.ts @@ -432,7 +432,7 @@ export function createTalkRealtimeRelaySession( }, onClose: (reason) => { const active = relaySessions.get(relaySessionId); - if (active !== relayRef.current) { + if (!active || active !== relayRef.current) { return; } active.harness.close(); diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index d6559482d752..b3015fe408b6 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -287,6 +287,33 @@ describe("talk realtime gateway relay", () => { }); }); + it("ignores a provider close before relay registration", () => { + const provider = createIdleRelayProvider(); + provider.createBridge = (request) => { + request.onClose?.("error"); + return createIdleRelayProvider().createBridge(request); + }; + + const session = createTalkRealtimeRelaySession({ + context: { + broadcastToConnIds: vi.fn(), + chatAbortControllers: new Map(), + getRuntimeConfig: () => ({}), + logGateway: { warn: vi.fn() }, + } as never, + connId: "conn-early-close", + provider, + providerConfig: {}, + instructions: "brief", + tools: [], + }); + + stopTalkRealtimeRelaySession({ + relaySessionId: session.relaySessionId, + connId: "conn-early-close", + }); + }); + it("appends finalized relay transcripts to the canonical agent session", async () => { const envSnapshot = captureEnv(["OPENCLAW_STATE_DIR"]); const tempDir = await fs.realpath( From 68faa23fa8c7cc3d20ebf6beff22c4f8e066c64c Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 31 Jul 2026 20:54:07 +0800 Subject: [PATCH 6/6] fix(gateway): reject terminal relay construction --- .../talk-realtime-relay-session-create.ts | 32 +++++++- src/gateway/talk-realtime-relay.test.ts | 80 +++++++++++++------ 2 files changed, 86 insertions(+), 26 deletions(-) diff --git a/src/gateway/talk-realtime-relay-session-create.ts b/src/gateway/talk-realtime-relay-session-create.ts index 250d950d3a97..fc75eefc87b5 100644 --- a/src/gateway/talk-realtime-relay-session-create.ts +++ b/src/gateway/talk-realtime-relay-session-create.ts @@ -11,7 +11,10 @@ import { shouldAutoControlRealtimeVoiceAgentText, } from "../talk/agent-run-control.js"; import { resolveTalkSessionAgentId } from "../talk/agent-target.js"; -import { REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ } from "../talk/provider-types.js"; +import { + REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ, + type RealtimeVoiceCloseReason, +} from "../talk/provider-types.js"; import { createRealtimeVoiceSessionHarness } from "../talk/realtime-session-harness.js"; import type { TalkEventInput } from "../talk/talk-session-controller.js"; import { registerChatAbortController } from "./chat-abort.js"; @@ -102,6 +105,9 @@ export function createTalkRealtimeRelaySession( let currentOutputResponseId: string | undefined; let ready = false; let failureEmitted = false; + const constructionTerminal: { + current?: { kind: "error"; error: Error } | { kind: "close"; reason: RealtimeVoiceCloseReason }; + } = {}; const relayRef: { current?: RelaySession } = {}; const getActiveRelay = (): RelaySession | undefined => { const relay = relayRef.current; @@ -414,7 +420,11 @@ export function createTalkRealtimeRelaySession( emit({ relaySessionId, type: "ready" }, { type: "session.ready", payload: null }); }, onError: (error) => { - if (!getActiveRelay()) { + const active = getActiveRelay(); + if (!active) { + if (!relayRef.current) { + constructionTerminal.current ??= { kind: "error", error }; + } return; } const issue = realtimeRelayIssue({ @@ -433,6 +443,9 @@ export function createTalkRealtimeRelaySession( onClose: (reason) => { const active = relaySessions.get(relaySessionId); if (!active || active !== relayRef.current) { + if (!relayRef.current) { + constructionTerminal.current ??= { kind: "close", reason }; + } return; } active.harness.close(); @@ -460,6 +473,21 @@ export function createTalkRealtimeRelaySession( ); }, }); + const earlyTerminal = constructionTerminal.current; + if (earlyTerminal) { + harness.close(); + try { + bridge.close(); + } catch (error) { + params.context.logGateway.warn( + `failed to close realtime relay bridge after provider terminated during creation: ${formatErrorMessage(error)}`, + ); + } + if (earlyTerminal.kind === "error") { + throw earlyTerminal.error; + } + throw new Error(`Realtime provider closed during session creation: ${earlyTerminal.reason}`); + } const initialSessionKey = params.sessionKey?.trim() || undefined; const relay: RelaySession = { id: relaySessionId, diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index b3015fe408b6..d715b63c274d 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -287,32 +287,64 @@ describe("talk realtime gateway relay", () => { }); }); - it("ignores a provider close before relay registration", () => { - const provider = createIdleRelayProvider(); - provider.createBridge = (request) => { - request.onClose?.("error"); - return createIdleRelayProvider().createBridge(request); - }; + it.each([ + { + name: "error before close", + terminate: (request: RealtimeVoiceBridgeCreateRequest) => { + request.onError?.(new Error("provider rejected session")); + request.onClose?.("error"); + }, + expectedError: "provider rejected session", + }, + { + name: "close before error", + terminate: (request: RealtimeVoiceBridgeCreateRequest) => { + request.onClose?.("completed"); + request.onError?.(new Error("late provider error")); + }, + expectedError: "Realtime provider closed during session creation: completed", + }, + ])( + "rejects a synchronous provider $name during bridge creation", + ({ terminate, expectedError }) => { + const connect = vi.fn(async () => undefined); + const sendAudio = vi.fn(); + const close = vi.fn(); + const bridge = { + ...createIdleRelayProvider().createBridge({} as never), + connect, + sendAudio, + close, + }; + const provider = createIdleRelayProvider(); + provider.createBridge = (request) => { + terminate(request); + return bridge; + }; + const broadcastToConnIds = vi.fn(); - const session = createTalkRealtimeRelaySession({ - context: { - broadcastToConnIds: vi.fn(), - chatAbortControllers: new Map(), - getRuntimeConfig: () => ({}), - logGateway: { warn: vi.fn() }, - } as never, - connId: "conn-early-close", - provider, - providerConfig: {}, - instructions: "brief", - tools: [], - }); + expect(() => + createTalkRealtimeRelaySession({ + context: { + broadcastToConnIds, + chatAbortControllers: new Map(), + getRuntimeConfig: () => ({}), + logGateway: { warn: vi.fn() }, + } as never, + connId: "conn-early-terminal", + provider, + providerConfig: {}, + instructions: "brief", + tools: [], + }), + ).toThrow(expectedError); - stopTalkRealtimeRelaySession({ - relaySessionId: session.relaySessionId, - connId: "conn-early-close", - }); - }); + expect(connect).not.toHaveBeenCalled(); + expect(sendAudio).not.toHaveBeenCalled(); + expect(close).toHaveBeenCalledOnce(); + expect(broadcastToConnIds).not.toHaveBeenCalled(); + }, + ); it("appends finalized relay transcripts to the canonical agent session", async () => { const envSnapshot = captureEnv(["OPENCLAW_STATE_DIR"]);