fix(talk): retain bounded relay close owners

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) {

src/gateway/talk-realtime-relay-state.ts

@@ -145,6 +145,10 @@ export type TalkRealtimeRelaySessionResult = {
 };
 
 export const relaySessions = new Map<string, RelaySession>();
+// 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<RelaySession>();
 
 export function broadcastToOwner(
   context: GatewayRequestContext,

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

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<void> {
     .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;
 }

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",
       );