From 15485a6bad4b3cf30b85df8a3e571868f0c4bf1f Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Thu, 29 Oct 2026 20:54:39 -0700 Subject: [PATCH] fix(ai): upstream message size, null usage, team 2FA and reconnect settings for realtime Relay: - Upstream messages may be up to 4 MiB; client messages stay at 64 KiB. An OpenAI event above 64 KiB (an echoed item near the client limit, a long reply, a retrieved audio item) closed the upstream before in-flight responses reported usage. - An upstream that closes with a response still in flight raises a warning alarm. - A null usage counts as none instead of closing the session. - Seats whose team requires 2FA need it for realtime too, at connect and on the credential recheck. SDK (puter.ai.convo): - A reconnect reapplies every acknowledged setting, native session.update events included, split to fit the message limit. - connect() during a reconnect settles with it. - Reconnect delays are jittered, and closing cancels a pending delay. The credential-recheck and browser worklet tests now fail when the behavior they cover breaks. --- .../OpenAIRealtimeRelayService.test.ts | 113 +++++++++++++- .../realtime/OpenAIRealtimeRelayService.ts | 49 ++++-- src/backend/services/realtime/README.md | 3 +- src/docs/src/AI/convo.md | 4 +- src/docs/src/AI/realtime.md | 2 +- src/docs/src/rate-limits-and-quotas.md | 13 +- src/puter-js/src/modules/ai/convo.js | 147 ++++++++++++++++-- src/puter-js/src/modules/ai/convo.test.js | 132 ++++++++++++++++ src/puter-js/tests/api/realtime/convo.test.ts | 2 + 9 files changed, 429 insertions(+), 36 deletions(-) diff --git a/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts b/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts index 9c3865ca5..2ec05f0d1 100644 --- a/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts +++ b/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts @@ -14,6 +14,7 @@ import WebSocket, { WebSocketServer } from 'ws'; import { withAiCostFactorHook } from '../../drivers/integrationTestUtil'; import type { IConfig } from '../../types'; import type { PuterServer } from '../../server'; +import { installSeatStub } from '../../testFixtures/seatStub'; import { allocateEphemeralPort, createTestUser, @@ -23,6 +24,7 @@ import { import { kv } from '../../util/kvSingleton'; import { MAX_REALTIME_MESSAGE_BYTES, + MAX_UPSTREAM_MESSAGE_BYTES, REALTIME_PATH, type RealtimeUsage, type RealtimeTranscriptionUsage, @@ -269,11 +271,20 @@ describe('OpenAIRealtimeRelayService relay', () => { ws.send(JSON.stringify(envelope)); return; } + if (event.type === 'test.retrieve') { + ws.send( + JSON.stringify({ + type: 'conversation.item.retrieved', + item: { audio: 'A'.repeat(event.bytes) }, + }), + ); + return; + } if (event.type === 'test.oversize') { ws.send( JSON.stringify({ type: 'response.output_text.delta', - delta: 'x'.repeat(MAX_REALTIME_MESSAGE_BYTES), + delta: 'x'.repeat(MAX_UPSTREAM_MESSAGE_BYTES), }), ); return; @@ -401,7 +412,19 @@ describe('OpenAIRealtimeRelayService relay', () => { }, { timeout: 5_000 }, ); - return { ws, events, closed, waitFor }; + const waitForEcho = (type: string) => + vi.waitFor( + () => + expect( + events.some( + (e) => + e.type === 'echo' && + (e.event as { type: string }).type === type, + ), + ).toBe(true), + { timeout: 5_000 }, + ); + return { ws, events, closed, waitFor, waitForEcho }; }; it('rejects a missing credential with an error event and 1008', async () => { @@ -510,9 +533,22 @@ describe('OpenAIRealtimeRelayService relay', () => { expect((await c.closed).code).toBe(1008); }); - it('rejects oversized upstream messages before forwarding them', async () => { + it('forwards upstream messages beyond the client limit', async () => { const c = open(`${relayUrl}?auth_token=${user.token}`); await c.waitFor('session.created'); + c.ws.send(JSON.stringify({ type: 'test.retrieve', bytes: 1024 * 1024 })); + await c.waitFor('conversation.item.retrieved'); + c.ws.send('{"type":"test.alive"}'); + await c.waitForEcho('test.alive'); + c.ws.close(); + }); + + it('closes on an oversized upstream message and alarms on the response it loses', async () => { + const alarm = vi.spyOn(server.clients.alarm, 'create'); + const c = open(`${relayUrl}?auth_token=${user.token}`); + await c.waitFor('session.created'); + c.ws.send(JSON.stringify({ type: 'hold.response' })); + await c.waitFor('response.created'); c.ws.send(JSON.stringify({ type: 'test.oversize' })); expect((await c.closed).code).toBe(1011); expect( @@ -520,12 +556,48 @@ describe('OpenAIRealtimeRelayService relay', () => { (event) => event.type === 'response.output_text.delta', ), ).toBe(false); + await vi.waitFor(() => + expect(alarm).toHaveBeenCalledWith( + 'realtime_unbilled_response', + expect.any(String), + expect.objectContaining({ inFlight: 1 }), + 'warning', + { dedup: true }, + ), + ); + }); + + it('treats a null usage as none and keeps the session', async () => { + const alarm = vi.spyOn(server.clients.alarm, 'create'); + const c = open(`${relayUrl}?auth_token=${user.token}`); + await c.waitFor('session.created'); + c.ws.send(JSON.stringify({ type: 'hold.response' })); + await c.waitFor('response.created'); + c.ws.send( + JSON.stringify({ + type: 'test.upstream', + payload: + '{"type":"response.done","response":{"id":"resp_hold","usage":null}}', + }), + ); + await c.waitFor('response.done'); + c.ws.send('{"type":"test.alive"}'); + await c.waitForEcho('test.alive'); + c.ws.close(); + await c.closed; + expect(alarm).not.toHaveBeenCalledWith( + 'realtime_unbilled_response', + expect.anything(), + expect.anything(), + expect.anything(), + expect.anything(), + ); }); it.each([ 'null', '{"type":"response.done","response":{"usage":{"input_tokens":-1}}}', - '{"type":"response.done","response":{"usage":null}}', + '{"type":"response.done","response":{"usage":[]}}', '{"type":"response.done","response":{"usage":{"input_token_details":null}}}', ])('rejects malformed upstream shape %s', async (payload) => { const c = open(`${relayUrl}?auth_token=${user.token}`); @@ -860,18 +932,47 @@ describe('OpenAIRealtimeRelayService relay', () => { it('keeps a session when the re-check lookup itself fails', async () => { const relay = server.services.openaiRealtime; const auth = vi.spyOn(server.services.auth, 'authenticate'); - const { ws, waitFor } = open( + const { ws, events, waitFor, waitForEcho } = open( `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, ); await waitFor('session.created'); auth.mockRejectedValue(new Error('db down')); kv.del(realtimeAuthKey(user.token)); await relay.recheckSessions(); - expect(ws.readyState).toBe(WebSocket.OPEN); + // Frames arrive in order, so a close from the recheck would beat this. + ws.send('{"type":"test.alive"}'); + await waitForEcho('test.alive'); + expect(events.some((e) => e.type === 'error')).toBe(false); auth.mockRestore(); ws.close(); }); + it('refuses a seat whose team requires 2FA the account lacks', async () => { + const relay = server.services.openaiRealtime; + const live = open( + `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, + ); + await live.waitFor('session.created'); + + const row = await server.stores.user.getByUsername(user.username); + installSeatStub(server as never, [row!.id], { require2fa: true }); + + const fresh = open( + `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, + ); + expect(await fresh.waitFor('error')).toMatchObject({ + error: { code: 'invalid_api_key' }, + }); + expect((await fresh.closed).code).toBe(1008); + + kv.del(realtimeAuthKey(user.token)); + await relay.recheckSessions(); + expect(await live.waitFor('error')).toMatchObject({ + error: { code: 'invalid_api_key' }, + }); + expect((await live.closed).code).toBe(1008); + }); + // Last: the relay stays in drain mode afterwards. it('drains: refuses new sessions, finishes in-flight responses, then closes with 1012', async () => { const relay = server.services.openaiRealtime; diff --git a/src/backend/services/realtime/OpenAIRealtimeRelayService.ts b/src/backend/services/realtime/OpenAIRealtimeRelayService.ts index 856838009..e023eb2c8 100644 --- a/src/backend/services/realtime/OpenAIRealtimeRelayService.ts +++ b/src/backend/services/realtime/OpenAIRealtimeRelayService.ts @@ -126,6 +126,9 @@ export interface RealtimeTranscriptionUsage { const OPENAI_KEY_PROTOCOL = 'openai-insecure-api-key.'; export const MAX_REALTIME_MESSAGE_BYTES = 64 * 1024; +// OpenAI events can carry whole items, such as a retrieved audio item or a +// long reply, so upstream messages get far more room than client ones. +export const MAX_UPSTREAM_MESSAGE_BYTES = 4 * 1024 * 1024; const MAX_PENDING_MESSAGES = 32; // Usage batches waiting on metering; past this the session closes, but every // batch already received is still recorded. @@ -160,15 +163,15 @@ const METERING_DRIVER = 'openai-realtime'; interface RealtimeEvent { type: string; session?: unknown; - response?: { id?: unknown; usage?: RealtimeUsage }; - usage?: RealtimeTranscriptionUsage; + response?: { id?: unknown; usage?: RealtimeUsage | null }; + usage?: RealtimeTranscriptionUsage | null; } const isRecord = (value: unknown): value is Record => value !== null && typeof value === 'object' && !Array.isArray(value); const validUsage = (usage: unknown): boolean => { - if (usage === undefined) return true; + if (usage === undefined || usage === null) return true; if (!isRecord(usage)) return false; const validNumbers = (value: Record) => [ @@ -760,7 +763,10 @@ export class OpenAIRealtimeRelayService extends PuterService { inFlightIds.delete(event.response.id); } meter( - realtimeUsageRows(billingModel, event.response?.usage), + realtimeUsageRows( + billingModel, + event.response?.usage ?? undefined, + ), ); if (inFlight === 0) { onIdle?.(); @@ -776,7 +782,7 @@ export class OpenAIRealtimeRelayService extends PuterService { meter( realtimeTranscriptionRows( transcriptionModel, - event.usage, + event.usage ?? undefined, ), ); break; @@ -942,9 +948,10 @@ export class OpenAIRealtimeRelayService extends PuterService { headers: { Authorization: `Bearer ${apiKey}` }, handshakeTimeout: UPSTREAM_HANDSHAKE_MS, perMessageDeflate: false, - maxPayload: MAX_REALTIME_MESSAGE_BYTES, - maxFragments: 64, - maxBufferedChunks: 128, + maxPayload: MAX_UPSTREAM_MESSAGE_BYTES, + // A large message arrives in many TLS records. + maxFragments: 1024, + maxBufferedChunks: 16 * 1024, }, ); const up = upstream; @@ -971,7 +978,7 @@ export class OpenAIRealtimeRelayService extends PuterService { if (closed || client.readyState !== WebSocket.OPEN) return; if ( client.bufferedAmount + buf.length > - HIGH_WATER_BYTES + MAX_REALTIME_MESSAGE_BYTES + HIGH_WATER_BYTES + MAX_UPSTREAM_MESSAGE_BYTES ) { reject('rate_limit_exceeded', 'Client queue is full'); return; @@ -1019,6 +1026,20 @@ export class OpenAIRealtimeRelayService extends PuterService { }); up.on('close', (code, reason) => { if (settleTimer) clearTimeout(settleTimer); + if (inFlight > 0) { + this.clients.alarm.create( + 'realtime_unbilled_response', + 'Realtime upstream closed before a response reported usage', + { + userId: actor?.user?.uuid, + model: billingModel, + inFlight, + code, + }, + 'warning', + { dedup: true }, + ); + } session.close( code === 1005 || code === 1006 ? 1011 : code, reason.toString().slice(0, 120), @@ -1079,7 +1100,15 @@ export class OpenAIRealtimeRelayService extends PuterService { userAgent: typeof ua === 'string' ? ua : undefined, }); if (result.blocked) return null; - const decision = decideSocketAuth(result, { allowAppActors: true }); + const user = result.actor?.user; + const seat = + user && !user.otp_enabled && typeof user.id === 'number' + ? await this.stores.team?.getOrgSeat(user.id) + : null; + const decision = decideSocketAuth(result, { + allowAppActors: true, + teamRequires2fa: Number(seat?.require_2fa) === 1, + }); return 'accept' in decision ? decision.accept : null; } } diff --git a/src/backend/services/realtime/README.md b/src/backend/services/realtime/README.md index d756dac23..3187d7057 100644 --- a/src/backend/services/realtime/README.md +++ b/src/backend/services/realtime/README.md @@ -35,7 +35,8 @@ conversation.sendText('Hello!'); All transport limits are published in [rate-limits-and-quotas.md](../../../docs/src/rate-limits-and-quotas.md). -Message size applies in both directions and includes reassembled fragments. +Message size includes reassembled fragments. Client messages are capped at +64 KiB; upstream messages at 4 MiB, since OpenAI events can carry whole items. Read backpressure is supplemented by hard queue admission limits because a paused socket can still deliver messages already in its receive buffer. diff --git a/src/docs/src/AI/convo.md b/src/docs/src/AI/convo.md index 7ca78946e..64f9f11fb 100644 --- a/src/docs/src/AI/convo.md +++ b/src/docs/src/AI/convo.md @@ -41,7 +41,7 @@ The conversation exposes these properties and methods. | --- | --- | | `state` | `prepared`, `connecting`, `connected`, or `closed`. | | `model` | Resolved model name. | -| `connect()` | Opens the connection and applies settings. Resolves after the provider acknowledges session configuration. Concurrent calls share one attempt. Calling it while connected opens no additional connection. | +| `connect()` | Opens the connection and applies settings. Resolves after the provider acknowledges session configuration. Concurrent calls share one attempt. Calling it while connected opens no additional connection. During a reconnection, it resolves when the reconnection succeeds and rejects if the conversation closes instead. | | `configure(options)` | Updates the given session settings and resolves after the provider acknowledges this update. Settings it isn't given, including ones set through native events, are left alone. Accepts the factory options except `model`. One update can be pending at a time. Provider rules apply to changes such as switching a voice after audio has been generated. | | `sendText(text)` | Adds a user message and requests a response. Text must be nonempty and at most 8192 characters. | | `appendAudio(pcm)` | Appends an `Int16Array` containing PCM16, 24 kHz mono audio. A chunk contains at most 12000 samples. | @@ -54,7 +54,7 @@ The conversation exposes these properties and methods. Attach listeners before connecting to receive initial session events. Closing during connection rejects the pending attempt. -If the connection drops (network loss, a server restart or a provider error), the conversation reconnects on its own: `state` returns to `connecting`, a `reconnecting` event fires, pending operations reject with `connection_lost`, and `ready` fires again with `reconnected: true` once settings are reapplied. Listeners, the microphone and the speaker carry over; microphone audio captured while reconnecting is dropped. The provider starts a fresh session, so earlier conversation items are not restored. After five failed attempts, or a refusal such as `insufficient_funds`, the conversation closes. A closed conversation cannot reconnect; prepare a new one to start another session. +If the connection drops because of network loss, a server restart or a provider error, the conversation reconnects on its own. While it reconnects, `state` is `connecting`, a `reconnecting` event fires, and pending operations reject with `connection_lost`. The new session gets every setting the provider acknowledged, from the factory options, `configure()` and native `session.update` events. Then `ready` fires again with `reconnected: true`. Listeners, the microphone and the speaker carry over. Microphone audio captured while reconnecting is dropped. The provider starts a fresh session, so earlier conversation items are not restored. After five failed attempts, or a refusal such as `insufficient_funds`, the conversation closes. A closed conversation cannot reconnect. To start another session, prepare a new one. Native event definitions are listed in the [OpenAI realtime reference](https://developers.openai.com/api/reference/resources/realtime/server-events). Native events retain their provider fields. Sending native events preserves direct control of the session. Audio helpers expect the session to use PCM16 at 24 kHz. diff --git a/src/docs/src/AI/realtime.md b/src/docs/src/AI/realtime.md index 77d0340f8..f2203b10c 100644 --- a/src/docs/src/AI/realtime.md +++ b/src/docs/src/AI/realtime.md @@ -94,4 +94,4 @@ API errors arrive as `{ "type": "error", "error": { "code": "...", "message": ". Policy failures generally close with 1008, oversized client messages with 1009, upstream or setup failures with 1011, and server restarts with 1012. Authentication and model failures can arrive after the WebSocket handshake succeeds. -The endpoint requires a paid Puter plan. Usage is charged after completed responses and input transcriptions. Each account can open 20 sessions at once by default. Each JSON message is limited to 64 KiB in either direction, including reassembled fragments. Stream audio in small chunks. All message rates, queue caps, timeouts, and other bounds are published in [rate limits and quotas](/rate-limits-and-quotas/). +The endpoint requires a paid Puter plan. Usage is charged after completed responses and input transcriptions. Each account can open 20 sessions at once by default. Each JSON message you send is limited to 64 KiB, including reassembled fragments. Messages from the provider can be up to 4 MiB. Stream audio in small chunks. All message rates, queue caps, timeouts, and other bounds are published in [rate limits and quotas](/rate-limits-and-quotas/). diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index 914201da8..0946738ee 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -95,17 +95,18 @@ Realtime sessions ([`puter.ai.convo()`](/AI/convo/), `/puterai/openai/v1/realtim | Realtime transport limit | Value | Behavior at the limit | | --- | --- | --- | -| JSON message size in either direction | 64 KiB, including reassembled fragments | Oversized client messages close with 1009. Oversized upstream messages close the session with 1011. | -| Fragments per message | 64 | Connection closes. | -| Buffered receive chunks | 128 per socket | Connection closes. | +| Client JSON message size | 64 KiB, including reassembled fragments | Close 1009. | +| Upstream JSON message size | 4 MiB, including reassembled fragments | Close 1011. | +| Fragments per client message | 64 | Connection closes. | +| Buffered receive chunks from the client | 128 | Connection closes. | | Event type length | 1–128 characters | Invalid client envelopes receive `invalid_event` and close with 1008. Invalid upstream envelopes close with 1011. | | Client message rate | 100/s, burst 100 | Error `rate_limit_exceeded`, close 1008. Ping and pong frames count toward this allowance. | | Client upload rate | 128 KiB/s, burst 64 KiB | Reads pause to refill the budget. More than 64 KiB of overdraft closes with `rate_limit_exceeded`. | | Input pending session setup | 32 messages or 64 KiB | Error `rate_limit_exceeded`, close 1008. | -| Forwarding queue in either direction | Reads pause at 128 KiB. Admission stops above 192 KiB. | A full queue closes with `rate_limit_exceeded`. | +| Forwarding queue in either direction | Reads pause at 128 KiB. Admission stops above 192 KiB toward the provider and above 4 MiB plus 128 KiB toward the client. | A full queue closes with `rate_limit_exceeded`. | | Client that stops reading | 30 seconds of paused upstream reads | Close 1008. | | Pending usage records | 32 | Error `rate_limit_exceeded`, close 1008. Records already received are still billed. | -| Numeric usage fields | Finite, nonnegative, at most 1 billion | Invalid upstream usage closes with 1011. | +| Numeric usage fields | Finite, nonnegative, at most 1 billion | Invalid upstream usage closes with 1011. A `null` usage counts as none. | | Session setup timeout | 15 seconds | Error `connection_timeout`, close 1011. | | Upstream handshake timeout | 10 seconds | Close 1011. | | Heartbeat interval | 30 seconds | A missing pong terminates the connection on the next heartbeat. | @@ -133,7 +134,7 @@ The [`puter.ai.convo()`](/AI/convo/) SDK also applies these limits before sendin | Capture chunk duration | 20 ms | AudioWorklet sends 480 PCM16 samples per chunk. | | Unacknowledged capture chunks | 25, or 500 ms | Further chunks are dropped until the page catches up. | | Queued playback | 300 seconds | Further audio for the reply is dropped and `queue_full` is reported once. | -| Reconnection attempts | 5, backing off from 500 ms to 8 seconds | The conversation closes with the original close code. | +| Reconnection attempts | 5, backing off from 500 ms to 8 seconds, each delay randomized between half and one and a half times its value | The conversation closes with the original close code. | | Remembered canceled responses | 32 | Older response IDs are evicted from the suppression set. | diff --git a/src/puter-js/src/modules/ai/convo.js b/src/puter-js/src/modules/ai/convo.js index ea5d35d72..9d64e28b7 100644 --- a/src/puter-js/src/modules/ai/convo.js +++ b/src/puter-js/src/modules/ai/convo.js @@ -47,6 +47,33 @@ const encoder = new TextEncoder(); */ const failure = (code, message) => ({ code, message }); +/** + * @param {unknown} value + * @returns {value is Record} + */ +const isPlainObject = (value) => + value !== null && typeof value === 'object' && !Array.isArray(value); + +/** + * Applies a partial `session.update` payload. Nested objects merge, except + * typed ones such as `turn_detection`, which replace like any other value. + * + * @param {Record} base + * @param {Record} patch + * @returns {Record} + */ +function mergeSession(base, patch) { + const merged = { ...base }; + for (const [key, value] of Object.entries(patch)) { + const current = merged[key]; + merged[key] = + isPlainObject(value) && !('type' in value) && isPlainObject(current) + ? mergeSession(current, value) + : value; + } + return merged; +} + /** * @param {ConvoOptions} options * @param {boolean} [allowModel] @@ -206,8 +233,29 @@ export class Conversation extends RealtimeEvents { // `session.update` event IDs awaiting `session.updated`, in send order. /** @type {string[]} */ #sessionUpdates = []; + /** @type {Map>} */ + #sentSessions = new Map(); + // Every acknowledged session setting, SDK and native, reapplied after a + // reconnect. + /** @type {Record} */ + #sessionState = {}; /** @type {string | undefined} */ #acknowledgedUpdate; + /** + * @type {{ + * timer: ReturnType; + * resolve: (value?: unknown) => void; + * } | null} + */ + #reconnectDelay = null; + #reconnecting = false; + /** + * @type {{ + * resolve: () => void; + * reject: (error: ConvoError) => void; + * }[]} + */ + #reconnectWaiters = []; // `response.create` event IDs whose response hasn't been created yet. /** @type {Set} */ #requestedResponses = new Set(); @@ -254,7 +302,8 @@ export class Conversation extends RealtimeEvents { } /** - * Opens and configures the session. Concurrent calls share one attempt. + * Opens and configures the session. Concurrent calls share one attempt; + * during a reconnection, resolves once it succeeds. * * @returns {Promise} */ @@ -263,6 +312,10 @@ export class Conversation extends RealtimeEvents { return Promise.reject( failure('session_closed', 'Conversation is closed'), ); + if (this.#reconnecting) + return new Promise((resolve, reject) => + this.#reconnectWaiters.push({ resolve, reject }), + ); if (this.#connection) return this.#connection; if (!this.#puter.authToken) return Promise.reject( @@ -295,9 +348,11 @@ export class Conversation extends RealtimeEvents { /** * Opens a socket, waits for the provider session and applies settings. * + * @param {boolean} [reapply] Reapply every acknowledged setting instead of + * the SDK options alone. * @returns {Promise} */ - async #open() { + async #open(reapply = false) { let socket; try { socket = createRealtimeSocket(this.#puter, this.model); @@ -312,11 +367,47 @@ export class Conversation extends RealtimeEvents { this.#receive(event), ); await this.#wait('session.created', undefined, undefined, true); - await this.#configure(this.#options, true); + if (reapply && Object.keys(this.#sessionState).length) + await this.#reapplySession(); + else await this.#configure(this.#options, true); if (this.#state === 'closed') throw failure('session_closed', 'Conversation is closed'); } + /** + * Sends the acknowledged session state, split by top-level setting so each + * update stays under the message limit. + * + * @returns {Promise} + */ + async #reapplySession() { + /** @type {Record[]} */ + const updates = []; + /** @type {Record} */ + let session = { type: 'realtime' }; + for (const [key, value] of Object.entries(this.#sessionState)) { + if (key === 'type') continue; + const next = { ...session, [key]: value }; + if ( + Object.keys(session).length > 1 && + encoder.encode(JSON.stringify(next)).byteLength > + MAX_MESSAGE_BYTES - 256 + ) { + updates.push(session); + session = { type: 'realtime', [key]: value }; + } else session = next; + } + updates.push(session); + for (const update of updates) { + const eventId = this.#eventId(); + await this.#request( + 'session.updated', + { type: 'session.update', event_id: eventId, session: update }, + () => this.#acknowledgedUpdate === eventId, + ); + } + } + /** * Replaces a dropped connection, keeping settings, listeners and media. The * provider session restarts, so earlier conversation items are gone. @@ -326,6 +417,7 @@ export class Conversation extends RealtimeEvents { */ async #reconnect(event) { this.#state = 'connecting'; + this.#reconnecting = true; this.#detach(); this.#resetTurnState(); this.#rejectPending( @@ -337,13 +429,22 @@ export class Conversation extends RealtimeEvents { reason: event.reason, }); for (let attempt = 0; attempt < RECONNECT_ATTEMPTS; attempt++) { - await new Promise((resolve) => - setTimeout(resolve, RECONNECT_DELAY_MS * 2 ** attempt), - ); + // Jittered so clients dropped together don't return together. + const delay = RECONNECT_DELAY_MS * 2 ** attempt; + await new Promise((resolve) => { + this.#reconnectDelay = { + timer: setTimeout(resolve, delay * (0.5 + Math.random())), + resolve, + }; + }); + this.#reconnectDelay = null; if (this.#state === 'closed') return; try { - await this.#open(); + await this.#open(true); this.#state = 'connected'; + this.#reconnecting = false; + for (const waiter of this.#reconnectWaiters.splice(0)) + waiter.resolve(); this.emitRealtime({ type: 'ready', model: this.model, @@ -383,6 +484,7 @@ export class Conversation extends RealtimeEvents { #resetTurnState() { this.#activeResponse = ''; this.#sessionUpdates = []; + this.#sentSessions.clear(); this.#requestedResponses.clear(); this.#interruption = null; this.#cancelled.clear(); @@ -699,11 +801,23 @@ export class Conversation extends RealtimeEvents { #finish(event) { if (this.#state === 'closed') return; this.#state = 'closed'; + this.#reconnecting = false; + if (this.#reconnectDelay) { + clearTimeout(this.#reconnectDelay.timer); + this.#reconnectDelay.resolve(); + } this.microphone.stop(); void this.speaker.stop().catch(() => {}); this.#rejectPending( failure('session_closed', 'Conversation is closed'), ); + for (const waiter of this.#reconnectWaiters.splice(0)) + waiter.reject( + failure( + 'session_closed', + event.reason || 'Conversation is closed', + ), + ); this.#detach(); this.#cancelled.clear(); this.emitRealtime(event); @@ -778,8 +892,15 @@ export class Conversation extends RealtimeEvents { const [requested] = this.#requestedResponses; if (requested) this.#requestedResponses.delete(requested); } - if (event.type === 'session.updated') - this.#acknowledgedUpdate = this.#sessionUpdates.shift(); + if (event.type === 'session.updated') { + const id = this.#sessionUpdates.shift(); + this.#acknowledgedUpdate = id; + const applied = id && this.#sentSessions.get(id); + if (applied) { + this.#sentSessions.delete(id); + this.#sessionState = mergeSession(this.#sessionState, applied); + } + } if ( event.type === 'session.created' || event.type === 'session.updated' @@ -794,6 +915,7 @@ export class Conversation extends RealtimeEvents { this.#requestedResponses.delete(failed); const index = this.#sessionUpdates.indexOf(failed); if (index >= 0) this.#sessionUpdates.splice(index, 1); + this.#sentSessions.delete(failed); } this.#promptUpgrade(event.error); } @@ -951,8 +1073,13 @@ export class Conversation extends RealtimeEvents { throw failure('queue_full', 'Realtime send queue is full'); for (const [i, { json }] of messages.entries()) { const event = events[i]; - if (event.type === 'session.update') + if (event.type === 'session.update') { this.#sessionUpdates.push(event.event_id); + // A copy, so later changes to the caller's object don't leak in. + const { session } = JSON.parse(json); + if (isPlainObject(session)) + this.#sentSessions.set(event.event_id, session); + } if (event.type === 'response.create') this.#requestedResponses.add(event.event_id); socket.send(json); diff --git a/src/puter-js/src/modules/ai/convo.test.js b/src/puter-js/src/modules/ai/convo.test.js index 73dfc74c9..5a3d77661 100644 --- a/src/puter-js/src/modules/ai/convo.test.js +++ b/src/puter-js/src/modules/ai/convo.test.js @@ -181,6 +181,7 @@ describe('Conversation lifecycle', () => { it('reconnects after a dropped connection and reapplies settings', async () => { const c = await connected({ instructions: 'Be brief.' }); vi.useFakeTimers(); + vi.spyOn(Math, 'random').mockReturnValue(0.5); const events = []; c.on('*', (event) => events.push(event.type)); const update = c.configure({ voice: 'marin' }); @@ -207,6 +208,7 @@ describe('Conversation lifecycle', () => { it('gives up after repeated failures or a refusal', async () => { const c = await connected(); vi.useFakeTimers(); + vi.spyOn(Math, 'random').mockReturnValue(0.5); const closed = vi.fn(); c.on('close', closed); socket().close(1012); @@ -237,6 +239,136 @@ describe('Conversation lifecycle', () => { }); }); +describe('reconnection', () => { + /** Drops the connection and answers the replacement's handshake. */ + const reconnect = async () => { + socket().close(1012); + await vi.advanceTimersByTimeAsync(500); + socket().deliver({ type: 'session.created', session: {} }); + await vi.advanceTimersByTimeAsync(0); + }; + + it('reapplies acknowledged native settings after the SDK ones', async () => { + const c = await connected({ instructions: 'Be brief.' }); + vi.useFakeTimers(); + vi.spyOn(Math, 'random').mockReturnValue(0.5); + c.send({ + type: 'session.update', + session: { + type: 'realtime', + audio: { input: { turn_detection: null } }, + tools: [{ type: 'function', name: 'lookup' }], + }, + }); + socket().deliver({ type: 'session.updated', session: {} }); + // Refused by the provider, so never reapplied. + c.send({ + type: 'session.update', + event_id: 'bad', + session: { type: 'realtime', tool_choice: 'bogus' }, + }); + socket().deliver({ + type: 'error', + error: { code: 'invalid_value', message: 'No', event_id: 'bad' }, + }); + await reconnect(); + const [update] = socket().sent; + expect(update.type).toBe('session.update'); + expect(update.session).toMatchObject({ + type: 'realtime', + instructions: 'Be brief.', + tools: [{ type: 'function', name: 'lookup' }], + audio: { + input: { + format: { type: 'audio/pcm', rate: 24000 }, + turn_detection: null, + }, + }, + }); + expect(update.session).not.toHaveProperty('tool_choice'); + socket().deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connected'); + }); + + it('splits reapplied settings that exceed one message', async () => { + const c = await connected(); + vi.useFakeTimers(); + vi.spyOn(Math, 'random').mockReturnValue(0.5); + c.send({ + type: 'session.update', + session: { type: 'realtime', instructions: 'x'.repeat(40_000) }, + }); + socket().deliver({ type: 'session.updated', session: {} }); + c.send({ + type: 'session.update', + session: { + type: 'realtime', + tools: [{ type: 'function', description: 'y'.repeat(40_000) }], + }, + }); + socket().deliver({ type: 'session.updated', session: {} }); + await reconnect(); + expect(socket().sent).toHaveLength(1); + expect(socket().sent[0].session.instructions).toHaveLength(40_000); + socket().deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(socket().sent).toHaveLength(2); + expect(socket().sent[1].session.tools).toHaveLength(1); + socket().deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connected'); + }); + + it('settles connect() with the reconnection', async () => { + const c = await connected(); + vi.useFakeTimers(); + vi.spyOn(Math, 'random').mockReturnValue(0.5); + socket().close(1006); + const resumed = vi.fn(); + const ready = c.connect().then(resumed); + await vi.advanceTimersByTimeAsync(500); + socket().deliver({ type: 'session.created', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(resumed).not.toHaveBeenCalled(); + socket().deliver({ type: 'session.updated', session: {} }); + await ready; + expect(c.state).toBe('connected'); + + socket().close(1006); + const failed = c.connect(); + await vi.advanceTimersByTimeAsync(500); + socket().deliver({ + type: 'error', + error: { code: 'insufficient_funds', message: 'No usage left' }, + }); + await expect(failed).rejects.toMatchObject({ + code: 'session_closed', + message: 'No usage left', + }); + }); + + it('jitters the delay and stops waiting once closed', async () => { + const c = await connected(); + vi.useFakeTimers(); + vi.spyOn(Math, 'random').mockReturnValue(0); + const first = socket(); + first.close(1012); + await vi.advanceTimersByTimeAsync(249); + expect(socket()).toBe(first); + await vi.advanceTimersByTimeAsync(1); + expect(socket()).not.toBe(first); + + const d = await connected(); + d.on('close', () => {}); + socket().close(1012); + expect(vi.getTimerCount()).toBeGreaterThan(0); + d.close(); + c.close(); + expect(vi.getTimerCount()).toBe(0); + }); +}); + describe('conversation helpers and limits', () => { it('requires readiness before sending', () => { const c = make(); diff --git a/src/puter-js/tests/api/realtime/convo.test.ts b/src/puter-js/tests/api/realtime/convo.test.ts index 1327195a3..22ea591a0 100644 --- a/src/puter-js/tests/api/realtime/convo.test.ts +++ b/src/puter-js/tests/api/realtime/convo.test.ts @@ -138,6 +138,8 @@ describe('convo built SDK and browser media', () => { }); it('captures through a real AudioWorklet and releases browser media on close', async () => { + // The first test appends a chunk of the same size. + received.length = 0; browser = await chromium.launch({ args: [ '--use-fake-ui-for-media-stream',