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 <noreply@anthropic.com>
This commit is contained in:
404oopsandClaude Opus 5.5 committed 2026-11-05 14:06:43 +01:00
1 parent e47d7e801e
commit 8ffe0320d6
5 files changed
+98 -23

No files matched your search

+24 -4
View File
@@ -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(
+37
View File
@@ -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(() => {
@@ -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);
@@ -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)));
+19 -17
View File
@@ -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' &&