From 8ffe0320d6a7b89d13be766bd18f1e561a2e59ef Mon Sep 17 00:00:00 2001 From: 404oops Date: Sat, 31 Oct 2026 15:51:51 +0100 Subject: [PATCH] fix(puter-js): end the conversation on an oversized event mid-reconnect A server event over the 4 MiB limit closes the socket with 4009, which ends a connected conversation. During a reconnect attempt the same close was reported as `session_closed`, which is retried, so the conversation tried ten more times and finally closed with the original drop code. A replacement session opened after a restart notice was reopened again and again for the same reason. A 4009 close now ends the conversation during a reconnect too, reporting 4009, and a replacement session that hits it isn't reopened. Tests cover both, a multibyte event of exactly 4 MiB, and a whole retrieved item through the relay. Co-Authored-By: Claude Opus 5.5 --- src/puter-js/src/modules/ai/convo.js | 28 ++++++++++++-- src/puter-js/src/modules/ai/convo.test.js | 37 +++++++++++++++++++ src/puter-js/src/modules/ai/convoTransport.js | 9 ++++- .../src/modules/ai/convoTransport.test.js | 11 ++++++ src/puter-js/tests/api/realtime/convo.test.ts | 36 +++++++++--------- 5 files changed, 98 insertions(+), 23 deletions(-) diff --git a/src/puter-js/src/modules/ai/convo.js b/src/puter-js/src/modules/ai/convo.js index d6fe10db7..d1cf69347 100644 --- a/src/puter-js/src/modules/ai/convo.js +++ b/src/puter-js/src/modules/ai/convo.js @@ -1,6 +1,10 @@ import { fetchUrl } from '../../lib/networkUtils.js'; import { promptIfUpgradeRequired } from '../../lib/upgradePrompt.js'; -import { RealtimeSession, createRealtimeSocket } from './convoTransport.js'; +import { + CLOSE_MESSAGE_TOO_LARGE, + RealtimeSession, + createRealtimeSocket, +} from './convoTransport.js'; import { RealtimeEvents } from './realtimeEvents.js'; import { ConvoMicrophone, ConvoSpeaker } from './convoMedia.js'; import { ConvoHistory } from './convoHistory.js'; @@ -45,6 +49,17 @@ const SESSION_TIMEOUT_MS = 20_000; // Close codes after which a dropped connection is replaced: going away, // abnormal closure, server error, service restart, try again later. const RECONNECT_CODES = new Set([1001, 1006, 1011, 1012, 1013]); +/** + * Why a replacement session ended: one that hit the size limit is refused for + * good, anything else can be retried. + * + * @param {{ type: string; code?: number }} event + * @returns {string} + */ +const closeFailure = (event) => + event.type === 'close' && event.code === CLOSE_MESSAGE_TOO_LARGE + ? 'message_too_large' + : 'session_closed'; // Failures while reconnecting that are worth another attempt. const RETRYABLE_ERRORS = new Set([ 'session_closed', @@ -779,7 +794,7 @@ export class Conversation extends RealtimeEvents { code: String(event.error?.code ?? 'upstream_error'), }); else if (event.type === 'close') - done({ code: 'session_closed' }); + done({ code: closeFailure(event) }); }); handover.stopProbe = () => done({}); }); @@ -799,7 +814,7 @@ export class Conversation extends RealtimeEvents { handover.probe === session && ['close', 'error', RECONNECT_NOTICE].includes(event.type) ) - this.#replacementFailed(handover, 'session_closed'); + this.#replacementFailed(handover, closeFailure(event)); }); this.#scheduleHandover(); } @@ -1280,7 +1295,12 @@ export class Conversation extends RealtimeEvents { code: event.code, reason: event.reason, }; - if (this.#state === 'connecting') { + // An event over the size limit would only arrive again, so it ends + // the conversation even mid-reconnect. + if ( + this.#state === 'connecting' && + event.code !== CLOSE_MESSAGE_TOO_LARGE + ) { // `connect()` or the reconnect loop handles the failure. this.#rejectPending( failure( diff --git a/src/puter-js/src/modules/ai/convo.test.js b/src/puter-js/src/modules/ai/convo.test.js index 3ec6cb91a..a260bf387 100644 --- a/src/puter-js/src/modules/ai/convo.test.js +++ b/src/puter-js/src/modules/ai/convo.test.js @@ -1289,6 +1289,23 @@ describe('restart notice', () => { expect(c.state).toBe('connected'); }); + it('stops opening replacements after one hits the size limit', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver(notice); + await vi.advanceTimersByTimeAsync(SPREAD); + socket().deliver({ + type: 'session.created', + session: { instructions: 'x'.repeat(4 * 1024 * 1024) }, + }); + const sockets = FakeWebSocket.instances.length; + await vi.advanceTimersByTimeAsync(60_000); + expect(FakeWebSocket.instances).toHaveLength(sockets); + expect(c.state).toBe('connected'); + expect(first.readyState).toBe(1); + }); + it('drops a held replacement that gets its own notice or an error', async () => { const c = await connected(); fakeTime(); @@ -2020,6 +2037,26 @@ describe('reconnect retries', () => { expect(FakeWebSocket.instances.length).toBe(attempts); }); + it('ends on an event over the size limit during a reconnect, without more attempts', async () => { + const c = await connected(); + fakeTime(); + const closed = vi.fn(); + c.on('close', closed); + socket().close(1006); + await vi.advanceTimersByTimeAsync(500); + const attempts = FakeWebSocket.instances.length; + socket().deliver({ + type: 'session.created', + session: { instructions: 'x'.repeat(4 * 1024 * 1024) }, + }); + await vi.advanceTimersByTimeAsync(60_000); + expect(c.state).toBe('closed'); + expect(closed).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ type: 'close', code: 4009 }), + ); + expect(FakeWebSocket.instances.length).toBe(attempts); + }); + describe('while the browser is offline', () => { let target; beforeEach(() => { diff --git a/src/puter-js/src/modules/ai/convoTransport.js b/src/puter-js/src/modules/ai/convoTransport.js index 7221ecebf..db9448574 100644 --- a/src/puter-js/src/modules/ai/convoTransport.js +++ b/src/puter-js/src/modules/ai/convoTransport.js @@ -7,6 +7,12 @@ const encoder = new TextEncoder(); // The relay's cap on provider messages, which can carry whole items. What the // SDK sends is capped separately, and much lower. const MAX_SERVER_MESSAGE_BYTES = 4 * 1024 * 1024; +/** + * Close code for a server event over the limit; browsers only accept 1000 and + * 3000-4999 from the client. + * @internal + */ +export const CLOSE_MESSAGE_TOO_LARGE = 4009; /** * Whether `data` encodes to more than `limit` UTF-8 bytes. A UTF-16 code unit @@ -39,8 +45,7 @@ export class RealtimeSession extends RealtimeEvents { const data = typeof msg.data === 'string' ? msg.data : String(msg.data); if (exceedsBytes(data, MAX_SERVER_MESSAGE_BYTES)) { - // Browsers only accept 1000 and 3000-4999 here. - socket.close(4009, 'message too large'); + socket.close(CLOSE_MESSAGE_TOO_LARGE, 'message too large'); return; } event = JSON.parse(data); diff --git a/src/puter-js/src/modules/ai/convoTransport.test.js b/src/puter-js/src/modules/ai/convoTransport.test.js index 58527fd5f..eb182f03a 100644 --- a/src/puter-js/src/modules/ai/convoTransport.test.js +++ b/src/puter-js/src/modules/ai/convoTransport.test.js @@ -90,6 +90,17 @@ describe('RealtimeSession', () => { expect(socket.close).toHaveBeenCalledExactlyOnceWith(4009, 'message too large'); }); + it('accepts a multibyte event of exactly 4 MiB', () => { + const socket = fakeSocket(); + const { typed } = listen(new RealtimeSession(/** @type {any} */ (socket))); + const data = eventOfBytes(LIMIT, 'é'); + expect(new TextEncoder().encode(data).byteLength).toBe(LIMIT); + expect(data.length * 3).toBeGreaterThan(LIMIT); + socket.deliverRaw(data); + expect(typed).toHaveBeenCalledOnce(); + expect(socket.close).not.toHaveBeenCalled(); + }); + it('counts encoded bytes, not characters', () => { const socket = fakeSocket(); const { typed } = listen(new RealtimeSession(/** @type {any} */ (socket))); diff --git a/src/puter-js/tests/api/realtime/convo.test.ts b/src/puter-js/tests/api/realtime/convo.test.ts index d9c1dab31..041bde60c 100644 --- a/src/puter-js/tests/api/realtime/convo.test.ts +++ b/src/puter-js/tests/api/realtime/convo.test.ts @@ -139,22 +139,21 @@ describe('convo built SDK and browser media', () => { type: 'conversation.item.added', item: { ...item, status: 'in_progress', content: [] }, }); - send({ - type: 'conversation.item.done', - item: { - ...item, - status: 'completed', - content: [ - { - type: 'output_audio', - audio: Buffer.alloc(1536 * 1024).toString( - 'base64', - ), - transcript: 'Hello there', - }, - ], - }, - }); + const done = { + ...item, + status: 'completed', + content: [ + { + type: 'output_audio', + audio: Buffer.alloc(1536 * 1024).toString( + 'base64', + ), + transcript: 'Hello there', + }, + ], + }; + send({ type: 'conversation.item.done', item: done }); + send({ type: 'conversation.item.retrieved', item: done }); } if (event.type === 'conversation.item.truncate') socket.send( @@ -508,7 +507,10 @@ describe('convo built SDK and browser media', () => { try { await conversation.connect(); conversation.send({ type: 'test.large' }); - await log.until(() => log.count('conversation.item.done') === 1); + await log.until( + () => log.count('conversation.item.retrieved') === 1, + ); + expect(log.count('conversation.item.done')).toBe(1); const update = log.events.find( (event) => event.type === 'session.updated' &&