From 28bc4817339ec20dbfd96c69359b25ece7265efb Mon Sep 17 00:00:00 2001 From: 404oops Date: Sat, 31 Oct 2026 15:05:03 +0100 Subject: [PATCH] fix(puter-js): keep microphone capture when the socket closes first A conversation still reads as connected for a moment after its socket starts closing. A microphone chunk sent in that moment got `session_closed`, and the capture handler treated that like a real media failure: it stopped the tracks and the AudioContext, so the replacement session after the reconnect had no microphone. The built-in microphone now drops a chunk the transport refuses with `session_closed` or `session_not_ready`, as it already did for `queue_full`, and keeps capturing. A reconnect resumes sending; a final close still stops capture through the conversation, and processor or other errors still release it. Dropped chunks aren't replayed, and public sends keep reporting the transport error. Co-Authored-By: Claude Opus 5.5 --- src/puter-js/src/modules/ai/convo.test.js | 101 ++++++++++++ src/puter-js/src/modules/ai/convoMedia.js | 12 +- .../src/modules/ai/convoMedia.test.js | 103 +++++++++--- src/puter-js/tests/api/realtime/convo.test.ts | 147 ++++++++++++++++++ 4 files changed, 342 insertions(+), 21 deletions(-) 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: 'Microphone reconnect test', + }), + ); + await page.goto(`${env.apiOrigin}/__convo_mic_test__`); + // Through the proxy, so the test can drop the connection. + await page.evaluate( + ({ origin, apiOrigin }) => { + Object.assign(globalThis, { + PUTER_ORIGIN: origin, + PUTER_API_ORIGIN: apiOrigin, + }); + }, + { origin: env.origin, apiOrigin: proxyOrigin }, + ); + await page.addScriptTag({ + content: readFileSync( + new URL('../../../dist/puter.js', import.meta.url), + 'utf8', + ), + }); + await page.evaluate(async (token) => { + const g = globalThis as typeof globalThis & { + puter: PuterSDK; + test: Record; + }; + g.puter.setAuthToken(token); + const sockets: WebSocket[] = []; + // `globalThis`: this file's own `WebSocket` is the `ws` import. + const Native = globalThis.WebSocket; + g.WebSocket = class extends Native { + constructor(...args: ConstructorParameters) { + super(...args); + sockets.push(this); + } + }; + const tracks: MediaStreamTrack[] = []; + let mediaCalls = 0; + const capture = navigator.mediaDevices.getUserMedia.bind( + navigator.mediaDevices, + ); + navigator.mediaDevices.getUserMedia = async (constraints) => { + mediaCalls++; + const stream = await capture(constraints); + tracks.push(...stream.getTracks()); + return stream; + }; + const c = await g.puter.ai.convo({ output: ['text'] }); + const events: string[] = []; + c.on('*', (event) => events.push(event.type)); + await c.connect(); + await c.microphone.start(); + g.test = { + c, + events, + tracks, + sockets, + mediaCalls: () => mediaCalls, + }; + }, env.users.user.token); + const before = connections.at(-1)!; + const appendsTo = (connection: (typeof connections)[number]) => + connection.received.filter( + (event) => event.type === 'input_audio_buffer.append', + ).length; + await vi.waitFor(() => + expect(appendsTo(before)).toBeGreaterThan(0), + ); + + // The socket reads as closing while the conversation still reads + // connected, for long enough that several chunks arrive. + const gap = await page.evaluate(async () => { + const t = (globalThis as unknown as { test: any }).test; + const socket = t.sockets.at(-1) as WebSocket; + Object.defineProperty(socket, 'readyState', { + configurable: true, + get: () => 2, + }); + await new Promise((resolve) => setTimeout(resolve, 300)); + delete (socket as { readyState?: number }).readyState; + return { + state: t.c.state, + live: t.tracks.every( + (track: MediaStreamTrack) => + track.readyState === 'live', + ), + events: t.events, + }; + }); + expect(gap.state).toBe('connected'); + expect(gap.live).toBe(true); + expect(gap.events).not.toContain('error'); + + for (const socket of proxied) socket.destroy(); + await vi.waitFor( + () => { + const replacement = connections.at(-1)!; + expect(replacement).not.toBe(before); + expect(appendsTo(replacement)).toBeGreaterThan(0); + }, + { timeout: 15_000 }, + ); + const after = await page.evaluate(() => { + const t = (globalThis as unknown as { test: any }).test; + const live = t.tracks.every( + (track: MediaStreamTrack) => track.readyState === 'live', + ); + const state = t.c.state; + t.c.close(); + return { + live, + state, + mediaCalls: t.mediaCalls(), + ended: t.tracks.every( + (track: MediaStreamTrack) => + track.readyState === 'ended', + ), + reconnected: t.events.includes('reconnecting'), + errors: t.events.filter((type: string) => type === 'error'), + }; + }); + expect(after).toEqual({ + live: true, + state: 'connected', + mediaCalls: 1, + ended: true, + reconnected: true, + errors: [], + }); + } finally { + await page.close(); + } + }, 45_000); + // Last: the relay refuses new sessions from here on. it('keeps the session working while a draining server refuses its replacement', async () => { const draining = loadNodePuter(