From 07784d894ce841a0857b345cd9ce0bd0e8ce1db0 Mon Sep 17 00:00:00 2001 From: 404oops Date: Sun, 1 Nov 2026 19:07:04 +0100 Subject: [PATCH] fix(convo): bound recovery attempts and report omitted context --- src/docs/src/AI/convo.md | 6 +- src/docs/src/rate-limits-and-quotas.md | 8 +- src/puter-js/src/modules/ai/convo.js | 107 +++++++-- src/puter-js/src/modules/ai/convo.test.js | 218 +++++++++++++++++- src/puter-js/src/modules/ai/convoHistory.js | 11 +- .../src/modules/ai/convoHistory.test.js | 27 ++- src/puter-js/src/modules/ai/convoTransport.js | 6 +- src/puter-js/tests/api/realtime/convo.test.ts | 22 ++ src/puter-js/tests/api/suites/ai.suite.ts | 15 ++ 9 files changed, 387 insertions(+), 33 deletions(-) diff --git a/src/docs/src/AI/convo.md b/src/docs/src/AI/convo.md index 618fc199a..bf199f73e 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. During a reconnection, it resolves when the reconnection succeeds and rejects if the conversation closes instead. | +| `connect()` | Opens the connection and applies settings. Resolves after the provider acknowledges session configuration. Concurrent calls share one connection Promise, including bounded retries for transient setup failures. 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,9 +54,9 @@ 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 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. +If the connection drops because of network loss, a server restart or a provider error, the conversation reconnects on its own. Normal closes with code 1000 end the conversation. 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 conversation also restores recent context. Before `ready` fires, it adds back up to 40 of the most recent conversation items, 48 KiB in total. Restored items are user text, user speech that has a transcript, assistant replies as text, and function calls with their outputs. User speech has a transcript only when input transcription is configured through native session settings. An assistant reply that was interrupted, or still playing when the connection dropped, is cut to what the user saw or heard. The cut is estimated from how much of the audio played, so it can be off by a few words. Images, speech without a transcript, deleted items, and replies still generating when the connection dropped are left out. Restored items get new item IDs, don't fire item events, and count as input tokens on the next response. Item IDs from before the reconnect don't work in later events. If the provider refuses a restored item, an `error` event fires with code `context_restore_failed` and the conversation stays connected. +The conversation also restores recent context. Before `ready` fires, it adds back up to 40 of the most recent conversation items, 48 KiB in total. Restored items are user text, user speech that has a transcript, assistant replies as text, and function calls with their outputs. User speech has a transcript only when input transcription is configured through native session settings. An assistant reply that was interrupted, or still playing when the connection dropped, is cut to what the user saw or heard. The cut is estimated from how much of the audio played, so it can be off by a few words. Images, speech without a transcript, deleted items, and replies still generating when the connection dropped are left out. Restored items get new item IDs, don't fire item events, and count as input tokens on the next response. Item IDs from before the reconnect don't work in later events. Restoration uses a contiguous suffix of restorable items. An oversized newest item leaves that suffix empty. Omitted context or a provider refusal fires one `error` event with code `context_restore_failed` per restoration, and the conversation stays connected. Items are preserved without truncating text, function arguments or function results. When Puter knows ahead of time that the server a session runs on will restart, it sends a `puter.reconnect` event, which also reaches your listeners. Without that notice, a restart drops the connection and the conversation reconnects as above. Within a few seconds, the conversation opens a replacement session in the background while the current one keeps working. It switches once no reply is generating or playing, no user turn is being spoken or transcribed, no function call is waiting for its output, and nothing you sent is waiting on the provider. The switch goes through the same steps as a reconnect, and the `reconnecting` event has code 1012. If the replacement can't open yet, the conversation keeps the current session and tries again later. A refusal that retrying can't fix, such as `insufficient_funds`, stops the attempts. If the server closes the session first, the conversation switches to the replacement, or reconnects if none is open. diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index b4395eec0..f51328bdf 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -116,13 +116,13 @@ Realtime sessions ([`puter.ai.convo()`](/AI/convo/), `/puterai/openai/v1/realtim | Unauthenticated upgrade sockets, including closing sockets | 64 per node | Further upgrades receive HTTP 503. The slot is retained until successful admission or transport closure. | | Authentication setups in progress | 64 per node | Further upgrades receive HTTP 503. | | Presented credential | 8192 UTF-8 bytes | Longer or malformed credentials receive `invalid_api_key`. Unverified credentials allocate no distributed credential limiter state. Limiter lookup failures close with `internal_error` and 1011. | -| Replacement attempts marked `purpose=handover` | 20 per credential per minute | This shared limit is checked before the general 60-attempt limit. A refused probe consumes no general attempt. Initial and dropped-session recovery attempts retain their general budget. | | Transcription reservation window | 64 items at 10,896,000 micro-cents each, before deployment cost factors. Each item budgets 8,000 audio input, 8,000 text input and 4,096 text output tokens at the highest known rates, or 160 seconds at the highest duration rate, whichever costs more | Enabling transcription reserves 697,344,000 micro-cents, or $6.97344, at the largest cost factor of the supported transcription models. Audio waits until this reservation succeeds. The reservation is a budget rather than a cap on reported usage. | | Pending transcription identity | 64 items | Exhausting the reserved window pauses client reads until its accepted items settle and another window can be reserved. Unexpected provider commits beyond the window close the connection and bill unresolved work conservatively. Unresolved item identities are retained. | | Transcription usage timeout | 30 seconds per committed item | Missing usage, transcription failure, or an upstream close bills 10,896,000 micro-cents per unresolved item before cost factors. A late or duplicate completion is ignored. | | Transcription reservation lifetime | 90 seconds, renewed on the 30-second heartbeat | Credit stays held until accepted usage is recorded. Unused reservation credit is released when the connection ends. | | Decoded audio per transcription reservation window | 81,920,000 bytes | Further audio receives `rate_limit_exceeded` and close 1013. The budget uses the smallest supported audio byte rate. | | Completed transcription identity history | 128 items | Recent duplicate commits are ignored. Pending identities have a separate lifetime. | +| Replacement attempts marked `purpose=handover` | 20 per credential per minute | This shared limit is checked before the general 60-attempt limit. A refused probe consumes no general attempt. Initial and dropped-session recovery attempts retain their general budget. | | Credit held per response | 8,000 audio input and 4,096 audio output tokens at the session model's rates, held until the response's usage is recorded | A session opens only if the account can cover one hold. A response the account can't cover gets `insufficient_funds`, close 1008, and is billed for what it generated. A response whose usage never arrives because the provider connection closed first is billed the full hold. | | Restart notice (`puter.reconnect`) | At most once a minute per session | Repeated announcements within the minute aren't sent. | | In-flight response drain at shutdown | 5 seconds | Close 1012. | @@ -149,10 +149,10 @@ 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 | 10, backing off from 500 ms to 10 seconds, each delay randomized between half and one and a half times its value | The conversation closes with the original close code. | -| Replacement session after a restart notice | Opened within 3 seconds of the notice, then retried from 1 second, backing off to 10 seconds, each delay randomized between half and one and a half times its value. Settings on a held replacement are acknowledged within 3 seconds. | The current session keeps working until the conversation switches or the server closes it. A refusal that retrying can't fix stops the attempts. | +| Connection and reconnection attempts | 10, backing off from 500 ms to 10 seconds, each delay randomized between half and one and a half times its value | Initial transient failures retry through the original connection Promise. Terminal credential, model, funds, normal-close and oversized-event failures stop retries. An established conversation closes with the original close code after retries are exhausted. | +| Replacement session after a restart notice | Opened within 3 seconds of the notice, then retried from 1 second, backing off to 10 seconds, each delay randomized between half and one and a half times its value. Settings on a held replacement are acknowledged within 3 seconds. | The current session keeps working until the conversation switches or the server closes it. Probes stop after 3 attempts or 30 seconds. Conversations sharing one SDK instance and credential share a budget of 20 probes per minute. Dropped-session recovery uses its separate attempt loop. A refusal that retrying cannot fix stops the probes. | | Offline wait while reconnecting | 2 minutes per reconnection, in browsers that report being offline | Attempts continue after the wait. | -| Context restored after a reconnection | The 40 most recent restorable items, 48 KiB in total | Older items are left out. | +| Context restored after a reconnection | The 40 most recent restorable items, 48 KiB in total | Restoration uses a contiguous suffix of restorable items. If the newest item exceeds the byte budget, the suffix is empty. Omitted context reports `context_restore_failed` once per restoration. Text, function arguments and function results are preserved without truncation. | | 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 91d6312a5..1b633ff15 100644 --- a/src/puter-js/src/modules/ai/convo.js +++ b/src/puter-js/src/modules/ai/convo.js @@ -24,6 +24,7 @@ import { ConvoHistory } from './convoHistory.js'; * timer: ReturnType | null; * retry: ReturnType | null; * failures: number; + * startedAt: number; * refused: boolean; * probe: RealtimeSession | null; * created: RealtimeServerEvent | null; @@ -71,7 +72,7 @@ const RETRYABLE_ERRORS = new Set([ 'rate_limit_exceeded', 'internal_error', ]); -// Ten attempts over about a minute, long enough for a server restart. +// Bound setup retries so a persistent outage eventually closes the conversation. const RECONNECT_ATTEMPTS = 10; const RECONNECT_DELAY_MS = 500; const RECONNECT_MAX_DELAY_MS = 10_000; @@ -91,6 +92,11 @@ const REPLACEMENT_SPREAD_MS = 3_000; // Acknowledgment timeout on a held replacement, whose socket may have died // with the connection it was meant to replace. const REPLACEMENT_ACK_MS = 3_000; +/** @type {WeakMap} */ +const handoverBudgets = new WeakMap(); +const MAX_HANDOVER_ATTEMPTS_PER_MINUTE = 20; +const MAX_REPLACEMENT_ATTEMPTS = 3; +const REPLACEMENT_BUDGET_MS = 30_000; const TRANSCRIPT_TIMEOUT_MS = 30_000; // Bounds for the idle check's bookkeeping. const MAX_OWED = 64; @@ -272,6 +278,8 @@ export class Conversation extends RealtimeEvents { #connection = null; /** @type {{ type: 'close'; code: number; reason: string } | null} */ #lastClose = null; + /** @type {string | undefined} */ + #setupErrorCode; /** * @type {Set<{ * type: string; @@ -408,7 +416,7 @@ export class Conversation extends RealtimeEvents { this.#state = 'connecting'; this.#connection = (async () => { try { - await this.#open(); + await this.#openInitial(); this.#state = 'connected'; this.emitRealtime({ type: 'ready', @@ -430,6 +438,47 @@ export class Conversation extends RealtimeEvents { return this.#connection; } + /** @returns {Promise} */ + async #openInitial() { + const offlineUntil = Date.now() + OFFLINE_WAIT_MS; + for (let attempt = 0; attempt < RECONNECT_ATTEMPTS; attempt++) { + if (attempt) await this.#waitForRecovery(attempt - 1, offlineUntil); + if (this.#state === 'closed') + throw failure('session_closed', 'Conversation is closed'); + try { + await this.#open(); + return; + } catch (error) { + this.#endHandover(); + this.#detach(); + this.#resetTurnState(); + this.#rejectPending( + failure('connection_lost', 'Connection lost; reconnecting'), + ); + if ( + this.#state === 'closed' || + !RETRYABLE_ERRORS.has(error?.code) || + attempt === RECONNECT_ATTEMPTS - 1 + ) + throw error; + } + } + } + + /** + * @param {number} attempt + * @param {number} offlineUntil + * @returns {Promise} + */ + async #waitForRecovery(attempt, offlineUntil) { + const delay = Math.min( + RECONNECT_DELAY_MS * 2 ** attempt, + RECONNECT_MAX_DELAY_MS, + ); + await this.#pause(delay * (0.5 + Math.random())); + if (this.#state !== 'closed') await this.#whileOffline(offlineUntil); + } + /** * Opens a socket, waits for the provider session and applies settings. * @@ -443,6 +492,8 @@ export class Conversation extends RealtimeEvents { * @returns {Promise} */ async #open(reapply = false, prepared = null) { + this.#lastClose = null; + this.#setupErrorCode = undefined; if (prepared) { this.#session = prepared.session; this.#unsubscribe = this.#session.on('*', (event) => @@ -527,10 +578,14 @@ export class Conversation extends RealtimeEvents { this.#replayEvents.clear(); this.#replayReported = false; this.#history.replayed([]); - const entries = this.#history.replay( + const { entries, omittedItems } = this.#history.replay( MAX_REPLAY_ITEMS, MAX_REPLAY_BYTES, ); + if (omittedItems) + this.#replayFailed( + `${omittedItems} items omitted from the restore window`, + ); if (!entries.length) return; const events = entries.map(({ item }) => ({ type: 'conversation.item.create', @@ -602,14 +657,7 @@ export class Conversation extends RealtimeEvents { return; } if (!ready) { - const delay = Math.min( - RECONNECT_DELAY_MS * 2 ** attempt, - RECONNECT_MAX_DELAY_MS, - ); - // Jittered so clients dropped together don't return together. - await this.#pause(delay * (0.5 + Math.random())); - if (this.#state === 'closed') return; - await this.#whileOffline(offlineUntil); + await this.#waitForRecovery(attempt, offlineUntil); if (this.#state === 'closed') return; } try { @@ -764,10 +812,27 @@ export class Conversation extends RealtimeEvents { * @returns {Promise} */ async #openReplacement(handover) { + if (Date.now() - handover.startedAt >= REPLACEMENT_BUDGET_MS) { + handover.refused = true; + return; + } + let budget = handoverBudgets.get(this.#puter); + if (!budget || budget.token !== this.#puter.authToken) { + budget = { token: this.#puter.authToken, attempts: [] }; + handoverBudgets.set(this.#puter, budget); + } + budget.attempts = budget.attempts.filter( + (time) => time > Date.now() - 60_000, + ); + if (budget.attempts.length >= MAX_HANDOVER_ATTEMPTS_PER_MINUTE) { + handover.refused = true; + return; + } + budget.attempts.push(Date.now()); let session; try { session = new RealtimeSession( - createRealtimeSocket(this.#puter, this.model), + createRealtimeSocket(this.#puter, this.model, 'handover'), ); } catch { return; @@ -785,7 +850,10 @@ export class Conversation extends RealtimeEvents { }; const timer = setTimeout( () => done({ code: 'operation_timeout' }), - SESSION_TIMEOUT_MS, + Math.min( + SESSION_TIMEOUT_MS, + REPLACEMENT_BUDGET_MS - (Date.now() - handover.startedAt), + ), ); const off = session.on('*', (event) => { if (event.type === 'session.created') done({ created: event }); @@ -831,7 +899,11 @@ export class Conversation extends RealtimeEvents { handover.created = null; handover.stopProbe = null; if (session && session.socket.readyState < 2) session.close(); - if (code && !RETRYABLE_ERRORS.has(code)) { + if ( + (code && !RETRYABLE_ERRORS.has(code)) || + handover.failures >= MAX_REPLACEMENT_ATTEMPTS - 1 || + Date.now() - handover.startedAt >= REPLACEMENT_BUDGET_MS + ) { handover.refused = true; return; } @@ -1303,7 +1375,9 @@ export class Conversation extends RealtimeEvents { // the conversation even mid-reconnect. if ( this.#state === 'connecting' && - event.code !== CLOSE_MESSAGE_TOO_LARGE + (RECONNECT_CODES.has(event.code) || + (event.code !== 1000 && + RETRYABLE_ERRORS.has(this.#setupErrorCode))) ) { // `connect()` or the reconnect loop handles the failure. this.#rejectPending( @@ -1320,6 +1394,8 @@ export class Conversation extends RealtimeEvents { return; } if (event.type === 'error') { + if (this.#state === 'connecting') + this.#setupErrorCode = event.error?.code; const entry = this.#replayEvents.get(event.error?.event_id); if (entry) { this.#replayEvents.delete(event.error.event_id); @@ -1338,6 +1414,7 @@ export class Conversation extends RealtimeEvents { timer: null, retry: null, failures: 0, + startedAt: Date.now(), refused: false, probe: null, created: null, diff --git a/src/puter-js/src/modules/ai/convo.test.js b/src/puter-js/src/modules/ai/convo.test.js index 2b75a27e6..1f627919d 100644 --- a/src/puter-js/src/modules/ai/convo.test.js +++ b/src/puter-js/src/modules/ai/convo.test.js @@ -33,14 +33,14 @@ const owner = () => ({ env: 'nodejs', }); const conversations = new Set(); -const make = (options = {}) => { - const c = new Conversation(owner(), options); +const make = (options = {}, puter = owner()) => { + const c = new Conversation(puter, options); conversations.add(c); return c; }; const socket = () => FakeWebSocket.instances.at(-1); -const connected = async (options = {}) => { - const c = make(options); +const connected = async (options = {}, puter = owner()) => { + const c = make(options, puter); const ready = c.connect(); socket().deliver({ type: 'session.created', session: {} }); await Promise.resolve(); @@ -80,6 +80,193 @@ describe('interrupt readiness', () => { ); }); +describe('initial recovery', () => { + it('resolves the original connect after a transient failure', async () => { + vi.useFakeTimers(); + const c = make(); + const ready = c.connect(); + socket().close(1012, 'restart'); + await vi.advanceTimersByTimeAsync(2_000); + expect(FakeWebSocket.instances).toHaveLength(2); + socket().deliver({ type: 'session.created', session: {} }); + await Promise.resolve(); + socket().deliver({ type: 'session.updated', session: {} }); + await ready; + expect(c.state).toBe('connected'); + }); + + it.each(['upstream_error', 'internal_error', 'rate_limit_exceeded'])( + 'retries an initial %s refusal followed by policy close', + async (code) => { + fakeTime(); + const c = make(); + const ready = c.connect(); + socket().deliver({ type: 'error', error: { code } }); + socket().close(1008, code); + await vi.advanceTimersByTimeAsync(1_000); + expect(FakeWebSocket.instances).toHaveLength(2); + socket().deliver({ type: 'session.created', session: {} }); + await Promise.resolve(); + socket().deliver({ type: 'session.updated', session: {} }); + await ready; + expect(c.state).toBe('connected'); + }, + ); + + it('stops at a terminal credential failure after a transient initial close', async () => { + fakeTime(); + const c = make(); + const closed = vi.fn(); + c.on('close', closed); + const rejected = expect(c.connect()).rejects.toMatchObject({ + code: 'invalid_api_key', + }); + socket().close(1012, 'restart'); + await vi.advanceTimersByTimeAsync(1_000); + socket().deliver({ type: 'error', error: { code: 'invalid_api_key' } }); + await rejected; + await vi.advanceTimersByTimeAsync(60_000); + expect(c.state).toBe('closed'); + expect(closed).toHaveBeenCalledOnce(); + expect(closed.mock.calls[0][0].code).not.toBe(1012); + expect(FakeWebSocket.instances).toHaveLength(2); + }); + + it('cancels an initial retry when closed', async () => { + vi.useFakeTimers(); + const c = make(); + const result = expect(c.connect()).rejects.toMatchObject({ + code: 'session_closed', + }); + socket().close(1012, 'restart'); + await Promise.resolve(); + c.close(); + await vi.runAllTimersAsync(); + await result; + expect(FakeWebSocket.instances).toHaveLength(1); + }); +}); + +describe('bounded overlapping handovers', () => { + it('hands over idle conversations as slots become available at the twenty-session cap', async () => { + fakeTime(); + const puter = owner(); + const active = []; + const originals = []; + for (let i = 0; i < 11; i++) { + active.push(await connected({}, puter)); + originals.push(socket()); + } + for (const ws of originals) + ws.deliver({ type: 'puter.reconnect', reason: 'restart' }); + await vi.advanceTimersByTimeAsync(SPREAD); + const probes = FakeWebSocket.instances.slice(11); + expect(probes).toHaveLength(11); + expect( + probes.every( + (ws) => + new URL(ws.url).searchParams.get('purpose') === 'handover', + ), + ).toBe(true); + for (const ws of probes.slice(0, 9)) + ws.deliver({ type: 'session.created', session: {} }); + for (const ws of probes.slice(9)) + ws.deliver({ + type: 'error', + error: { code: 'rate_limit_exceeded' }, + }); + await vi.advanceTimersByTimeAsync(0); + for (const ws of probes.slice(0, 9)) + ws.deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(1_001); + const retries = FakeWebSocket.instances.slice(22); + expect(retries).toHaveLength(2); + for (const ws of retries) + ws.deliver({ type: 'session.created', session: {} }); + await vi.advanceTimersByTimeAsync(0); + for (const ws of retries) + ws.deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(active.every((c) => c.state === 'connected')).toBe(true); + expect(originals.every((ws) => ws.readyState === 3)).toBe(true); + expect(FakeWebSocket.instances).toHaveLength(24); + }); + + it('bounds a stalled probe by the elapsed budget while preserving the original', async () => { + const c = await connected(); + fakeTime(); + const original = socket(); + original.deliver({ type: 'puter.reconnect', reason: 'restart' }); + await vi.advanceTimersByTimeAsync(31_000); + expect(c.state).toBe('connected'); + expect(original.readyState).toBe(1); + expect( + FakeWebSocket.instances.slice(1).every((ws) => ws.readyState === 3), + ).toBe(true); + const count = FakeWebSocket.instances.length; + await vi.advanceTimersByTimeAsync(60_000); + expect(FakeWebSocket.instances).toHaveLength(count); + }); + + it('keeps busy originals and reserves capacity for dropped connections', async () => { + vi.useFakeTimers(); + const puter = owner(); + const originals = []; + for (let i = 0; i < 11; i++) { + await connected({}, puter); + const ws = socket(); + originals.push(ws); + ws.deliver({ type: 'response.created', response: { id: `r${i}` } }); + ws.deliver({ type: 'puter.reconnect', reason: 'restart' }); + } + for (let step = 0; step < 35; step++) { + await vi.advanceTimersByTimeAsync(1_000); + for (const probe of FakeWebSocket.instances.filter( + (ws) => !originals.includes(ws) && ws.readyState < 2, + )) { + probe.deliver({ + type: 'error', + error: { code: 'rate_limit_exceeded' }, + }); + } + } + const probes = FakeWebSocket.instances.length - originals.length; + expect(probes).toBeLessThanOrEqual(20); + expect(originals.every((ws) => ws.readyState === 1)).toBe(true); + for (const ws of originals) ws.close(1012, 'restart'); + await vi.advanceTimersByTimeAsync(2_000); + const recoveries = FakeWebSocket.instances.slice(11 + probes); + expect(recoveries).toHaveLength(11); + expect( + recoveries.every( + (ws) => !new URL(ws.url).searchParams.has('purpose'), + ), + ).toBe(true); + for (const ws of recoveries) + ws.deliver({ type: 'session.created', session: {} }); + await Promise.resolve(); + for (const ws of recoveries) + ws.deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(FakeWebSocket.instances.length).toBeLessThanOrEqual(60); + }); +}); + +describe('normal close policy', () => { + it('keeps a normal initial provider close terminal', async () => { + vi.useFakeTimers(); + const c = make(); + const result = expect(c.connect()).rejects.toMatchObject({ + code: 'session_closed', + }); + socket().close(1000, 'done'); + await result; + await vi.advanceTimersByTimeAsync(60_000); + expect(FakeWebSocket.instances).toHaveLength(1); + expect(c.state).toBe('closed'); + }); +}); + describe('conversation preparation', () => { it.each([ null, @@ -179,7 +366,7 @@ describe('Conversation lifecycle', () => { // Longer than the server's own 15-second setup budget. await vi.advanceTimersByTimeAsync(15_000); expect(c.state).toBe('connecting'); - await vi.advanceTimersByTimeAsync(5_000); + await vi.advanceTimersByTimeAsync(300_000); await result; expect(c.state).toBe('closed'); expect(vi.getTimerCount()).toBe(0); @@ -1517,6 +1704,27 @@ describe('restart notice', () => { }); describe('context replay after a reconnect', () => { + it.each(['x'.repeat(49 * 1024), '😀'.repeat(13 * 1024)])( + 'reports omitted context once and remains usable', + async (text) => { + const c = await connected(); + fakeTime(); + const error = vi.fn(); + c.on('error', error); + socket().deliver(userText('old', 'tiny')); + socket().deliver(userText('large', text)); + socket().close(1006); + await answerReconnect(); + expect(replayed()).toEqual([]); + expect(error).toHaveBeenCalledTimes(1); + expect(error.mock.calls[0][0]).toMatchObject({ + code: 'context_restore_failed', + }); + expect(c.state).toBe('connected'); + expect(() => c.sendText('continue')).not.toThrow(); + }, + ); + it('restores finished items in order, before ready, after a dropped connection', async () => { const c = await connected(); fakeTime(); diff --git a/src/puter-js/src/modules/ai/convoHistory.js b/src/puter-js/src/modules/ai/convoHistory.js index 2ab8b778b..9f36f5b35 100644 --- a/src/puter-js/src/modules/ai/convoHistory.js +++ b/src/puter-js/src/modules/ai/convoHistory.js @@ -238,7 +238,7 @@ export class ConvoHistory { * * @param {number} maxItems * @param {number} maxBytes - * @returns {ReplayEntry[]} + * @returns {{ entries: ReplayEntry[]; omittedItems: number }} */ replay(maxItems, maxBytes) { /** @type {ReplayEntry[]} */ @@ -264,10 +264,17 @@ export class ConvoHistory { .filter(({ item }) => item.type === 'function_call') .map(({ item }) => item.call_id), ); - return entries.filter( + const filtered = entries.filter( ({ item }) => item.type !== 'function_call_output' || calls.has(item.call_id), ); + const restorable = this.#records.filter((record) => + this.#replayItem(record), + ).length; + return { + entries: filtered, + omittedItems: restorable - filtered.length, + }; } /** diff --git a/src/puter-js/src/modules/ai/convoHistory.test.js b/src/puter-js/src/modules/ai/convoHistory.test.js index fc18eb201..7641f415b 100644 --- a/src/puter-js/src/modules/ai/convoHistory.test.js +++ b/src/puter-js/src/modules/ai/convoHistory.test.js @@ -6,7 +6,9 @@ const user = (id, content) => ({ item: { id, type: 'message', role: 'user', status: 'completed', content }, }); const texts = (history) => - history.replay(40, 48 * 1024).map(({ item }) => item.content[0].text); + history + .replay(40, 48 * 1024) + .entries.map(({ item }) => item.content[0].text); describe('ConvoHistory', () => { it('cuts text without word breaks at the played share', () => { @@ -60,7 +62,7 @@ describe('ConvoHistory', () => { history.observe( user(`u${i}`, [{ type: 'input_text', text: `${i}` }]), ); - expect(history.replay(100, Infinity)).toHaveLength(80); + expect(history.replay(100, Infinity).entries).toHaveLength(80); expect(texts(history).at(-1)).toBe('99'); }); @@ -74,6 +76,25 @@ describe('ConvoHistory', () => { expect(texts(history)).toEqual(['new']); }); + it.each(['x'.repeat(49 * 1024), '😀'.repeat(13 * 1024)])( + 'reports an oversized newest item without truncating it', + (text) => { + const history = new ConvoHistory(() => false); + history.observe( + user('old', [{ type: 'input_text', text: 'tiny' }]), + ); + history.observe(user('large', [{ type: 'input_text', text }])); + expect(history.replay(40, 48 * 1024)).toEqual({ + entries: [], + omittedItems: 2, + }); + expect( + history.replay(40, Infinity).entries.at(-1).item.content[0] + .text, + ).toBe(text); + }, + ); + it('waits on every unanswered call since the last user message', () => { const history = new ConvoHistory(() => false); const call = (id, status = 'completed') => @@ -114,7 +135,7 @@ describe('ConvoHistory', () => { it('reports the new session copies of replayed items until they are done', () => { const history = new ConvoHistory(() => false); history.observe(user('old', [{ type: 'input_text', text: 'hi' }])); - history.replayed(history.replay(40, 48 * 1024)); + history.replayed(history.replay(40, 48 * 1024).entries); const copy = user('new', [{ type: 'input_text', text: 'hi' }]); expect(history.observe(copy)).toBe(true); expect( diff --git a/src/puter-js/src/modules/ai/convoTransport.js b/src/puter-js/src/modules/ai/convoTransport.js index db9448574..72491bd3f 100644 --- a/src/puter-js/src/modules/ai/convoTransport.js +++ b/src/puter-js/src/modules/ai/convoTransport.js @@ -10,6 +10,7 @@ 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; @@ -29,6 +30,7 @@ const exceedsBytes = (data, limit) => /** * One live realtime session. Events in both directions are OpenAI Realtime API * events, unchanged. + * * @internal */ export class RealtimeSession extends RealtimeEvents { @@ -77,13 +79,15 @@ export class RealtimeSession extends RealtimeEvents { /** * @param {import('../../index.js').Puter} puter * @param {string} [model] + * @param {'handover'} [purpose] * @returns {WebSocket} * @internal */ -export function createRealtimeSocket(puter, model) { +export function createRealtimeSocket(puter, model, purpose) { const url = new URL(REALTIME_PATH, puter.APIOrigin); url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; if (model) url.searchParams.set('model', model); + if (purpose) url.searchParams.set('purpose', purpose); return new WebSocket(url.toString(), [ 'realtime', `openai-insecure-api-key.${puter.authToken}`, diff --git a/src/puter-js/tests/api/realtime/convo.test.ts b/src/puter-js/tests/api/realtime/convo.test.ts index f1a80bcd3..351079ef8 100644 --- a/src/puter-js/tests/api/realtime/convo.test.ts +++ b/src/puter-js/tests/api/realtime/convo.test.ts @@ -33,6 +33,7 @@ describe('convo built SDK and browser media', () => { received: { type: string; item?: unknown }[]; }[] = []; let ids = 0; + let upstreamRefusals = 0; beforeAll(async () => { upstream = new WebSocketServer({ port: 0 }); upstream.on('connection', (socket) => { @@ -41,6 +42,11 @@ describe('convo built SDK and browser media', () => { received: [] as { type: string; item?: unknown }[], }; connections.push(connection); + if (upstreamRefusals > 0) { + upstreamRefusals--; + socket.close(1012, 'restart'); + return; + } socket.send( JSON.stringify({ type: 'session.created', session: {} }), ); @@ -211,6 +217,22 @@ describe('convo built SDK and browser media', () => { await new Promise((resolve) => upstream?.close(() => resolve())); }, 60_000); + it('resolves the original built SDK connect after an initial provider restart', async () => { + upstreamRefusals = 1; + const before = connections.length; + const conversation = await puter.ai.convo({ output: ['text'] }); + try { + const ready = conversation.connect(); + expect(conversation.connect()).toBe(ready); + await ready; + expect(conversation.state).toBe('connected'); + expect(connections.length - before).toBe(2); + } finally { + upstreamRefusals = 0; + conversation.close(); + } + }); + it('prepares, configures, sends, interrupts and truncates using the built SDK', async () => { const conversation = await puter.ai.convo({ output: ['text'], diff --git a/src/puter-js/tests/api/suites/ai.suite.ts b/src/puter-js/tests/api/suites/ai.suite.ts index ed931f068..1befb144f 100644 --- a/src/puter-js/tests/api/suites/ai.suite.ts +++ b/src/puter-js/tests/api/suites/ai.suite.ts @@ -1789,6 +1789,21 @@ export default suite('ai', { } }, + 'convo rejects interruption readiness through its Promise': async (t) => { + const conversation = await t.puter.ai.convo({ output: ['text'] }); + try { + const prepared = conversation.interrupt(); + t.assert.equal(typeof prepared.then, 'function'); + await t.assert.rejects(() => prepared); + conversation.close(); + const closed = conversation.interrupt(); + t.assert.equal(typeof closed.then, 'function'); + await t.assert.rejects(() => closed); + } finally { + conversation.close(); + } + }, + 'convo is the realtime SDK entry point': async (t) => { t.assert.equal('assistant' in t.puter.ai, false); t.assert.equal('conversation' in t.puter.ai, false);