diff --git a/ui/src/pages/chat/realtime-talk-lifecycle.test.ts b/ui/src/pages/chat/realtime-talk-lifecycle.test.ts index f54ce1530d62..adb961546c63 100644 --- a/ui/src/pages/chat/realtime-talk-lifecycle.test.ts +++ b/ui/src/pages/chat/realtime-talk-lifecycle.test.ts @@ -5,6 +5,7 @@ import type { RealtimeTalkTransportContext } from "./realtime-talk-shared.ts"; const transportMock = vi.hoisted(() => ({ relayContexts: [] as RealtimeTalkTransportContext[], webRtcContexts: [] as RealtimeTalkTransportContext[], + webRtcStops: [] as Array>, start: vi.fn(async () => undefined), stop: vi.fn(), })); @@ -27,7 +28,9 @@ vi.mock("./realtime-talk-webrtc.ts", () => ({ context: RealtimeTalkTransportContext, ) { transportMock.webRtcContexts.push(context); - return { start: transportMock.start, stop: transportMock.stop }; + const stop = vi.fn((options?: { emitClosed?: boolean }) => transportMock.stop(options)); + transportMock.webRtcStops.push(stop); + return { start: transportMock.start, stop }; }), })); @@ -64,6 +67,7 @@ describe("RealtimeTalkSession lifecycle", () => { beforeEach(() => { transportMock.relayContexts.length = 0; transportMock.webRtcContexts.length = 0; + transportMock.webRtcStops.length = 0; transportMock.start.mockClear(); transportMock.stop.mockClear(); }); @@ -137,8 +141,9 @@ describe("RealtimeTalkSession lifecycle", () => { await secondContext.flushTranscriptWrites?.(); expect(transcriptEntryIds).toEqual(["1", "2"]); - expect(transportMock.stop).toHaveBeenCalledOnce(); - expect(transportMock.stop).toHaveBeenCalledWith({ emitClosed: false }); + expect(transportMock.webRtcStops[0]).toHaveBeenCalledOnce(); + expect(transportMock.webRtcStops[0]).toHaveBeenCalledWith({ emitClosed: false }); + expect(transportMock.webRtcStops[1]).not.toHaveBeenCalled(); expect(request.mock.calls.filter(([method]) => method === "talk.client.create")).toEqual([ [ "talk.client.create", @@ -156,6 +161,8 @@ describe("RealtimeTalkSession lifecycle", () => { ], ]); session.stop(); + expect(transportMock.webRtcStops[1]).toHaveBeenCalledOnce(); + expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith(); await vi.waitFor(() => expect(request.mock.calls.filter(([method]) => method === "talk.client.close")).toHaveLength( 1, @@ -196,8 +203,9 @@ describe("RealtimeTalkSession lifecycle", () => { await expect(session.start()).rejects.toThrow("replacement startup failed"); expect(request.mock.calls.some(([method]) => method === "talk.client.close")).toBe(false); - expect(transportMock.stop).toHaveBeenCalledOnce(); - expect(transportMock.stop).toHaveBeenCalledWith({ emitClosed: false }); + expect(transportMock.webRtcStops[0]).not.toHaveBeenCalled(); + expect(transportMock.webRtcStops[1]).toHaveBeenCalledOnce(); + expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false }); existingContext.callbacks.onTranscript?.({ role: "user", text: "still active", final: true }); await existingContext.flushTranscriptWrites?.(); @@ -209,6 +217,44 @@ describe("RealtimeTalkSession lifecycle", () => { session.stop(); }); + it("ignores transcripts from a superseded pending replacement", async () => { + const firstReplacementStart = createDeferred(); + const transcriptEntryIds: string[] = []; + const request = vi.fn(async (method: string, params?: { entryId?: string }) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: "voice-overlapping-replacements", + clientSecret: "secret", + }; + } + if (method === "talk.client.transcript") { + transcriptEntryIds.push(String(params?.entryId)); + } + return { ok: true }; + }); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); + await session.start(); + transportMock.start.mockImplementationOnce(async () => await firstReplacementStart.promise); + + const firstReplacement = session.start(); + await vi.waitFor(() => expect(transportMock.webRtcContexts).toHaveLength(2)); + await session.start(); + + const supersededContext = transcriptContext(transportMock.webRtcContexts, 1); + const activeContext = transcriptContext(transportMock.webRtcContexts, 2); + supersededContext.callbacks.onTranscript?.({ role: "user", text: "stale", final: true }); + activeContext.callbacks.onTranscript?.({ role: "user", text: "active", final: true }); + await activeContext.flushTranscriptWrites?.(); + expect(transcriptEntryIds).toEqual(["1"]); + expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false }); + + firstReplacementStart.resolve(); + await firstReplacement; + session.stop(); + }); + it("releases newly allocated owners after transport startup failures", async () => { let createCount = 0; const request = vi.fn(async (method: string) => { @@ -274,13 +320,24 @@ describe("RealtimeTalkSession lifecycle", () => { const replacing = session.start(); await vi.waitFor(() => expect(transportMock.webRtcContexts).toHaveLength(2)); + existingContext.callbacks.onTranscript?.({ + role: "user", + text: "while replacement starts", + final: true, + }); + await existingContext.flushTranscriptWrites?.(); + expect(transcriptEntryIds).toEqual(["1"]); + session.stop(); replacementStart.reject(new Error("replacement failed after stop")); await expect(replacing).rejects.toThrow("replacement failed after stop"); existingContext.callbacks.onTranscript?.({ role: "user", text: "too late", final: true }); await Promise.resolve(); - expect(transcriptEntryIds).toEqual([]); + expect(transcriptEntryIds).toEqual(["1"]); + expect(transportMock.webRtcStops[0]).toHaveBeenCalledOnce(); + expect(transportMock.webRtcStops[0]).toHaveBeenCalledWith(); + expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false }); await vi.waitFor(() => expect(request.mock.calls.filter(([method]) => method === "talk.client.close")).toHaveLength( 1, diff --git a/ui/src/pages/chat/realtime-talk.ts b/ui/src/pages/chat/realtime-talk.ts index ceb02a4c958c..715c4ffe2e11 100644 --- a/ui/src/pages/chat/realtime-talk.ts +++ b/ui/src/pages/chat/realtime-talk.ts @@ -142,6 +142,7 @@ function compactLaunchParams( export class RealtimeTalkSession { private transport: RealtimeTalkTransport | null = null; + private pendingTransport: RealtimeTalkTransport | null = null; private closed = false; private lifecycleGeneration = 0; private videoEnabled = false; @@ -164,22 +165,14 @@ export class RealtimeTalkSession { async start(): Promise { const lifecycleGeneration = ++this.lifecycleGeneration; + const supersededPendingTransport = this.pendingTransport; + this.pendingTransport = null; + supersededPendingTransport?.stop({ emitClosed: false }); this.closed = false; this.callbacks.onStatus?.("connecting"); const existingTransport = this.transport; - const existingTransportGeneration = this.transportGeneration; const existingVoiceSessionId = this.voiceSessionId; - const existingAcceptingTranscripts = this.acceptingTranscripts; - const existingServerOwnedVoiceSession = this.serverOwnedVoiceSession; const existingOwner = this.clientVoiceSessionOwner; - const restoreExistingSession = () => { - this.transport = existingTransport; - this.transportGeneration = existingTransportGeneration; - this.voiceSessionId = existingVoiceSessionId; - this.acceptingTranscripts = existingAcceptingTranscripts; - this.serverOwnedVoiceSession = existingServerOwnedVoiceSession; - this.clientVoiceSessionOwner = existingOwner; - }; const owner = reserveClientVoiceSessionOwner(this.client, this.sessionKey); let ownerTransferred = false; try { @@ -226,22 +219,15 @@ export class RealtimeTalkSession { if (adoptedOwner !== owner) { owner.release(); } - this.voiceSessionId = voiceSessionId; - this.acceptingTranscripts = true; - this.serverOwnedVoiceSession = transport === "gateway-relay"; - this.transportGeneration += 1; - if (this.serverOwnedVoiceSession) { - owner.release(); - } else { - this.clientVoiceSessionOwner = adoptedOwner; - } - ownerTransferred = true; + // Candidate generations must be unique without retiring the committed transport. + // Overlapping starts can then fence every superseded candidate independently. + const nextTransportGeneration = lifecycleGeneration; const callbacks = transport === "gateway-relay" ? this.callbacks : this.clientOwnedTranscriptCallbacks( voiceSessionId, - this.transportGeneration, + nextTransportGeneration, adoptedOwner.signal, ); const transcriptQueue = this.transcriptQueue; @@ -258,41 +244,46 @@ export class RealtimeTalkSession { consultThinkingLevel: session.consultThinkingLevel, consultFastMode: session.consultFastMode, }); - this.transport = nextTransport; + this.pendingTransport = nextTransport; this.callbacks.onVideoCapability?.( providerVideoCapable && typeof nextTransport.setVideoEnabled === "function", ); await nextTransport.start(); - if ( - this.closed || - lifecycleGeneration !== this.lifecycleGeneration || - this.transport !== nextTransport - ) { - nextTransport.stop({ emitClosed: false }); - return; - } - existingTransport?.stop({ emitClosed: false }); } catch (error) { - const canRollback = - lifecycleGeneration === this.lifecycleGeneration && - (!nextTransport || this.transport === nextTransport); + if (this.pendingTransport === nextTransport) { + this.pendingTransport = null; + } nextTransport?.stop({ emitClosed: false }); - if (!canRollback) { - throw error; - } - if (existingOwner && adoptedOwner === existingOwner) { - // A same-session replacement borrows the existing owner and queue. - // Restore the live transport instead of closing their shared allocation. - restoreExistingSession(); - } else { - const detached = this.detachVoiceSession(); - restoreExistingSession(); - if (detached) { - this.closeLogicalVoiceSession(detached); - } + if (!(existingOwner && adoptedOwner === existingOwner)) { + this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner); } + ownerTransferred = true; throw error; } + if (this.pendingTransport === nextTransport) { + this.pendingTransport = null; + } + if (this.closed || lifecycleGeneration !== this.lifecycleGeneration) { + nextTransport.stop({ emitClosed: false }); + if (!(existingOwner && adoptedOwner === existingOwner)) { + this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner); + } + ownerTransferred = true; + return; + } + this.voiceSessionId = voiceSessionId; + this.acceptingTranscripts = true; + this.serverOwnedVoiceSession = transport === "gateway-relay"; + this.transportGeneration = nextTransportGeneration; + this.transport = nextTransport; + if (this.serverOwnedVoiceSession) { + owner.release(); + this.clientVoiceSessionOwner = undefined; + } else { + this.clientVoiceSessionOwner = adoptedOwner; + } + ownerTransferred = true; + existingTransport?.stop({ emitClosed: false }); } finally { if (!ownerTransferred) { owner.release(); @@ -402,6 +393,9 @@ export class RealtimeTalkSession { this.videoEnabled = false; activeRealtimeTalkSessions.delete(this); this.callbacks.onStatus?.("idle"); + const pendingTransport = this.pendingTransport; + this.pendingTransport = null; + pendingTransport?.stop({ emitClosed: false }); const detached = this.detachVoiceSession(); this.transport?.stop(); this.transport = null;