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 <noreply@anthropic.com>

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

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

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

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: '<!doctype html><title>Microphone reconnect test</title>',
+                }),
+            );
+            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<string, unknown>;
+                };
+                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<typeof Native>) {
+                        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(