diff --git a/src/puter-js/src/modules/ai/convo.test.js b/src/puter-js/src/modules/ai/convo.test.js index a260bf387..923b1b816 100644 --- a/src/puter-js/src/modules/ai/convo.test.js +++ b/src/puter-js/src/modules/ai/convo.test.js @@ -2126,3 +2126,104 @@ describe('reconnect retries', () => { }); }); }); + +describe('microphone through a reconnect', () => { + const worklets = []; + const audioTrack = { enabled: true, readyState: 'live', stop: vi.fn() }; + const getUserMedia = vi.fn(async () => ({ + getTracks: () => [audioTrack], + getAudioTracks: () => [audioTrack], + })); + beforeEach(() => { + worklets.length = 0; + audioTrack.stop.mockClear(); + getUserMedia.mockClear(); + vi.stubGlobal('window', {}); + vi.stubGlobal( + 'AudioContext', + class { + state = 'running'; + destination = {}; + audioWorklet = { addModule: async () => {} }; + resume = async () => {}; + close = vi.fn(async () => {}); + createMediaStreamSource() { + return { connect() {}, disconnect() {} }; + } + }, + ); + vi.stubGlobal( + 'AudioWorkletNode', + class { + port = { + onmessage: null, + postMessage: vi.fn(), + close: vi.fn(), + }; + connect() {} + disconnect() {} + constructor() { + worklets.push(this); + } + }, + ); + vi.stubGlobal('navigator', { mediaDevices: { getUserMedia } }); + vi.spyOn(URL, 'createObjectURL').mockReturnValue('blob:test'); + vi.spyOn(URL, 'revokeObjectURL').mockImplementation(() => {}); + }); + const appends = (s) => + s.sent.filter((event) => event.type === 'input_audio_buffer.append'); + + it.each([1012, 1006])( + 'keeps capturing when the socket closes (%i) before the conversation notices', + async (code) => { + const c = await connected(); + await c.microphone.start(); + fakeTime(); + const events = track(c); + const port = worklets[0].port; + const first = socket(); + port.onmessage({ data: { pcm: new Int16Array(480) } }); + expect(appends(first)).toHaveLength(1); + + // Closing, but no close event yet: the conversation still reads + // as connected. + first.readyState = 2; + port.onmessage({ data: { pcm: new Int16Array(480).fill(7) } }); + expect(port.postMessage).toHaveBeenCalledTimes(2); + expect(audioTrack.stop).not.toHaveBeenCalled(); + expect(port.close).not.toHaveBeenCalled(); + expect(events.errors()).toEqual([]); + + first.readyState = 1; + first.close(code); + port.onmessage({ data: { pcm: new Int16Array(480) } }); + await answerReconnect(); + expect(c.state).toBe('connected'); + const replacement = socket(); + expect(replacement).not.toBe(first); + expect(appends(replacement)).toHaveLength(0); + + port.onmessage({ data: { pcm: new Int16Array(480) } }); + expect(appends(replacement)).toHaveLength(1); + expect(getUserMedia).toHaveBeenCalledOnce(); + expect(worklets).toHaveLength(1); + expect(events.errors()).toEqual([]); + + c.close(); + expect(audioTrack.stop).toHaveBeenCalledOnce(); + expect(port.close).toHaveBeenCalledOnce(); + }, + ); + + it('releases capture once on a final close', async () => { + const c = await connected(); + await c.microphone.start(); + socket().close(1008, 'invalid_api_key'); + expect(c.state).toBe('closed'); + expect(audioTrack.stop).toHaveBeenCalledOnce(); + expect(worklets[0].port.close).toHaveBeenCalledOnce(); + c.close(); + expect(audioTrack.stop).toHaveBeenCalledOnce(); + }); +}); diff --git a/src/puter-js/src/modules/ai/convoMedia.js b/src/puter-js/src/modules/ai/convoMedia.js index 604674dad..b61ee2fd5 100644 --- a/src/puter-js/src/modules/ai/convoMedia.js +++ b/src/puter-js/src/modules/ai/convoMedia.js @@ -7,6 +7,12 @@ const SAMPLE_RATE = 24000; const MEDIA_TIMEOUT_MS = 10_000; // Replies arrive faster than they play; this bounds what may be scheduled. const MAX_SCHEDULED_SECONDS = 300; +// Transport refusals that cost one microphone chunk, not the capture. +const DROPPED_CHUNK_ERRORS = new Set([ + 'queue_full', + 'session_closed', + 'session_not_ready', +]); /** * @template T @@ -232,8 +238,10 @@ export class ConvoMicrophone { try { this.#convo.appendAudio(data.pcm); } catch (error) { - // A congested socket drops audio instead of ending capture. - if (error?.code === 'queue_full') return; + // A congested socket, or one closing before the conversation + // knows it, drops the chunk instead of ending capture: a + // reconnect resumes it and a final close stops it. + if (DROPPED_CHUNK_ERRORS.has(error?.code)) return; this.stop(); this.#convo.reportMediaError(error); } diff --git a/src/puter-js/src/modules/ai/convoMedia.test.js b/src/puter-js/src/modules/ai/convoMedia.test.js index 4d15e74b5..a4312b141 100644 --- a/src/puter-js/src/modules/ai/convoMedia.test.js +++ b/src/puter-js/src/modules/ai/convoMedia.test.js @@ -161,36 +161,101 @@ it('reports browser playback failures with SDK error codes', async () => { message: 'Could not start playback', }); }); -it('drops chunks under backpressure or while reconnecting, and stops on other errors', async () => { - const c = makeConvo(); +/** A started microphone over fake browser media. */ +const startedMic = async (c = makeConvo()) => { const audioTrack = track(); - vi.stubGlobal('navigator', { - mediaDevices: { - getUserMedia: async () => ({ getTracks: () => [audioTrack] }), - }, - }); + const getUserMedia = vi.fn(async () => ({ + getTracks: () => [audioTrack], + getAudioTracks: () => [audioTrack], + })); + vi.stubGlobal('navigator', { mediaDevices: { getUserMedia } }); const mic = new ConvoMicrophone(c); helpers.push(mic); await mic.start(); - const port = FakeWorklet.instances[0].port; - const pcm = new Int16Array(480); - c.appendAudio.mockImplementationOnce(() => { - throw { code: 'queue_full', message: 'full' }; - }); - port.onmessage({ data: { pcm } }); + const port = FakeWorklet.instances.at(-1).port; + const context = FakeContext.instances.at(-1); + return { c, mic, port, context, audioTrack, getUserMedia }; +}; +const pcm = new Int16Array(480); + +it('drops a chunk the transport refuses and keeps capturing', async () => { + const { c, port, context, audioTrack } = await startedMic(); + for (const code of ['queue_full', 'session_closed', 'session_not_ready']) { + c.appendAudio.mockImplementationOnce(() => { + throw { code, message: code }; + }); + port.onmessage({ data: { pcm } }); + } + expect(port.postMessage).toHaveBeenCalledTimes(3); expect(audioTrack.stop).not.toHaveBeenCalled(); + expect(port.close).not.toHaveBeenCalled(); + expect(context.close).not.toHaveBeenCalled(); + expect(c.reportMediaError).not.toHaveBeenCalled(); + + // Later chunks go out as usual. + port.onmessage({ data: { pcm } }); + expect(c.appendAudio).toHaveBeenCalledTimes(4); + expect(c.appendAudio).toHaveBeenLastCalledWith(pcm); +}); + +it('drops chunks while reconnecting or muted, without sending them later', async () => { + const { c, mic, port, audioTrack } = await startedMic(); c.state = 'connecting'; port.onmessage({ data: { pcm } }); - expect(c.appendAudio).toHaveBeenCalledOnce(); - expect(port.postMessage).toHaveBeenCalledTimes(2); + mic.mute(); c.state = 'connected'; + port.onmessage({ data: { pcm } }); + expect(c.appendAudio).not.toHaveBeenCalled(); + expect(port.postMessage).toHaveBeenCalledTimes(2); + mic.unmute(); + port.onmessage({ data: { pcm } }); + expect(c.appendAudio).toHaveBeenCalledOnce(); + expect(audioTrack.stop).not.toHaveBeenCalled(); +}); + +it('ignores callbacks from a stopped capture, including one stopped while reconnecting', async () => { + const { c, mic, port, audioTrack } = await startedMic(); + const onmessage = port.onmessage; + c.state = 'connecting'; + mic.stop(); + c.state = 'connected'; + onmessage({ data: { pcm } }); + expect(c.appendAudio).not.toHaveBeenCalled(); + expect(port.postMessage).not.toHaveBeenCalled(); + expect(audioTrack.stop).toHaveBeenCalledOnce(); + expect(c.reportMediaError).not.toHaveBeenCalled(); +}); + +it('releases capture once on a genuine error, and ignores chunks after a final close', async () => { + const { c, mic, port, context, audioTrack } = await startedMic(); c.appendAudio.mockImplementationOnce(() => { - throw { code: 'session_closed', message: 'closed' }; + throw { code: 'invalid_argument', message: 'bad chunk' }; }); port.onmessage({ data: { pcm } }); - expect(audioTrack.stop).toHaveBeenCalled(); - expect(c.reportMediaError).toHaveBeenCalledWith( - expect.objectContaining({ code: 'session_closed' }), + mic.stop(); + expect(audioTrack.stop).toHaveBeenCalledOnce(); + expect(port.close).toHaveBeenCalledOnce(); + expect(context.close).toHaveBeenCalledOnce(); + expect(c.reportMediaError).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ code: 'invalid_argument' }), + ); + + const next = await startedMic(); + const onmessage = next.port.onmessage; + next.c.state = 'closed'; + onmessage({ data: { pcm } }); + expect(next.c.appendAudio).not.toHaveBeenCalled(); + expect(next.c.reportMediaError).not.toHaveBeenCalled(); +}); + +it('releases capture once when the processor fails', async () => { + const { c, port, context, audioTrack } = await startedMic(); + FakeWorklet.instances.at(-1).onprocessorerror(); + expect(audioTrack.stop).toHaveBeenCalledOnce(); + expect(port.close).toHaveBeenCalledOnce(); + expect(context.close).toHaveBeenCalledOnce(); + expect(c.reportMediaError).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ code: 'media_error' }), ); }); it('captures at the device rate', async () => { diff --git a/src/puter-js/tests/api/realtime/convo.test.ts b/src/puter-js/tests/api/realtime/convo.test.ts index 041bde60c..c805d8d11 100644 --- a/src/puter-js/tests/api/realtime/convo.test.ts +++ b/src/puter-js/tests/api/realtime/convo.test.ts @@ -613,6 +613,153 @@ describe('convo built SDK and browser media', () => { } }); + it('keeps real microphone capture through a socket that closes before the conversation notices', async () => { + browser ??= await chromium.launch({ + args: [ + '--use-fake-ui-for-media-stream', + '--use-fake-device-for-media-stream', + '--autoplay-policy=no-user-gesture-required', + '--disable-features=LocalNetworkAccessChecks,PrivateNetworkAccessChecks,BlockInsecurePrivateNetworkRequests', + ], + }); + const page = await browser.newPage(); + try { + await page.route('**/__convo_mic_test__', (route) => + route.fulfill({ + contentType: 'text/html', + body: '