diff --git a/src/gateway/talk-realtime-relay-operations.ts b/src/gateway/talk-realtime-relay-operations.ts index bbbc20356de7..7ebd5325def8 100644 --- a/src/gateway/talk-realtime-relay-operations.ts +++ b/src/gateway/talk-realtime-relay-operations.ts @@ -26,6 +26,7 @@ import { MAX_RELAY_SESSIONS_GLOBAL, MAX_RELAY_SESSIONS_PER_CONN, broadcastToOwner, + drainingRelaySessions, ensureRelayTurn, noFallbackRelayOutputFlush, relaySessions, @@ -143,12 +144,17 @@ function countRelaySessionsForConn(connId: string): number { count += 1; } } + for (const session of drainingRelaySessions.values()) { + if (session.connId === connId) { + count += 1; + } + } return count; } export function enforceRelaySessionLimits(connId: string): void { pruneExpiredRelaySessions(); - if (relaySessions.size >= MAX_RELAY_SESSIONS_GLOBAL) { + if (relaySessions.size + drainingRelaySessions.size >= MAX_RELAY_SESSIONS_GLOBAL) { throw new Error("Too many active realtime relay sessions"); } if (countRelaySessionsForConn(connId) >= MAX_RELAY_SESSIONS_PER_CONN) { diff --git a/src/gateway/talk-realtime-relay-state.ts b/src/gateway/talk-realtime-relay-state.ts index 6e98cea8498a..5531e207fd86 100644 --- a/src/gateway/talk-realtime-relay-state.ts +++ b/src/gateway/talk-realtime-relay-state.ts @@ -145,6 +145,10 @@ export type TalkRealtimeRelaySessionResult = { }; export const relaySessions = new Map(); +// Closed relays leave the active map immediately so late provider/client events +// are ignored, but their accepted transcript prefix still owns bounded memory +// until durable close settles. Session limits count both maps. +export const drainingRelaySessions = new Set(); export function broadcastToOwner( context: GatewayRequestContext, diff --git a/src/gateway/talk-realtime-relay-voice.test.ts b/src/gateway/talk-realtime-relay-voice.test.ts index f3c55dcc041b..f22620fc1e65 100644 --- a/src/gateway/talk-realtime-relay-voice.test.ts +++ b/src/gateway/talk-realtime-relay-voice.test.ts @@ -64,9 +64,13 @@ describe("realtime relay voice transcript persistence", () => { }, ); const { session, failVoiceTranscriptPersistence } = createRelaySession(); - let accepted = 0; + let accepted = enqueueRelayVoiceTranscript(session, "user", ` ${"x".repeat(9_000)} `) ? 1 : 0; for (let index = 0; index < 10_000; index += 1) { + expect(enqueueRelayVoiceTranscript(session, "user", " \t\n ")).toBe(true); + } + + for (let index = 1; index < 10_000; index += 1) { if ( enqueueRelayVoiceTranscript( session, @@ -109,6 +113,11 @@ describe("realtime relay voice transcript persistence", () => { session.sessionKey = undefined; session.agentId = undefined; + for (let index = 0; index < 100; index += 1) { + expect(enqueueRelayVoiceTranscript(session, "user", " \t\n ")).toBe(true); + } + expect(session.pendingVoiceTranscripts).toHaveLength(0); + for (let index = 0; index < 100; index += 1) { expect(enqueueRelayVoiceTranscript(session, "user", ` ${"x".repeat(9_000)} `)).toBe(true); } diff --git a/src/gateway/talk-realtime-relay-voice.ts b/src/gateway/talk-realtime-relay-voice.ts index 444854265824..393d16023bdf 100644 --- a/src/gateway/talk-realtime-relay-voice.ts +++ b/src/gateway/talk-realtime-relay-voice.ts @@ -9,7 +9,7 @@ import { normalizeVoiceTranscriptText, VOICE_TRANSCRIPT_QUEUE_POLICY, } from "../talk/voice-transcript.js"; -import type { RelaySession } from "./talk-realtime-relay-state.js"; +import { drainingRelaySessions, type RelaySession } from "./talk-realtime-relay-state.js"; const RELAY_TRANSCRIPT_RETRY_DELAYS_MS = [0, 500, 2_000] as const; @@ -73,6 +73,9 @@ export function enqueueRelayVoiceTranscript( text: string, ): boolean { const normalizedText = normalizeVoiceTranscriptText(text); + if (!normalizedText) { + return true; + } if (!session.sessionKey) { // Lazy-bound relays hear audio before talk.client.toolCall supplies the session // key; buffer bounded finals so the call's opening turns survive the binding. @@ -154,5 +157,9 @@ export function closeRelayVoiceSession(session: RelaySession): Promise { .catch((error: unknown) => { logRelayVoiceFailure(session, "realtime relay voice session close failed", error); }); + drainingRelaySessions.add(session); + void session.voiceSessionClose.finally(() => { + drainingRelaySessions.delete(session); + }); return session.voiceSessionClose; } diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index 653031a04a52..731a5b51238f 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -23,7 +23,7 @@ import type { } from "../talk/provider-types.js"; import { captureEnv, setTestEnvValue } from "../test-utils/env.js"; import { createChatRunState } from "./server-chat-state.js"; -import { relaySessions } from "./talk-realtime-relay-state.js"; +import { drainingRelaySessions, relaySessions } from "./talk-realtime-relay-state.js"; import { acknowledgeTalkRealtimeRelayMark, cancelTalkRealtimeRelayTurn, @@ -60,7 +60,7 @@ function stopTalkRealtimeRelaySession( } describe("talk realtime gateway relay", () => { - afterEach(() => { + afterEach(async () => { for (const [relaySessionId, connId] of activeRelaySessions) { try { stopTalkRealtimeRelaySessionRaw({ relaySessionId, connId }); @@ -74,6 +74,9 @@ describe("talk realtime gateway relay", () => { } } activeRelaySessions.clear(); + await Promise.all( + [...drainingRelaySessions].map((session) => session.voiceSessionClose ?? Promise.resolve()), + ); vi.useRealTimers(); embeddedRunTesting.resetActiveEmbeddedRuns(); }); @@ -504,9 +507,38 @@ describe("talk realtime gateway relay", () => { ]); expect(bridgeClose).toHaveBeenCalledOnce(); expect(relaySessions.has(session.relaySessionId)).toBe(false); + expect(drainingRelaySessions.has(relay)).toBe(true); + + createTalkRealtimeRelaySession({ + context: { + broadcastToConnIds: vi.fn(), + getRuntimeConfig: () => ({}), + logGateway: { warn: vi.fn() }, + } as never, + connId: "conn-voice-overflow", + provider, + providerConfig: {}, + instructions: "brief", + tools: [], + }); + expect(() => + createTalkRealtimeRelaySession({ + context: { + broadcastToConnIds: vi.fn(), + getRuntimeConfig: () => ({}), + logGateway: { warn: vi.fn() }, + } as never, + connId: "conn-voice-overflow", + provider, + providerConfig: {}, + instructions: "brief", + tools: [], + }), + ).toThrow("Too many active realtime relay sessions for this connection"); releaseQueue(); await relay.voiceSessionClose; + expect(drainingRelaySessions.has(relay)).toBe(false); expect(clientVoiceSessionTesting.readRecord("main", session.relaySessionId)?.status).toBe( "closed", );