From 4881264c88ffeda4331b33d2e2b1890a5d190be6 Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Fri, 30 Oct 2026 18:06:54 -0700 Subject: [PATCH] feat(ai): realtime reconnect notice, context restore and retry hardening (#4415) * test(ai): end every realtime relay test's upstream before the next The fake upstream now answers response.create with response.created first, as OpenAI does, so closing those sessions no longer waits out the settle window. * feat(ai): ask realtime clients to reconnect before a restart Open realtime sessions get a puter.reconnect event when the drain starts, with close_in_ms set to the drain window, and new sessions are still refused until the 1012 close at shutdown. A host extension can announce a restart early by emitting serverRestartAnnounced; that only sends the notice, so a false alarm leaves the server working normally. A session gets the notice at most once a minute. onServerDrainStart now receives the drain duration. * fix(ai): retry realtime reconnects for about a minute and wait out offline browsers Five attempts gave up after about 15 seconds, shorter than a server restart. The conversation makes ten attempts with the backoff capped at 10 seconds, and while the browser reports being offline it waits for the online event, up to two minutes per reconnect, instead of spending attempts. close() ends either wait and removes its listener. * feat(ai): restore recent conversation context after a realtime reconnect The provider starts a fresh session on every reconnect, so the conversation lost all earlier items. The SDK now records finished items it can recreate from the provider's own events: user text, user speech with a transcript, assistant replies as text, and function calls with their outputs. After the new session acknowledges its settings, and before ready fires, it adds back the newest of them, up to 40 items and 48 KiB, which stays under the relay's upload burst. Assistant replies are cut to what the user saw or heard: the deltas shown before an interruption, and the share of the transcript matching the audio played before a truncation or a drop. Deleted items, unfinished replies, images, speech without a transcript and incomplete function calls are left out, as are function outputs whose call falls outside the window. Restored copies take over their new item IDs. A refused item is reported once as context_restore_failed without failing the reconnect. * feat(ai): move a realtime conversation to a new session on the restart notice On puter.reconnect the conversation waits until it is idle: no reply requested, generating or playing, no user turn being spoken or transcribed, no function call awaiting its output and no acknowledgment pending. Then it reconnects through the usual path, emitting reconnecting (code 1012) and ready with reconnected: true, without an error. Until then the session keeps working, and if the server closes it first the normal 1012 reconnect takes over. close() at any point stops the wait. * docs(ai): realtime restart notice, context restore and reconnect limits * test(ai): realtime reconnects against the local server The mock provider answers with item events and whole replies like OpenAI. New cases cover a dropped connection (cut through a TCP proxy so the client sees 1006), a host-announced restart, and a draining server refusing the reconnect; the first two check the provider receives the restored items before the next turn. The browser media test waits for captured audio and the reply instead of a fixed sleep. The live suite gains a tool and text context round trip across a restart notice. * ci: run the live realtime conversation tests with the AI provider suite They need the OpenAI key, which only this workflow has, and skip themselves without it. * fix(ai): open the replacement before leaving a realtime session on the restart notice The handover closed the current session and then reconnected, so every notice cost a second or two in which sends failed and microphone audio was dropped. On a single server that refuses new sessions while it drains, the retries ran out during the drain and the conversation closed sooner than if it had stayed. The conversation now opens a replacement session as soon as the notice arrives and switches only once it is open and the conversation is idle. A refused replacement is retried with backoff while the current session keeps working; when the server closes it first, the held replacement is used right away. The idle check also missed some in-progress work: - a reply created while the user was still speaking ended the user's turn; speech now lasts until its audio is committed or cleared, and a committed turn waits for the reply server turn detection creates - only the newest function call was checked; every finished call since the last user message needs its output, and calls cut off by a cancellation don't count - native session.update and conversation.item.create events the provider hadn't acknowledged could be lost; the switch waits for them, up to the acknowledgment timeout for items - only the newest transcription was waited on; every committed item is - clearing playback didn't recheck the switch Restored items stay internal: the new session's copies of them no longer reach listeners. At a reconnect, every queued item the user didn't hear is cut, not just the one playing. A notice from a failed reconnect attempt is dropped. * test(ai): realtime replay after split settings, and a draining server refusing the replacement The relay accepts the context replay right after two full-size settings updates. Against the local server, a draining relay refuses the replacement session while the current one keeps answering. * ci: build the SDK in the live realtime step so it stands on its own * fix(ai): a failed realtime reconnect attempt no longer poisons the next An attempt that dropped after sending its session.update but before the acknowledgment left that event ID queued, so the next attempt's acknowledgment was matched to it and timed out, and so on until the conversation closed. Turn state is now reset after each failed attempt. * fix(ai): bound the realtime handover's waits and stop retries that can't succeed - The first replacement opens within a randomized 3 seconds of the notice, so sessions notified together don't all open at once. - A replacement refused with a code retrying can't fix, such as insufficient_funds, ends the attempts; the server's close still falls back to the usual reconnect. - A held replacement that gets an error or its own server's notice is closed and replaced. - A held replacement used after a drop gets a 3 second acknowledgment timeout, since its socket may have died with the network. - close() during the switch closes the held replacement too. - The reply to a committed turn and each transcription hold the switch for at most 10 and 30 seconds, like app items already did, and an app item is released only by an added item of its type, role and call ID. * docs(ai): the realtime restart notice isn't guaranteed before every restart --- .../ai-provider-integration-tests.yaml | 10 + src/backend/clients/event/types.ts | 9 + src/backend/server.test.ts | 12 +- src/backend/server.ts | 8 +- .../OpenAIRealtimeRelayService.test.ts | 221 ++- .../realtime/OpenAIRealtimeRelayService.ts | 56 +- src/backend/services/realtime/README.md | 12 +- src/backend/types.ts | 7 +- src/backend/util/gracefulShutdown.test.ts | 7 +- src/backend/util/gracefulShutdown.ts | 4 +- src/docs/src/AI/convo.md | 12 +- src/docs/src/AI/realtime.md | 17 + src/docs/src/rate-limits-and-quotas.md | 6 +- src/puter-js/src/modules/ai/convo.js | 679 ++++++++- src/puter-js/src/modules/ai/convo.test.js | 1240 ++++++++++++++++- src/puter-js/src/modules/ai/convoHistory.js | 524 +++++++ .../src/modules/ai/convoHistory.test.js | 134 ++ src/puter-js/src/modules/ai/convoMedia.js | 56 +- .../src/modules/ai/convoMedia.test.js | 60 + src/puter-js/tests/api/realtime/convo.test.ts | 314 ++++- .../tests/api/realtime/convoLive.test.ts | 125 +- 21 files changed, 3341 insertions(+), 172 deletions(-) create mode 100644 src/puter-js/src/modules/ai/convoHistory.js create mode 100644 src/puter-js/src/modules/ai/convoHistory.test.js diff --git a/.github/workflows/ai-provider-integration-tests.yaml b/.github/workflows/ai-provider-integration-tests.yaml index 614e7b027..b319e61ae 100644 --- a/.github/workflows/ai-provider-integration-tests.yaml +++ b/.github/workflows/ai-provider-integration-tests.yaml @@ -47,3 +47,13 @@ jobs: PUTER_TEST_AI_MOONSHOT_API_KEY: ${{ secrets.PUTER_TEST_AI_MOONSHOT_API_KEY }} PUTER_TEST_AI_ZAI_API_KEY: ${{ secrets.PUTER_TEST_AI_ZAI_API_KEY }} PUTER_TEST_AI_ELEVENLABS_API_KEY: ${{ secrets.PUTER_TEST_AI_ELEVENLABS_API_KEY }} + + # Realtime conversations through the SDK against live OpenAI. Builds + # the SDK bundle these tests load itself, so it runs even when the + # step above fails; the tests skip themselves without the key. + - name: Run live realtime conversation tests + if: ${{ !cancelled() }} + run: npm run setupExtensions && (cd src/puter-js && npm run build) && npx vitest run --config src/puter-js/tests/api/vitest.config.ts src/puter-js/tests/api/realtime/convoLive.test.ts + env: + CI: 'true' + PUTER_TEST_AI_OPENAI_API_KEY: ${{ secrets.PUTER_TEST_AI_OPENAI_API_KEY }} diff --git a/src/backend/clients/event/types.ts b/src/backend/clients/event/types.ts index b9ca49a9b..e20d5fda1 100644 --- a/src/backend/clients/event/types.ts +++ b/src/backend/clients/event/types.ts @@ -75,6 +75,15 @@ export type EventMap = { serverStart: Record; serverPrepareShutdown: Record; serverShutdown: Record; + /** + * Emitted by a host extension that knows this server will restart soon. + * Long-lived connections ask their clients to reconnect; nothing is refused + * or closed, so a false alarm is harmless. + */ + serverRestartAnnounced: { + /** How long until the server stops, when known. */ + closeInMs?: number; + }; // ---- AI ---- 'ai.prompt.validate': { diff --git a/src/backend/server.test.ts b/src/backend/server.test.ts index 17ab78c26..b5ef0aefb 100644 --- a/src/backend/server.test.ts +++ b/src/backend/server.test.ts @@ -1400,7 +1400,7 @@ describe('PuterServer keep-alive timeout', () => { }); describe('PuterServer lifecycle hooks', () => { - it('runs prepare-shutdown bottom-up, once, before tearing down', async () => { + it('runs drain-start and prepare-shutdown bottom-up, once, before tearing down', async () => { const port = await allocateEphemeralPort(); const server = await setupTestServer({ port } as unknown as IConfig, { listen: true, @@ -1408,6 +1408,9 @@ describe('PuterServer lifecycle hooks', () => { const order: string[] = []; const probe = (label: string) => ({ + onServerDrainStart: (info?: { drainMs?: number }) => { + order.push(`drain:${label}:${info?.drainMs}`); + }, onServerPrepareShutdown: () => { order.push(`prepare:${label}`); }, @@ -1425,10 +1428,15 @@ describe('PuterServer lifecycle hooks', () => { (layer as Record).hookOrderProbe = probe(label); } - await server.prepareShutdown(); + await server.prepareShutdown({ drainMs: 1_000 }); await server.shutdown(); expect(order).toEqual([ + 'drain:client:1000', + 'drain:store:1000', + 'drain:service:1000', + 'drain:controller:1000', + 'drain:driver:1000', 'prepare:client', 'prepare:store', 'prepare:service', diff --git a/src/backend/server.ts b/src/backend/server.ts index b57bee1f6..a2206286a 100644 --- a/src/backend/server.ts +++ b/src/backend/server.ts @@ -117,6 +117,7 @@ import type { IConfig, LayerInstances, PagerSeverity, + ServerDrainInfo, WithControllerRegistration, WithLifecycle, } from './types'; @@ -1698,6 +1699,7 @@ export class PuterServer { | 'onServerDrainStart' | 'onServerPrepareShutdown' | 'onServerShutdown', + info?: ServerDrainInfo, ) { const layers = [ this.clients, @@ -1709,7 +1711,7 @@ export class PuterServer { if (order === 'topDown') layers.reverse(); for (const layer of layers) { for (const instance of Object.values(layer) as WithLifecycle[]) { - await instance[hook]?.(); + await instance[hook]?.(info); } } } @@ -1726,10 +1728,10 @@ export class PuterServer { await this.#runHook('bottomUp', 'onServerPrepareShutdown'); } - async prepareShutdown() { + async prepareShutdown(info: ServerDrainInfo = {}) { if (this.#server) { try { - await this.#runHook('bottomUp', 'onServerDrainStart'); + await this.#runHook('bottomUp', 'onServerDrainStart', info); } catch (err) { console.error('[server] onServerDrainStart failed', err); } diff --git a/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts b/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts index d7c2d0cb9..b15a406b5 100644 --- a/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts +++ b/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts @@ -164,7 +164,10 @@ describe('realtimeTranscriptionRows', () => { describe('realtimeBillingModel', () => { it('bills the model the session reports, dated snapshots included', () => { expect( - realtimeBillingModel({ model: 'gpt-realtime' }, 'gpt-realtime-mini'), + realtimeBillingModel( + { model: 'gpt-realtime' }, + 'gpt-realtime-mini', + ), ).toBe('gpt-realtime'); expect( realtimeBillingModel( @@ -193,6 +196,7 @@ describe('OpenAIRealtimeRelayService relay', () => { const upstreamRequests: IncomingMessage[] = []; const upstreamEvents: { type: string; response_id?: string }[] = []; const clients = new Set(); + let responseCount = 0; // Realtime is subscriber-only; the test user is on a paid plan with a $10 // allowance unless a test says otherwise. @@ -208,10 +212,14 @@ describe('OpenAIRealtimeRelayService relay', () => { ); }); - afterEach(() => { + afterEach(async () => { for (const client of clients) client.terminate(); clients.clear(); vi.restoreAllMocks(); + // Every session, settle included, has ended its upstream. + await vi.waitFor(() => expect(upstream.clients.size).toBe(0), { + timeout: 8_000, + }); }); beforeAll(async () => { @@ -405,10 +413,18 @@ describe('OpenAIRealtimeRelayService relay', () => { 150, ); } else if (event.type === 'response.create') { + const id = `resp_${++responseCount}`; + ws.send( + JSON.stringify({ + type: 'response.created', + response: { id }, + }), + ); ws.send( JSON.stringify({ type: 'response.done', response: { + id, usage: { input_token_details: { text_tokens: 7 }, output_token_details: { text_tokens: 3 }, @@ -586,7 +602,9 @@ describe('OpenAIRealtimeRelayService relay', () => { 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 })); + 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'); @@ -852,11 +870,20 @@ describe('OpenAIRealtimeRelayService relay', () => { await vi.waitFor(() => expect( spy.mock.calls.map(([, rows]) => - rows.map((row) => [row.usageType, row.costOverride]), + rows.map((row) => [ + row.usageType, + row.costOverride, + ]), ), ).toContainEqual([ - ['openai:gpt-realtime-mini:realtime:text_input', 7 * 60 * 2], - ['openai:gpt-realtime-mini:realtime:text_output', 3 * 240 * 2], + [ + 'openai:gpt-realtime-mini:realtime:text_input', + 7 * 60 * 2, + ], + [ + 'openai:gpt-realtime-mini:realtime:text_output', + 3 * 240 * 2, + ], ]), ); } finally { @@ -1120,7 +1147,9 @@ describe('OpenAIRealtimeRelayService relay', () => { ); await c.waitFor('session.created'); c.ws.send(JSON.stringify({ type: 'transcribe.switch' })); - await c.waitFor('conversation.item.input_audio_transcription.completed'); + await c.waitFor( + 'conversation.item.input_audio_transcription.completed', + ); await vi.waitFor(() => expect( spy.mock.calls.flatMap(([, rows]) => @@ -1168,29 +1197,167 @@ describe('OpenAIRealtimeRelayService relay', () => { c.ws.close(); }); - // 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; - const live = open( - `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, + it('accepts a context replay burst right after full-size settings', async () => { + // The SDK reapplies large settings as several updates, each sent after + // the last is acknowledged, then replays up to 40 items and 48 KiB in + // one burst. + const c = open(`${relayUrl}?auth_token=${user.token}`); + await c.waitFor('session.created'); + for (let i = 0; i < 2; i++) { + c.ws.send( + JSON.stringify({ + type: `test.settings${i}`, + pad: 'x'.repeat(MAX_REALTIME_MESSAGE_BYTES - 64), + }), + ); + await c.waitForEcho(`test.settings${i}`); + } + const item = JSON.stringify({ + type: 'test.replay', + item: { type: 'message', role: 'user', pad: 'x'.repeat(1100) }, + }); + expect(item.length * 40).toBeLessThanOrEqual(48 * 1024); + for (let i = 0; i < 40; i++) c.ws.send(item); + await vi.waitFor( + () => + expect( + c.events.filter( + (e) => + e.type === 'echo' && + (e.event as { type: string }).type === + 'test.replay', + ), + ).toHaveLength(40), + { timeout: 5_000 }, ); - await live.waitFor('session.created'); - live.ws.send(JSON.stringify({ type: 'slow.response' })); - await live.waitFor('response.created'); + expect(c.events.some((e) => e.type === 'error')).toBe(false); + c.ws.close(); + }); - relay.onServerDrainStart(); - const late = open( - `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, - ); - const error = (await late.waitFor('error')) as { - error: { code: string }; + describe('restart', () => { + type Notice = { + type: string; + event_id: string; + reason: string; + close_in_ms?: number; }; - expect(error.error.code).toBe('server_restarting'); - expect((await late.closed).code).toBe(1012); - expect(live.ws.readyState).toBe(WebSocket.OPEN); + const notices = (c: ReturnType) => + c.events.filter((e) => e.type === 'puter.reconnect') as Notice[]; - await relay.onServerPrepareShutdown(); - expect((await live.closed).code).toBe(1012); - expect(live.events.some((e) => e.type === 'response.done')).toBe(true); + it('sends only a reconnect notice when a host extension announces a restart', async () => { + const relay = server.services.openaiRealtime; + const c = open(`${relayUrl}?auth_token=${user.token}`); + await c.waitFor('session.created'); + server.clients.event.emit( + 'serverRestartAnnounced', + { closeInMs: 300_000.4 }, + {}, + ); + const notice = (await c.waitFor('puter.reconnect')) as Notice; + expect(notice).toEqual({ + type: 'puter.reconnect', + event_id: expect.stringMatching(/^event_puter_/), + reason: 'server_restarting', + close_in_ms: 300_000, + }); + + // Repeats within a minute are dropped; the session keeps working. + server.clients.event.emit('serverRestartAnnounced', {}, {}); + await relay.recheckSessions(); + c.ws.send('{"type":"test.alive"}'); + await c.waitForEcho('test.alive'); + expect(notices(c)).toHaveLength(1); + + // New sessions are still accepted, and get no notice. + const fresh = open(`${relayUrl}?auth_token=${user.token}`); + await fresh.waitFor('session.created'); + fresh.ws.send('{"type":"test.alive"}'); + await fresh.waitForEcho('test.alive'); + expect(notices(fresh)).toHaveLength(0); + expect(c.ws.readyState).toBe(WebSocket.OPEN); + c.ws.close(); + fresh.ws.close(); + }); + + it('omits close_in_ms when the announcement has no usable duration', async () => { + const c = open(`${relayUrl}?auth_token=${user.token}`); + await c.waitFor('session.created'); + server.clients.event.emit( + 'serverRestartAnnounced', + { closeInMs: -1 }, + {}, + ); + const notice = (await c.waitFor('puter.reconnect')) as Notice; + expect(notice).not.toHaveProperty('close_in_ms'); + c.ws.close(); + }); + + // Last: the relay is shut down afterwards. + it('drains: notifies open sessions, refuses new ones, bills leavers, then closes the rest with 1012', async () => { + const relay = server.services.openaiRealtime; + const spy = vi.spyOn( + server.services.metering, + 'batchIncrementUsages', + ); + const url = `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`; + const idle = open(url); + const busy = open(url); + const leaver = open(url); + await Promise.all( + [idle, busy, leaver].map((c) => c.waitFor('session.created')), + ); + leaver.ws.send(JSON.stringify({ type: 'hold.response' })); + await leaver.waitFor('response.created'); + + await relay.onServerDrainStart({ drainMs: 90_000 }); + for (const c of [idle, busy, leaver]) { + expect(await c.waitFor('puter.reconnect')).toMatchObject({ + reason: 'server_restarting', + close_in_ms: 90_000, + }); + } + + const late = open(url); + expect(await late.waitFor('error')).toMatchObject({ + error: { code: 'server_restarting' }, + }); + expect((await late.closed).code).toBe(1012); + + // A client that leaves after the notice is billed as before: its + // response is cancelled and settled. + spy.mockClear(); + leaver.ws.close(); + await vi.waitFor(() => + expect(upstreamEvents).toContainEqual({ + type: 'response.cancel', + response_id: 'resp_hold', + }), + ); + await vi.waitFor(() => + expect( + spy.mock.calls.some(([, rows]) => + rows.some( + (r) => + r.usageType === + 'openai:gpt-realtime-mini:realtime:audio_output', + ), + ), + ).toBe(true), + ); + + // The rest keep working until shutdown. + idle.ws.send('{"type":"test.alive"}'); + await idle.waitForEcho('test.alive'); + busy.ws.send(JSON.stringify({ type: 'slow.response' })); + await busy.waitFor('response.created'); + + await relay.onServerPrepareShutdown(); + expect((await idle.closed).code).toBe(1012); + expect((await busy.closed).code).toBe(1012); + expect(busy.events.some((e) => e.type === 'response.done')).toBe( + true, + ); + expect(notices(idle)).toHaveLength(1); + }); }); }); diff --git a/src/backend/services/realtime/OpenAIRealtimeRelayService.ts b/src/backend/services/realtime/OpenAIRealtimeRelayService.ts index 44706019f..e8f637a0f 100644 --- a/src/backend/services/realtime/OpenAIRealtimeRelayService.ts +++ b/src/backend/services/realtime/OpenAIRealtimeRelayService.ts @@ -38,6 +38,7 @@ import { kv } from '../../util/kvSingleton.js'; import { assertActorHasSubscription } from '../metering/enforcement.js'; import type { CreditHold, UsageInput } from '../metering/types.js'; import { decideSocketAuth } from '../socket/SocketService.js'; +import type { ServerDrainInfo } from '../../types.js'; import { PuterService } from '../types.js'; export const REALTIME_PATH = '/puterai/openai/v1/realtime'; @@ -165,6 +166,11 @@ const DRAIN_RESPONSE_MS = 5_000; const SETTLE_MS = 5_000; // RFC 6455 "Service Restart": the client should reconnect. const CLOSE_RESTART = 1012; +// Puter's own server event asking the client to reconnect when idle. OpenAI +// event types never start with `puter.`. +export const RECONNECT_EVENT = 'puter.reconnect'; +// A session is sent the reconnect notice at most this often. +const RECONNECT_NOTICE_INTERVAL_MS = 60_000; const TRANSCRIPTION_DONE = 'conversation.item.input_audio_transcription.completed'; const METERING_DRIVER = 'openai-realtime'; @@ -438,6 +444,18 @@ const errorEvent = (code: string, message: string): string => }, }); +/** + * Asks the client to reconnect at its next idle moment. `close_in_ms`, when + * known, is how long until the server closes the session itself. + */ +const reconnectEvent = (closeInMs: number | undefined): string => + JSON.stringify({ + type: RECONNECT_EVENT, + event_id: `event_puter_${Date.now()}`, + reason: 'server_restarting', + ...(closeInMs !== undefined ? { close_in_ms: closeInMs } : {}), + }); + /** * Relays OpenAI's Realtime API over WebSocket at `REALTIME_PATH` on the api * host, one upstream connection per client connection. Frames pass through @@ -456,6 +474,7 @@ export class OpenAIRealtimeRelayService extends PuterService { heartbeat: () => void; recheck: () => Promise; drain: () => Promise; + notify: (closeInMs: number | undefined) => void; }>(); #heartbeatTimer: ReturnType | null = null; #authTimer: ReturnType | null = null; @@ -519,12 +538,37 @@ export class OpenAIRealtimeRelayService extends PuterService { } } - /** New sessions are refused; open ones keep running through the drain. */ - override onServerDrainStart(): void { + /** A host extension can announce a restart before the drain starts. */ + override onServerStart(): void { + this.clients.event.on('serverRestartAnnounced', (_key, data) => + this.#announceRestart(data?.closeInMs), + ); + } + + /** + * New sessions are refused; open ones keep running through the drain and + * are asked to reconnect. + */ + override onServerDrainStart(info?: ServerDrainInfo): void { this.#draining = true; console.info( `[realtime] draining; ${this.#sessions.size} sessions stay open`, ); + this.#announceRestart(info?.drainMs); + } + + /** + * Sends every open session the reconnect notice, at most once a minute + * each. Nothing is refused or closed here. + */ + #announceRestart(closeInMs: unknown): void { + const ms = + typeof closeInMs === 'number' && + Number.isFinite(closeInMs) && + closeInMs >= 0 + ? Math.round(closeInMs) + : undefined; + for (const session of this.#sessions) session.notify(ms); } /** @@ -581,6 +625,7 @@ export class OpenAIRealtimeRelayService extends PuterService { // The transcription model in force when each input item was committed. const committedModels = new Map(); let onIdle: (() => void) | null = null; + let noticeAt = -Infinity; const model = url.searchParams.get('model') || DEFAULT_REALTIME_MODEL; let billingModel = model; let transcriptionModel: string | undefined; @@ -698,6 +743,13 @@ export class OpenAIRealtimeRelayService extends PuterService { } session.close(CLOSE_RESTART, 'server restarting'); }, + notify: (closeInMs: number | undefined) => { + if (closed || client.readyState !== WebSocket.OPEN) return; + const now = Date.now(); + if (now - noticeAt < RECONNECT_NOTICE_INTERVAL_MS) return; + noticeAt = now; + client.send(reconnectEvent(closeInMs)); + }, }; this.#sessions.add(session); diff --git a/src/backend/services/realtime/README.md b/src/backend/services/realtime/README.md index 994751769..b6b8857c6 100644 --- a/src/backend/services/realtime/README.md +++ b/src/backend/services/realtime/README.md @@ -63,8 +63,13 @@ the transcription models in force at commit and at completion. Session attempts are limited per credential before authentication runs. Credential checks run in batches of at most 16 with no overlapping sweeps. -During drain, new sessions receive `server_restarting` and close with 1012. -Open sessions can finish an in-flight response for up to five seconds at shutdown. +When the drain starts, open sessions get a `puter.reconnect` event and new +sessions receive `server_restarting` and close with 1012. Open sessions can +finish an in-flight response for up to five seconds at shutdown, then close +with 1012. A host extension can announce a restart early by emitting +`serverRestartAnnounced` (optionally with `closeInMs`); that only sends the +notice, so a false alarm leaves the server working normally. Each session gets +the notice at most once a minute. ## Configuration @@ -83,4 +88,5 @@ Responses and input transcriptions are billed after completion; holds bound what is in flight. Sessions require a paid plan (`subscription.gate.realtime.openai` can waive it) and accept session, app and full-access API tokens. Conversation state lives upstream and is lost when a connection ends; the SDK reconnects -with the same settings but an empty conversation. +with the same settings and replays its recent items, with assistant replies cut +to what the user heard. diff --git a/src/backend/types.ts b/src/backend/types.ts index c622a0edd..2fcc9a46c 100644 --- a/src/backend/types.ts +++ b/src/backend/types.ts @@ -1337,7 +1337,12 @@ export interface WithLifecycle extends Object { * every connection (upgraded sockets included) to close or the drain to * end. */ - onServerDrainStart?: () => Promise | void; + onServerDrainStart?: (info?: ServerDrainInfo) => Promise | void; +} + +export interface ServerDrainInfo { + /** The longest the drain runs before teardown, when known. */ + drainMs?: number; } export interface WithCostsReporting extends WithLifecycle { diff --git a/src/backend/util/gracefulShutdown.test.ts b/src/backend/util/gracefulShutdown.test.ts index 339df2b2b..05cc8fb47 100644 --- a/src/backend/util/gracefulShutdown.test.ts +++ b/src/backend/util/gracefulShutdown.test.ts @@ -25,6 +25,7 @@ import { import { registerTelemetryShutdown } from './telemetryShutdown.js'; let events: string[] = []; +let drainInfo: unknown[] = []; let unregisterTelemetry: (() => void) | null = null; const makeTarget = (opts?: { @@ -32,8 +33,9 @@ const makeTarget = (opts?: { shutdownFails?: boolean; shutdownHangs?: boolean; }): ShutdownTarget => ({ - prepareShutdown: async () => { + prepareShutdown: async (info) => { events.push('prepare'); + drainInfo.push(info); if (opts?.prepareFails) throw new Error('prepare failed'); }, shutdown: async () => { @@ -49,6 +51,7 @@ beforeEach(() => { vi.spyOn(console, 'warn').mockImplementation(() => {}); vi.spyOn(console, 'error').mockImplementation(() => {}); events = []; + drainInfo = []; unregisterTelemetry = registerTelemetryShutdown(async () => { events.push('telemetry'); }); @@ -77,6 +80,7 @@ describe('createGracefulShutdown', () => { await vi.advanceTimersByTimeAsync(1); await done; expect(events).toEqual(['prepare', 'shutdown', 'telemetry', 'exit:0']); + expect(drainInfo).toEqual([{ drainMs: 90_000 }]); }); it('SIGINT with a drain function returning 0 for SIGINT exits without waiting', async () => { @@ -89,6 +93,7 @@ describe('createGracefulShutdown', () => { await trigger('SIGINT'); expect(events).toEqual(['prepare', 'shutdown', 'telemetry', 'exit:0']); + expect(drainInfo).toEqual([{ drainMs: 0 }]); }); it('runs prepare/shutdown/exit exactly once across repeat signals during the drain', async () => { diff --git a/src/backend/util/gracefulShutdown.ts b/src/backend/util/gracefulShutdown.ts index 234cbf7de..b65d41fb5 100644 --- a/src/backend/util/gracefulShutdown.ts +++ b/src/backend/util/gracefulShutdown.ts @@ -21,7 +21,7 @@ import { shutdownTelemetry } from './telemetryShutdown.js'; /** What a graceful shutdown drives; `PuterServer` satisfies it. */ export interface ShutdownTarget { - prepareShutdown(): Promise; + prepareShutdown(info?: { drainMs?: number }): Promise; shutdown(): Promise; } @@ -80,7 +80,7 @@ export const createGracefulShutdown = ( }, drain + hardLimitMs); let exitCode = 0; try { - await target.prepareShutdown(); + await target.prepareShutdown({ drainMs: drain }); } catch (err) { console.error('[shutdown] prepareShutdown failed', err); } diff --git a/src/docs/src/AI/convo.md b/src/docs/src/AI/convo.md index bc8a0fb7e..14bb54809 100644 --- a/src/docs/src/AI/convo.md +++ b/src/docs/src/AI/convo.md @@ -54,7 +54,13 @@ 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. 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. +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 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. + +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. + +The conversation makes up to ten attempts over about a minute. In a browser that reports being offline, it first waits up to two minutes for the connection to come back. If every attempt fails, or the server refuses with an error 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. @@ -65,7 +71,7 @@ The conversation emits these events alongside native events. | Event | Fields | | --- | --- | | `ready` | `model`, `provider`, and `reconnected: true` after a reconnection | -| `reconnecting` | `code`, `reason` of the dropped connection | +| `reconnecting` | `code`, `reason` of the dropped connection. After a `puter.reconnect` event, `code` is 1012. | | `text` | `delta`, `responseId` | | `transcript` | `role`, `text`, `final`. Assistant deltas use `final: false`. Completed assistant and user transcripts use `final: true`. User transcripts require transcription to be configured through native session settings. | | `audio` | `pcm`, `sampleRate`, `responseId`, `itemId`, `contentIndex`. The PCM chunk is an `Int16Array`. | @@ -94,7 +100,7 @@ Both start methods require a connected session. Capture and playback are indepen Failures use objects containing `code` and `message`. Preparation can reject with `invalid_argument`, `model_not_found`, `auth_required`, `auth_canceled`, or `auth_failed`. Connection can reject with `subscription_required` or `insufficient_funds`, which also show the Puter upgrade prompt. Connection and asynchronous operations can reject with provider errors, `session_closed`, `connection_lost`, `operation_timeout`, or `operation_in_progress`. Sending before readiness, or while reconnecting, throws `session_not_ready`. -Invalid input throws `invalid_argument`, `invalid_event`, or `input_too_large`. A full SDK send queue throws `queue_full`. Media helpers report `unsupported_environment`, `permission_denied`, `media_error`, `media_timeout`, or `queue_full`. Runtime media errors arrive through the `error` event. Provider errors also arrive as native `error` events with a nested `error` object. +Invalid input throws `invalid_argument`, `invalid_event`, or `input_too_large`. A full SDK send queue throws `queue_full`. Media helpers report `unsupported_environment`, `permission_denied`, `media_error`, `media_timeout`, or `queue_full`. Runtime media errors and `context_restore_failed` arrive through the `error` event. Provider errors also arrive as native `error` events with a nested `error` object. Each native event is capped at 64 KiB. The SDK bounds JSON traversal, outgoing queues, acknowledgment waits, microphone work, and playback. All numeric limits are listed on [rate limits and quotas](/rate-limits-and-quotas/). diff --git a/src/docs/src/AI/realtime.md b/src/docs/src/AI/realtime.md index 3ad6edf93..3f3e02bc4 100644 --- a/src/docs/src/AI/realtime.md +++ b/src/docs/src/AI/realtime.md @@ -88,6 +88,23 @@ To interrupt, stop local playback immediately and send `response.cancel` with th Connection loss ends the session. Reconnecting creates a new conversation. Tool execution and automatic reconnection belong to your client. +## Server restarts + +When Puter knows ahead of time that the server a session runs on will restart, it sends each open session this event, at most once a minute: + +```json +{ + "type": "puter.reconnect", + "event_id": "event_puter_1767225600000", + "reason": "server_restarting", + "close_in_ms": 90000 +} +``` + +The `reason` field is `server_restarting`. The `close_in_ms` field, when present, is about how long until the server closes the session itself. The session keeps working until then. Open a new connection after a short, randomized delay, and once it gets `session.created` and your client is idle, with no response generating and no user turn in progress, move to it and close the old one. A connection that reaches a server that is also restarting gets a `server_restarting` error and closes with 1012. In that case keep the old session and try again after a short, randomized delay. If the server closes the old session first, the close code is 1012, and your client reconnects the same way. A restart can also drop the connection without any notice, so treat an unexpected close the same way. + +A new connection starts an empty conversation. To keep context, add earlier items back with `conversation.item.create` before you request a response. Event types that start with `puter.` come from Puter, not the provider. + ## Errors, billing, and limits API errors arrive as `{ "type": "error", "error": { "code": "...", "message": "..." } }`. Handle socket errors and close frames too. Authentication failures use `invalid_api_key`. Other relay codes include `model_not_found`, `insufficient_funds`, `subscription_required`, `rate_limit_exceeded`, `invalid_event`, `connection_timeout`, `provider_unavailable`, `upstream_error`, and `server_restarting`. diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index ce70e09f5..cd7d591ea 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -113,6 +113,7 @@ Realtime sessions ([`puter.ai.convo()`](/AI/convo/), `/puterai/openai/v1/realtim | Credential cache and recheck | Cache 60 seconds, sweep every 15 seconds, at most 16 checks in parallel | Invalid credentials close with 1008. Overlapping sweeps are skipped. | | Session attempts per credential | 60 per minute | Error `rate_limit_exceeded`, close 1008. | | 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. | +| 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. | | Billing a response after the client disconnects | 5 seconds | The response is cancelled and billed for what it generated, including one that was requested but starts only after the disconnect. | @@ -136,7 +137,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 | 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. | +| 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. | +| 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. | | 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 e49f43d5a..d6fe10db7 100644 --- a/src/puter-js/src/modules/ai/convo.js +++ b/src/puter-js/src/modules/ai/convo.js @@ -3,6 +3,7 @@ import { promptIfUpgradeRequired } from '../../lib/upgradePrompt.js'; import { RealtimeSession, createRealtimeSocket } from './convoTransport.js'; import { RealtimeEvents } from './realtimeEvents.js'; import { ConvoMicrophone, ConvoSpeaker } from './convoMedia.js'; +import { ConvoHistory } from './convoHistory.js'; /** @typedef {import('./types.js').ConvoOptions} ConvoOptions */ /** @typedef {import('./types.js').ConvoSessionOptions} ConvoSessionOptions */ @@ -10,6 +11,22 @@ import { ConvoMicrophone, ConvoSpeaker } from './convoMedia.js'; /** @typedef {import('./types.js').RealtimeServerEvent} RealtimeServerEvent */ /** @typedef {import('./types.js').RealtimeClientEvent} RealtimeClientEvent */ /** @typedef {{ code: string; message: string }} ConvoError */ +/** + * A pending reconnect notice. A replacement session opens right away and takes + * over once the conversation is idle. + * + * @typedef {{ + * reason: string; + * timer: ReturnType | null; + * retry: ReturnType | null; + * failures: number; + * refused: boolean; + * probe: RealtimeSession | null; + * created: RealtimeServerEvent | null; + * stopProbe: (() => void) | null; + * }} ConvoHandover + * @internal + */ const MODELS = new Set([ 'gpt-realtime-2.1-mini', @@ -39,8 +56,30 @@ const RETRYABLE_ERRORS = new Set([ 'rate_limit_exceeded', 'internal_error', ]); -const RECONNECT_ATTEMPTS = 5; +// Ten attempts over about a minute, long enough for a server restart. +const RECONNECT_ATTEMPTS = 10; const RECONNECT_DELAY_MS = 500; +const RECONNECT_MAX_DELAY_MS = 10_000; +// Per reconnect, the longest wait for a browser that reports being offline. +const OFFLINE_WAIT_MS = 120_000; +// RFC 6455 "Service Restart". +const CLOSE_RESTART = 1012; +// The server's request to reconnect once the conversation is idle. +const RECONNECT_NOTICE = 'puter.reconnect'; +// Context restored after a reconnect. The byte cap stays under the server's +// 64 KiB upload burst, so the replay never trips its rate limit. +const MAX_REPLAY_ITEMS = 40; +const MAX_REPLAY_BYTES = 48 * 1024; +// A restart notice opens the replacement session within this long, spread so +// sessions notified together don't open together. +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; +const TRANSCRIPT_TIMEOUT_MS = 30_000; +// Bounds for the idle check's bookkeeping. +const MAX_OWED = 64; +const MAX_TRACKED_INPUTS = 16; const encoder = new TextEncoder(); /** @@ -244,13 +283,9 @@ export class Conversation extends RealtimeEvents { #sessionState = {}; /** @type {string | undefined} */ #acknowledgedUpdate; - /** - * @type {{ - * timer: ReturnType; - * resolve: (value?: unknown) => void; - * } | null} - */ - #reconnectDelay = null; + // Ends the reconnect loop's current wait early. + /** @type {(() => void) | null} */ + #reconnectWait = null; #reconnecting = false; /** * @type {{ @@ -272,6 +307,37 @@ export class Conversation extends RealtimeEvents { */ #interruption = null; #cancelled = new Set(); + #history = new ConvoHistory((id) => this.#cancelled.has(id)); + /** @type {ConvoHandover | null} */ + #handover = null; + // Speech started and its audio isn't committed or cleared yet. + #speaking = false; + // Audio appended in manual turn mode and not yet committed or cleared. + #bufferedAudio = false; + #autoResponse = true; + #transcribes = false; + /** + * What the provider still owes before a switch, each until a timeout: + * `reply` to a turn server turn detection committed, `transcript:` + * per committed item, and `item:` per item the app created. + * + * @type {Map< + * string, + * { + * timer: ReturnType; + * item?: { type: unknown; role: unknown; callId: unknown }; + * } + * >} + */ + #owed = new Map(); + // Items added by committing audio, which no `conversation.item.create` + // event is waiting for. + /** @type {Set} */ + #committedItems = new Set(); + // Replayed `conversation.item.create` event IDs on the current session. + /** @type {Map} */ + #replayEvents = new Map(); + #replayReported = false; /** @type {ConvoMicrophone} */ microphone; /** @type {ConvoSpeaker} */ @@ -334,6 +400,7 @@ export class Conversation extends RealtimeEvents { model: this.model, provider: 'openai', }); + this.#scheduleHandover(); } catch (error) { this.#finish( this.#lastClose ?? { @@ -352,33 +419,48 @@ 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. + * the SDK options alone, and restore earlier conversation items. + * @param {{ + * session: RealtimeSession; + * created: RealtimeServerEvent; + * } | null} [prepared] + * A session already past `session.created`. * @returns {Promise} */ - async #open(reapply = false) { - let socket; - try { - socket = createRealtimeSocket(this.#puter, this.model); - } catch { - throw failure( - 'unsupported_environment', - 'Realtime conversations require WebSocket support', + async #open(reapply = false, prepared = null) { + if (prepared) { + this.#session = prepared.session; + this.#unsubscribe = this.#session.on('*', (event) => + this.#receive(event), + ); + this.#receive(prepared.created); + } else { + let socket; + try { + socket = createRealtimeSocket(this.#puter, this.model); + } catch { + throw failure( + 'unsupported_environment', + 'Realtime conversations require WebSocket support', + ); + } + this.#session = new RealtimeSession(socket); + this.#unsubscribe = this.#session.on('*', (event) => + this.#receive(event), + ); + await this.#wait( + 'session.created', + undefined, + undefined, + true, + SESSION_TIMEOUT_MS, ); } - this.#session = new RealtimeSession(socket); - this.#unsubscribe = this.#session.on('*', (event) => - this.#receive(event), - ); - await this.#wait( - 'session.created', - undefined, - undefined, - true, - SESSION_TIMEOUT_MS, - ); + const ackMs = prepared ? REPLACEMENT_ACK_MS : OPERATION_TIMEOUT_MS; if (reapply && Object.keys(this.#sessionState).length) - await this.#reapplySession(); - else await this.#configure(this.#options, true); + await this.#reapplySession(ackMs); + else await this.#configure(this.#options, true, ackMs); + if (reapply) this.#replayHistory(); if (this.#state === 'closed') throw failure('session_closed', 'Conversation is closed'); } @@ -387,9 +469,10 @@ export class Conversation extends RealtimeEvents { * Sends the acknowledged session state, split by top-level setting so each * update stays under the message limit. * + * @param {number} ackMs * @returns {Promise} */ - async #reapplySession() { + async #reapplySession(ackMs) { /** @type {Record[]} */ const updates = []; /** @type {Record} */ @@ -413,20 +496,79 @@ export class Conversation extends RealtimeEvents { 'session.updated', { type: 'session.update', event_id: eventId, session: update }, () => this.#acknowledgedUpdate === eventId, + ackMs, ); } } /** - * Replaces a dropped connection, keeping settings, listeners and media. The - * provider session restarts, so earlier conversation items are gone. + * Sends the newest recreatable conversation items to a new session. Not + * acknowledged: the provider handles them before anything sent later, and a + * refusal is reported without failing the reconnect. + * + * @returns {void} + */ + #replayHistory() { + this.#replayEvents.clear(); + this.#replayReported = false; + this.#history.replayed([]); + const entries = this.#history.replay( + MAX_REPLAY_ITEMS, + MAX_REPLAY_BYTES, + ); + if (!entries.length) return; + const events = entries.map(({ item }) => ({ + type: 'conversation.item.create', + event_id: this.#eventId(), + item, + })); + try { + this.#write(events); + } catch (error) { + if (error?.code === 'session_closed') throw error; + this.#replayFailed(error?.message); + return; + } + events.forEach((event, i) => + this.#replayEvents.set(event.event_id, entries[i]), + ); + this.#history.replayed(entries); + } + + /** + * @param {string} [detail] + * @returns {void} + */ + #replayFailed(detail) { + if (this.#replayReported) return; + this.#replayReported = true; + this.reportMediaError( + failure( + 'context_restore_failed', + `Earlier conversation items could not be restored${detail ? `: ${detail}` : ''}`, + ), + ); + } + + /** + * Replaces a dropped connection, or one the server asked the client to + * leave, keeping settings, listeners, media and recent conversation items. * * @param {{ code: number; reason: string }} event + * @param {{ + * session: RealtimeSession; + * created: RealtimeServerEvent; + * } | null} [prepared] + * A replacement session already open, used for the first attempt. * @returns {Promise} */ - async #reconnect(event) { + async #reconnect(event, prepared = null) { this.#state = 'connecting'; this.#reconnecting = true; + this.#endHandover(); + // Playback stops here, so only what was heard is restored. + for (const { itemId, playedMs } of this.speaker.unheard()) + this.#history.truncate(itemId, playedMs); this.#detach(); this.#resetTurnState(); this.#rejectPending( @@ -437,19 +579,26 @@ export class Conversation extends RealtimeEvents { code: event.code, reason: event.reason, }); + const offlineUntil = Date.now() + OFFLINE_WAIT_MS; for (let attempt = 0; attempt < RECONNECT_ATTEMPTS; 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; + const ready = attempt === 0 ? prepared : null; + if (ready && this.#state === 'closed') { + if (ready.session.socket.readyState < 2) ready.session.close(); + 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); + if (this.#state === 'closed') return; + } try { - await this.#open(true); + await this.#open(true, ready); this.#state = 'connected'; this.#reconnecting = false; for (const waiter of this.#reconnectWaiters.splice(0)) @@ -460,10 +609,18 @@ export class Conversation extends RealtimeEvents { provider: 'openai', reconnected: true, }); + this.#scheduleHandover(); return; } catch (error) { - if (this.#state === 'closed') return; + if (this.#state === 'closed') { + this.#detach(); + return; + } + // Nothing from a session that failed setup carries over, such + // as its notice or an unacknowledged settings update. + this.#endHandover(); this.#detach(); + this.#resetTurnState(); this.#rejectPending( failure('connection_lost', 'Connection lost; reconnecting'), ); @@ -480,6 +637,242 @@ export class Conversation extends RealtimeEvents { this.#finish({ type: 'close', code: event.code, reason: event.reason }); } + /** + * @param {number} ms + * @returns {Promise} + */ + #pause(ms) { + return new Promise((resolve) => { + const done = () => { + clearTimeout(timer); + this.#reconnectWait = null; + resolve(); + }; + const timer = setTimeout(done, ms); + this.#reconnectWait = done; + }); + } + + /** + * Waits, until `deadline` at most, while the browser reports being offline, + * so attempts aren't spent on a connection that can't open. + * + * @param {number} deadline + * @returns {Promise} + */ + #whileOffline(deadline) { + const scope = /** @type {any} */ (globalThis); + const remaining = deadline - Date.now(); + if ( + typeof navigator === 'undefined' || + navigator.onLine !== false || + typeof scope.addEventListener !== 'function' || + remaining <= 0 + ) + return Promise.resolve(); + return new Promise((resolve) => { + const done = () => { + clearTimeout(timer); + scope.removeEventListener('online', done); + this.#reconnectWait = null; + resolve(); + }; + const timer = setTimeout(done, remaining); + scope.addEventListener('online', done); + this.#reconnectWait = done; + }); + } + + /** + * Whether the conversation can move to a new session without cutting + * anything off: no reply requested, generating or playing, no user turn or + * transcript in progress, no tool output awaited and nothing waiting on the + * provider. + * + * @returns {boolean} + */ + #idle() { + return ( + this.#state === 'connected' && + !this.#activeResponse && + !this.#requestedResponses.size && + !this.#pending.size && + !this.#interruption && + !this.#configuring && + !this.#sessionUpdates.length && + !this.#owed.size && + !this.#speaking && + !this.#bufferedAudio && + !this.speaker.playing && + !this.#history.awaitingToolOutput + ); + } + + /** + * Moves a pending reconnect notice along, after promise callbacks already + * queued have run: opens a replacement session if none is open or being + * retried, and switches to it once the conversation is idle. Until then the + * current session keeps working; if the server closes it first, the usual + * reconnect takes over and uses the replacement if it's open. + * + * @returns {void} + */ + #scheduleHandover() { + const handover = this.#handover; + if (!handover || handover.timer || this.#state !== 'connected') return; + handover.timer = setTimeout(() => { + handover.timer = null; + if (this.#handover !== handover || this.#state !== 'connected') + return; + if (!handover.probe && !handover.retry) { + if (!handover.refused) void this.#openReplacement(handover); + return; + } + if (!handover.created || !this.#idle()) return; + const replacement = this.#takeReplacement(); + if (replacement) + void this.#reconnect( + { code: CLOSE_RESTART, reason: handover.reason }, + replacement, + ); + else this.#replacementFailed(handover, 'session_closed'); + }, 0); + } + + /** + * Opens a replacement session and holds it once the provider creates it. A + * refusal more attempts can fix, such as from a server that is itself + * restarting, is retried with backoff while the current session keeps + * working; any other refusal ends the attempts. + * + * @param {ConvoHandover} handover + * @returns {Promise} + */ + async #openReplacement(handover) { + let session; + try { + session = new RealtimeSession( + createRealtimeSocket(this.#puter, this.model), + ); + } catch { + return; + } + handover.probe = session; + /** @type {{ created?: RealtimeServerEvent; code?: string }} */ + const result = await new Promise((resolve) => { + const done = ( + /** @type {{ created?: RealtimeServerEvent; code?: string }} */ value, + ) => { + clearTimeout(timer); + off(); + handover.stopProbe = null; + resolve(value); + }; + const timer = setTimeout( + () => done({ code: 'operation_timeout' }), + SESSION_TIMEOUT_MS, + ); + const off = session.on('*', (event) => { + if (event.type === 'session.created') done({ created: event }); + else if (event.type === 'error') + done({ + code: String(event.error?.code ?? 'upstream_error'), + }); + else if (event.type === 'close') + done({ code: 'session_closed' }); + }); + handover.stopProbe = () => done({}); + }); + if (this.#handover !== handover) { + if (session.socket.readyState < 2) session.close(); + return; + } + if (!result.created) { + this.#replacementFailed(handover, result.code); + return; + } + handover.created = result.created; + // While held, a close, an error or the new server's own notice means + // the replacement can't be used. + handover.stopProbe = session.on('*', (event) => { + if ( + handover.probe === session && + ['close', 'error', RECONNECT_NOTICE].includes(event.type) + ) + this.#replacementFailed(handover, 'session_closed'); + }); + this.#scheduleHandover(); + } + + /** + * @param {ConvoHandover} handover + * @param {string} [code] + * @returns {void} + */ + #replacementFailed(handover, code) { + const session = handover.probe; + handover.stopProbe?.(); + handover.probe = null; + handover.created = null; + handover.stopProbe = null; + if (session && session.socket.readyState < 2) session.close(); + if (code && !RETRYABLE_ERRORS.has(code)) { + handover.refused = true; + return; + } + const delay = Math.min( + RECONNECT_DELAY_MS * 2 ** ++handover.failures, + RECONNECT_MAX_DELAY_MS, + ); + handover.retry = setTimeout( + () => { + handover.retry = null; + this.#scheduleHandover(); + }, + delay * (0.5 + Math.random()), + ); + } + + /** + * The held replacement session, if it's open, detached from the notice. + * + * @returns {{ + * session: RealtimeSession; + * created: RealtimeServerEvent; + * } | null} + */ + #takeReplacement() { + const handover = this.#handover; + const session = handover?.probe; + if (!handover?.created || !session || session.socket.readyState !== 1) + return null; + handover.stopProbe?.(); + handover.stopProbe = null; + handover.probe = null; + return { session, created: handover.created }; + } + + /** @returns {void} */ + #endHandover() { + const handover = this.#handover; + if (!handover) return; + this.#handover = null; + if (handover.timer) clearTimeout(handover.timer); + if (handover.retry) clearTimeout(handover.retry); + handover.stopProbe?.(); + const session = handover.probe; + handover.probe = null; + if (session && session.socket.readyState < 2) session.close(); + } + + /** + * @returns {void} + * @internal + */ + playbackEnded() { + this.#scheduleHandover(); + } + /** @returns {void} */ #detach() { this.#unsubscribe?.(); @@ -497,6 +890,10 @@ export class Conversation extends RealtimeEvents { this.#requestedResponses.clear(); this.#interruption = null; this.#cancelled.clear(); + this.#speaking = false; + this.#bufferedAudio = false; + this.#settleAll(); + this.#committedItems.clear(); this.speaker.clearPlayback(); } @@ -544,9 +941,10 @@ export class Conversation extends RealtimeEvents { * * @param {ConvoSessionOptions} options * @param {boolean} [full] + * @param {number} [ackMs] * @returns {Promise} */ - async #configure(options, full = false) { + async #configure(options, full = false, ackMs = OPERATION_TIMEOUT_MS) { if (this.#configuring) throw failure( 'operation_in_progress', @@ -592,6 +990,7 @@ export class Conversation extends RealtimeEvents { 'session.updated', { type: 'session.update', event_id: eventId, session }, () => this.#acknowledgedUpdate === eventId, + ackMs, ); this.#options = next; } finally { @@ -811,10 +1210,10 @@ export class Conversation extends RealtimeEvents { if (this.#state === 'closed') return; this.#state = 'closed'; this.#reconnecting = false; - if (this.#reconnectDelay) { - clearTimeout(this.#reconnectDelay.timer); - this.#reconnectDelay.resolve(); - } + this.#reconnectWait?.(); + this.#endHandover(); + this.#replayEvents.clear(); + this.#settleAll(); this.microphone.stop(); void this.speaker.stop().catch(() => {}); this.#rejectPending( @@ -890,12 +1289,45 @@ export class Conversation extends RealtimeEvents { ), ); } else if (RECONNECT_CODES.has(event.code)) { - void this.#reconnect(event); + void this.#reconnect(event, this.#takeReplacement()); } else { this.#finish(event); } return; } + if (event.type === 'error') { + const entry = this.#replayEvents.get(event.error?.event_id); + if (entry) { + this.#replayEvents.delete(event.error.event_id); + this.#history.refused(entry); + this.#replayFailed(event.error?.message); + return; + } + } + if (event.type === RECONNECT_NOTICE && !this.#handover) { + /** @type {ConvoHandover} */ + const handover = { + reason: + typeof event.reason === 'string' && event.reason + ? event.reason + : 'server_restarting', + timer: null, + retry: null, + failures: 0, + refused: false, + probe: null, + created: null, + stopProbe: null, + }; + handover.retry = setTimeout(() => { + handover.retry = null; + this.#scheduleHandover(); + }, Math.random() * REPLACEMENT_SPREAD_MS); + this.#handover = handover; + } + // The new session's copies of restored items stay internal. + const restored = this.#history.observe(event); + this.#trackTurn(event, restored); if (event.type === 'response.created') { this.#activeResponse = event.response?.id ?? ''; const [requested] = this.#requestedResponses; @@ -915,8 +1347,13 @@ export class Conversation extends RealtimeEvents { event.type === 'session.updated' ) { const input = event.session?.audio?.input; - if (input && 'turn_detection' in input) + if (input && 'turn_detection' in input) { this.#turnDetection = input.turn_detection != null; + this.#autoResponse = + input.turn_detection?.create_response !== false; + } + if (input && 'transcription' in input) + this.#transcribes = input.transcription != null; } if (event.type === 'error') { const failed = event.error?.event_id; @@ -925,6 +1362,7 @@ export class Conversation extends RealtimeEvents { const index = this.#sessionUpdates.indexOf(failed); if (index >= 0) this.#sessionUpdates.splice(index, 1); this.#sentSessions.delete(failed); + this.#settle(`item:${failed}`); } this.#promptUpgrade(event.error); } @@ -1057,7 +1495,114 @@ export class Conversation extends RealtimeEvents { usage: event.response?.usage, status: event.response?.status, }); - this.emitRealtime(event); + if (!restored) this.emitRealtime(event); + this.#scheduleHandover(); + } + + /** + * Follows the state the idle check needs: user turns, transcriptions and + * the provider adding items the app created. + * + * @param {RealtimeServerEvent} event + * @param {boolean} restored The event is about a restored item. + * @returns {void} + */ + #trackTurn(event, restored) { + const itemId = typeof event.item_id === 'string' ? event.item_id : ''; + switch (event.type) { + case 'input_audio_buffer.speech_started': + this.#speaking = true; + break; + case 'input_audio_buffer.committed': + this.#speaking = false; + this.#bufferedAudio = false; + // Server turn detection answers a committed turn on its own. + if (this.#turnDetection && this.#autoResponse) + this.#expect('reply', OPERATION_TIMEOUT_MS); + if (!itemId) break; + if (this.#transcribes) + this.#expect(`transcript:${itemId}`, TRANSCRIPT_TIMEOUT_MS); + this.#committedItems.add(itemId); + if (this.#committedItems.size > MAX_TRACKED_INPUTS) + this.#committedItems.delete( + this.#committedItems.values().next().value, + ); + break; + case 'input_audio_buffer.cleared': + this.#speaking = false; + this.#bufferedAudio = false; + this.#settle('reply'); + break; + case 'response.created': + this.#settle('reply'); + break; + case 'conversation.item.input_audio_transcription.completed': + case 'conversation.item.input_audio_transcription.failed': + case 'conversation.item.deleted': + this.#settle(`transcript:${itemId}`); + break; + case 'conversation.item.added': { + // Items the model generates arrive in progress. + const item = event.item; + if ( + restored || + item?.status === 'in_progress' || + this.#committedItems.delete(item?.id) + ) + break; + for (const [key, entry] of this.#owed) { + const sent = entry.item; + if ( + sent && + sent.type === item?.type && + (sent.role === undefined || sent.role === item?.role) && + (sent.callId === undefined || + sent.callId === item?.call_id) + ) { + this.#settle(key); + break; + } + } + break; + } + } + } + + /** + * Holds off a switch until `key` is settled or `timeoutMs` passes. + * + * @param {string} key + * @param {number} timeoutMs + * @param {{ type: unknown; role: unknown; callId: unknown }} [item] The + * item an `item:` key waits to see added. + * @returns {void} + */ + #expect(key, timeoutMs, item) { + this.#settle(key); + const timer = setTimeout(() => { + this.#owed.delete(key); + this.#scheduleHandover(); + }, timeoutMs); + this.#owed.set(key, { timer, item }); + if (this.#owed.size > MAX_OWED) + this.#settle(this.#owed.keys().next().value); + } + + /** + * @param {string} key + * @returns {void} + */ + #settle(key) { + const entry = this.#owed.get(key); + if (!entry) return; + clearTimeout(entry.timer); + this.#owed.delete(key); + } + + /** @returns {void} */ + #settleAll() { + for (const { timer } of this.#owed.values()) clearTimeout(timer); + this.#owed.clear(); } /** @returns {string} */ @@ -1092,6 +1637,26 @@ export class Conversation extends RealtimeEvents { } if (event.type === 'response.create') this.#requestedResponses.add(event.event_id); + if ( + event.type === 'conversation.item.delete' && + typeof event.item_id === 'string' + ) + this.#history.forget(event.item_id); + if ( + event.type === 'input_audio_buffer.append' && + !this.#turnDetection + ) + this.#bufferedAudio = true; + // Restored items are written while connecting and aren't tracked. + if ( + event.type === 'conversation.item.create' && + this.#state === 'connected' + ) + this.#expect(`item:${event.event_id}`, OPERATION_TIMEOUT_MS, { + type: event.item?.type, + role: event.item?.role, + callId: event.item?.call_id, + }); socket.send(json); } } @@ -1133,6 +1698,7 @@ export class Conversation extends RealtimeEvents { `Timed out waiting for ${type}`, ), ); + this.#scheduleHandover(); }, timeoutMs), }; this.#pending.add(operation); @@ -1145,9 +1711,10 @@ export class Conversation extends RealtimeEvents { * @param {string} type * @param {RealtimeServerEvent} event * @param {(event: RealtimeServerEvent) => boolean} matches + * @param {number} [timeoutMs] * @returns {Promise} */ - #request(type, event, matches) { + #request(type, event, matches, timeoutMs) { if (this.#pending.size >= 8) return Promise.reject( failure( @@ -1160,7 +1727,7 @@ export class Conversation extends RealtimeEvents { } catch (error) { return Promise.reject(error); } - return this.#wait(type, matches, event.event_id); + return this.#wait(type, matches, event.event_id, false, timeoutMs); } /** * @param {ConvoError} error diff --git a/src/puter-js/src/modules/ai/convo.test.js b/src/puter-js/src/modules/ai/convo.test.js index f509ef085..a9a451f5c 100644 --- a/src/puter-js/src/modules/ai/convo.test.js +++ b/src/puter-js/src/modules/ai/convo.test.js @@ -15,12 +15,10 @@ class FakeWebSocket extends EventTarget { send(json) { this.sent.push(JSON.parse(json)); } - close(code = 1000) { + close(code = 1000, reason = '') { if (this.readyState === 3) return; this.readyState = 3; - this.dispatchEvent( - Object.assign(new Event('close'), { code, reason: '' }), - ); + this.dispatchEvent(Object.assign(new Event('close'), { code, reason })); } deliver(event) { this.readyState = 1; @@ -228,8 +226,10 @@ describe('Conversation lifecycle', () => { const dClosed = vi.fn(); d.on('close', dClosed); socket().close(1011); - for (let i = 0; i < 5; i++) { - await vi.advanceTimersByTimeAsync(500 * 2 ** i); + // Ten attempts, backing off to 10 seconds: about a minute in all. + for (let i = 0; i < 10; i++) { + await vi.advanceTimersByTimeAsync(Math.min(500 * 2 ** i, 10_000)); + expect(d.state).toBe('connecting'); socket().close(1006); } await vi.advanceTimersByTimeAsync(0); @@ -749,3 +749,1231 @@ it('preflights both events of a text turn before sending either', async () => { ); expect(socket().sent).toHaveLength(before); }); + +// -- Restart notice, context replay and retries -- + +const notice = { type: 'puter.reconnect', reason: 'server_restarting' }; +const userText = (id, text, previous = null) => ({ + type: 'conversation.item.added', + previous_item_id: previous, + item: { + id, + object: 'realtime.item', + type: 'message', + role: 'user', + status: 'completed', + content: [{ type: 'input_text', text }], + }, +}); +/** PCM16 24 kHz base64 for `ms` of silence. */ +const pcmFor = (ms) => 'A'.repeat(((ms * 48) / 3) * 4); +/** Streams a whole assistant message the way OpenAI does. */ +const reply = ( + ws, + { + responseId, + itemId, + text, + audioMs = 0, + status = 'completed', + finish = true, + }, +) => { + const audio = audioMs > 0; + ws.deliver({ type: 'response.created', response: { id: responseId } }); + ws.deliver({ + type: 'response.output_item.added', + response_id: responseId, + item: { + id: itemId, + type: 'message', + role: 'assistant', + status: 'in_progress', + content: [], + }, + }); + if (audio) + ws.deliver({ + type: 'response.output_audio.delta', + response_id: responseId, + item_id: itemId, + content_index: 0, + delta: pcmFor(audioMs), + }); + ws.deliver({ + type: audio + ? 'response.output_audio_transcript.delta' + : 'response.output_text.delta', + response_id: responseId, + item_id: itemId, + content_index: 0, + delta: text, + }); + if (!finish) return; + ws.deliver({ + type: 'response.output_item.done', + response_id: responseId, + item: { + id: itemId, + type: 'message', + role: 'assistant', + status, + content: [ + audio + ? { type: 'output_audio', transcript: text } + : { type: 'output_text', text }, + ], + }, + }); + ws.deliver({ + type: 'response.done', + response: { + id: responseId, + status: status === 'completed' ? 'completed' : 'cancelled', + }, + }); +}; +/** A function call reply followed by the app's output for it. */ +const toolTurn = (c, ws, { responseId, callId, output = '42' }) => { + ws.deliver({ type: 'response.created', response: { id: responseId } }); + const call = { + id: `fc_${callId}`, + type: 'function_call', + call_id: callId, + name: 'lookup', + arguments: '{"q":1}', + }; + ws.deliver({ + type: 'response.output_item.done', + response_id: responseId, + item: { ...call, status: 'completed' }, + }); + ws.deliver({ type: 'response.done', response: { id: responseId } }); + if (output === null) return; + c.send({ + type: 'conversation.item.create', + item: { type: 'function_call_output', call_id: callId, output }, + }); + ws.deliver({ + type: 'conversation.item.added', + item: { + id: `out_${callId}`, + type: 'function_call_output', + call_id: callId, + output, + status: 'completed', + }, + }); +}; +/** Answers the replacement socket's handshake once the backoff elapses. */ +const answerReconnect = async (delay = 500) => { + const before = socket(); + await vi.advanceTimersByTimeAsync(delay); + expect(socket()).not.toBe(before); + socket().deliver({ type: 'session.created', session: {} }); + await vi.advanceTimersByTimeAsync(0); + socket().deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(0); +}; +const replayed = () => + socket() + .sent.filter((event) => event.type === 'conversation.item.create') + .map((event) => event.item); +const track = (c) => { + const events = []; + c.on('*', (event) => events.push(event)); + return { + events, + types: () => events.map((event) => event.type), + errors: () => events.filter((event) => event.type === 'error'), + }; +}; +const fakeTime = () => { + vi.useFakeTimers(); + vi.spyOn(Math, 'random').mockReturnValue(0.5); +}; + +// Longest spread before a notice opens its replacement, plus the tick a timer +// scheduled during a tick waits. +const SPREAD = 3_001; +/** Answers the replacement session a notice opens. */ +const openReplacement = async () => { + await vi.advanceTimersByTimeAsync(SPREAD); + const replacement = socket(); + replacement.deliver({ type: 'session.created', session: {} }); + await vi.advanceTimersByTimeAsync(0); + return replacement; +}; +/** Acknowledges the settings a replacement session was sent. */ +const settle = async (ws) => { + ws.deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(0); +}; + +describe('restart notice', () => { + it('switches an idle conversation to a replacement session without an error', async () => { + const c = await connected({ instructions: 'Be brief.' }); + fakeTime(); + const log = track(c); + const first = socket(); + first.deliver({ ...notice, close_in_ms: 90_000 }); + // Opened after a randomized spread, here half of 3 seconds. + await vi.advanceTimersByTimeAsync(1_499); + expect(socket()).toBe(first); + await vi.advanceTimersByTimeAsync(2); + const replacement = socket(); + expect(replacement).not.toBe(first); + // The current session keeps working while the replacement opens. + expect(c.state).toBe('connected'); + expect(first.readyState).toBe(1); + replacement.deliver({ type: 'session.created', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(first.readyState).toBe(3); + expect(c.state).toBe('connecting'); + expect(log.events.find((e) => e.type === 'reconnecting')).toEqual({ + type: 'reconnecting', + code: 1012, + reason: 'server_restarting', + }); + expect(replacement.sent[0].session.instructions).toBe('Be brief.'); + await settle(replacement); + expect(c.state).toBe('connected'); + expect(log.events.at(-1)).toMatchObject({ + type: 'ready', + reconnected: true, + }); + expect(log.types().filter((t) => t === 'session.created')).toHaveLength( + 1, + ); + expect(log.errors()).toEqual([]); + expect(log.types()).not.toContain('close'); + expect(FakeWebSocket.instances).toHaveLength(2); + }); + + it('waits for the reply in flight to finish', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver({ type: 'response.created', response: { id: 'r1' } }); + first.deliver(notice); + await openReplacement(); + await vi.advanceTimersByTimeAsync(5_000); + expect(c.state).toBe('connected'); + expect(first.readyState).toBe(1); + first.deliver({ type: 'response.done', response: { id: 'r1' } }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + }); + + it('waits through user speech, even past a reply created mid-speech, until the turn is answered', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver({ type: 'input_audio_buffer.speech_started' }); + first.deliver(notice); + await openReplacement(); + for (const event of [ + { type: 'response.created', response: { id: 'r0' } }, + { type: 'response.done', response: { id: 'r0' } }, + { type: 'input_audio_buffer.speech_stopped' }, + // Server turn detection replies to a committed turn by itself. + { type: 'input_audio_buffer.committed', item_id: 'u1' }, + { type: 'response.created', response: { id: 'r1' } }, + ]) { + first.deliver(event); + await vi.advanceTimersByTimeAsync(1_000); + expect(c.state).toBe('connected'); + } + first.deliver({ type: 'response.done', response: { id: 'r1' } }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + }); + + it('waits for uncommitted audio in manual turn mode', async () => { + const c = await connected({ turnDetection: false }); + fakeTime(); + const first = socket(); + c.appendAudio(new Int16Array(480)); + first.deliver(notice); + await openReplacement(); + await vi.advanceTimersByTimeAsync(1_000); + expect(c.state).toBe('connected'); + first.deliver({ type: 'input_audio_buffer.cleared' }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + }); + + it('waits for a configuration update to be acknowledged', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + const update = c.configure({ voice: 'marin' }); + first.deliver(notice); + await openReplacement(); + await vi.advanceTimersByTimeAsync(1_000); + expect(c.state).toBe('connected'); + first.deliver({ type: 'session.updated', session: {} }); + await update; + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + }); + + it('waits for native settings and items the provider has not taken yet, then keeps them', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver(notice); + // An app adding context right after connecting, without a reply. + c.send({ + type: 'session.update', + session: { type: 'realtime', tools: [{ type: 'function' }] }, + }); + c.send({ + type: 'conversation.item.create', + item: { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'context' }], + }, + }); + await openReplacement(); + await vi.advanceTimersByTimeAsync(1_000); + expect(c.state).toBe('connected'); + first.deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connected'); + // A generated item doesn't count as the app's. + first.deliver({ + type: 'conversation.item.added', + item: { + id: 'g1', + type: 'message', + role: 'assistant', + status: 'in_progress', + content: [], + }, + }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connected'); + first.deliver(userText('u1', 'context')); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + const replacement = socket(); + expect(replacement.sent[0].session.tools).toEqual([ + { type: 'function' }, + ]); + await settle(replacement); + expect(replayed().map((item) => item.content[0].text)).toEqual([ + 'context', + ]); + }); + + it('stops waiting for an item the provider refused', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver(notice); + c.send({ + type: 'conversation.item.create', + event_id: 'bad', + item: { type: 'message', role: 'user', content: [] }, + }); + await openReplacement(); + await vi.advanceTimersByTimeAsync(1_000); + expect(c.state).toBe('connected'); + first.deliver({ + type: 'error', + error: { code: 'invalid_value', message: 'No', event_id: 'bad' }, + }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + }); + + it('waits for built-in playback to end', async () => { + const c = await connected(); + fakeTime(); + const playing = vi.spyOn(c.speaker, 'playing', 'get'); + playing.mockReturnValue(true); + socket().deliver(notice); + await openReplacement(); + await vi.advanceTimersByTimeAsync(1_000); + expect(c.state).toBe('connected'); + playing.mockReturnValue(false); + c.playbackEnded(); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + }); + + it('waits for every function call output, then restores calls and outputs', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver({ type: 'response.created', response: { id: 'r1' } }); + for (const id of ['a', 'b']) + first.deliver({ + type: 'response.output_item.done', + response_id: 'r1', + item: { + id: `fc_${id}`, + type: 'function_call', + call_id: id, + name: 'lookup', + arguments: '{}', + status: 'completed', + }, + }); + first.deliver({ type: 'response.done', response: { id: 'r1' } }); + first.deliver(notice); + await openReplacement(); + const answer = (id) => { + c.send({ + type: 'conversation.item.create', + item: { type: 'function_call_output', call_id: id, output: id }, + }); + first.deliver({ + type: 'conversation.item.added', + item: { + id: `out_${id}`, + type: 'function_call_output', + call_id: id, + output: id, + status: 'completed', + }, + }); + }; + answer('b'); + await vi.advanceTimersByTimeAsync(1_000); + expect(c.state).toBe('connected'); + answer('a'); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + await settle(socket()); + expect(replayed().map((item) => [item.type, item.call_id])).toEqual([ + ['function_call', 'a'], + ['function_call', 'b'], + ['function_call_output', 'b'], + ['function_call_output', 'a'], + ]); + }); + + it('keeps the current session while a restarting server refuses the replacement', async () => { + const c = await connected(); + fakeTime(); + const log = track(c); + const first = socket(); + first.deliver(notice); + await vi.advanceTimersByTimeAsync(SPREAD); + const refuse = () => { + socket().deliver({ + type: 'error', + error: { code: 'server_restarting', message: 'Restarting' }, + }); + socket().close(1012, 'server_restarting'); + }; + refuse(); + // Retried with backoff from 1 second. + await vi.advanceTimersByTimeAsync(999); + expect(FakeWebSocket.instances).toHaveLength(2); + await vi.advanceTimersByTimeAsync(2); + expect(FakeWebSocket.instances).toHaveLength(3); + refuse(); + await vi.advanceTimersByTimeAsync(2_001); + expect(FakeWebSocket.instances).toHaveLength(4); + refuse(); + expect(c.state).toBe('connected'); + expect(first.readyState).toBe(1); + expect(log.types()).not.toContain('reconnecting'); + expect(log.errors()).toEqual([]); + + // The server's own close starts the usual reconnect. + first.close(1012, 'server restarting'); + await answerReconnect(); + expect(c.state).toBe('connected'); + expect(log.types().filter((t) => t === 'reconnecting')).toHaveLength(1); + await vi.advanceTimersByTimeAsync(10_000); + expect(FakeWebSocket.instances).toHaveLength(5); + }); + + it('switches to the held replacement right away when the server closes first', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver({ type: 'response.created', response: { id: 'r1' } }); + first.deliver(notice); + const replacement = await openReplacement(); + first.close(1012, 'server restarting'); + await vi.advanceTimersByTimeAsync(0); + expect(socket()).toBe(replacement); + expect(replacement.sent[0].type).toBe('session.update'); + await settle(replacement); + expect(c.state).toBe('connected'); + expect(FakeWebSocket.instances).toHaveLength(2); + }); + + it('opens another replacement when the held one closes', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver({ type: 'response.created', response: { id: 'r1' } }); + first.deliver(notice); + const replacement = await openReplacement(); + replacement.close(1006); + await vi.advanceTimersByTimeAsync(1_001); + expect(socket()).not.toBe(replacement); + const next = await openReplacement(); + first.deliver({ type: 'response.done', response: { id: 'r1' } }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + expect(next.sent[0].type).toBe('session.update'); + }); + + it('stops opening replacements after a refusal more attempts will not fix', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver(notice); + await vi.advanceTimersByTimeAsync(SPREAD); + socket().deliver({ + type: 'error', + error: { code: 'insufficient_funds', message: 'No usage left' }, + }); + socket().close(1008); + 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); + // The server's close still reconnects the usual way. + first.close(1012); + await answerReconnect(); + expect(c.state).toBe('connected'); + }); + + it('drops a held replacement that gets its own notice or an error', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver({ type: 'response.created', response: { id: 'r1' } }); + first.deliver(notice); + for (const event of [ + notice, + { type: 'error', error: { code: 'server_error', message: 'x' } }, + ]) { + const held = await openReplacement(); + held.deliver(event); + expect(held.readyState).toBe(3); + } + const held = await openReplacement(); + first.deliver({ type: 'response.done', response: { id: 'r1' } }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + expect(held.sent[0].type).toBe('session.update'); + }); + + it('gives a held replacement a short acknowledgment timeout after a drop', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver({ type: 'response.created', response: { id: 'r1' } }); + first.deliver(notice); + // The held socket died with the network but still reads as open. + const held = await openReplacement(); + first.close(1006); + await vi.advanceTimersByTimeAsync(0); + expect(held.sent[0].type).toBe('session.update'); + await vi.advanceTimersByTimeAsync(2_999); + expect(socket()).toBe(held); + await vi.advanceTimersByTimeAsync(1); + await answerReconnect(1_000); + expect(c.state).toBe('connected'); + }); + + it('close() while switching to a held replacement closes it too', async () => { + const c = await connected(); + fakeTime(); + c.on('reconnecting', () => c.close()); + socket().deliver(notice); + const held = await openReplacement(); + expect(c.state).toBe('closed'); + expect(held.readyState).toBe(3); + expect(held.sent).toEqual([]); + expect(vi.getTimerCount()).toBe(0); + }); + + it('keeps waiting for an app item when the provider adds a different one', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver(notice); + c.send({ + type: 'conversation.item.create', + item: { + type: 'function_call_output', + call_id: 'call_9', + output: '1', + }, + }); + await openReplacement(); + first.deliver({ + type: 'conversation.item.added', + item: { id: 'm1', type: 'mcp_list_tools', status: 'completed' }, + }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connected'); + first.deliver({ + type: 'conversation.item.added', + item: { + id: 'o9', + type: 'function_call_output', + call_id: 'call_9', + output: '1', + status: 'completed', + }, + }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + }); + + it('waits for a transcription, but not past its timeout', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver({ + type: 'session.updated', + session: { + audio: { + input: { + transcription: { model: 'gpt-4o-mini-transcribe' }, + turn_detection: { type: 'server_vad' }, + }, + }, + }, + }); + first.deliver({ type: 'input_audio_buffer.speech_started' }); + first.deliver({ type: 'input_audio_buffer.committed', item_id: 'u1' }); + reply(first, { responseId: 'r1', itemId: 'a1', text: 'Hi.' }); + first.deliver(notice); + await openReplacement(); + await vi.advanceTimersByTimeAsync(20_000); + expect(c.state).toBe('connected'); + // Transcription that never reports stops holding the switch. + await vi.advanceTimersByTimeAsync(10_000); + expect(c.state).toBe('connecting'); + }); + + it('waits for the reply to a committed turn, but not past its timeout', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver({ type: 'input_audio_buffer.committed', item_id: 'u1' }); + first.deliver(notice); + await openReplacement(); + await vi.advanceTimersByTimeAsync(5_000); + expect(c.state).toBe('connected'); + await vi.advanceTimersByTimeAsync(5_000); + expect(c.state).toBe('connecting'); + }); + + it('carries out a notice that arrives while connecting once ready', async () => { + fakeTime(); + const c = make(); + const ready = c.connect(); + socket().deliver({ type: 'session.created', session: {} }); + await vi.advanceTimersByTimeAsync(0); + socket().deliver(notice); + await vi.advanceTimersByTimeAsync(0); + expect(FakeWebSocket.instances).toHaveLength(1); + socket().deliver({ type: 'session.updated', session: {} }); + await ready; + const replacement = await openReplacement(); + expect(c.state).toBe('connecting'); + await settle(replacement); + expect(c.state).toBe('connected'); + }); + + it('drops a notice from a reconnect attempt that failed', async () => { + const c = await connected(); + fakeTime(); + socket().close(1006); + await vi.advanceTimersByTimeAsync(500); + socket().deliver(notice); + socket().close(1011); + await answerReconnect(1_000); + expect(c.state).toBe('connected'); + const sockets = FakeWebSocket.instances.length; + await vi.advanceTimersByTimeAsync(5_000); + expect(FakeWebSocket.instances).toHaveLength(sockets); + }); + + it('close() at any point of a handover leaves nothing running', async () => { + // While the replacement opens. + const c = await connected(); + fakeTime(); + socket().deliver(notice); + await vi.advanceTimersByTimeAsync(SPREAD); + const opening = socket(); + c.close(); + expect(opening.readyState).toBe(3); + expect(vi.getTimerCount()).toBe(0); + + // While a replacement is held. + const d = await connected(); + socket().deliver({ type: 'response.created', response: { id: 'r1' } }); + socket().deliver(notice); + const held = await openReplacement(); + d.close(); + expect(held.readyState).toBe(3); + expect(vi.getTimerCount()).toBe(0); + + // While a refused replacement waits to retry. + const e = await connected(); + socket().deliver(notice); + await vi.advanceTimersByTimeAsync(SPREAD); + socket().close(1012); + await vi.advanceTimersByTimeAsync(0); + expect(vi.getTimerCount()).toBeGreaterThan(0); + e.close(); + expect(vi.getTimerCount()).toBe(0); + const sockets = FakeWebSocket.instances.length; + await vi.advanceTimersByTimeAsync(60_000); + expect(FakeWebSocket.instances).toHaveLength(sockets); + }); +}); + +describe('context replay after a reconnect', () => { + it('restores finished items in order, before ready, after a dropped connection', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + c.sendText('What is the weather?'); + first.deliver(userText('u1', 'What is the weather?')); + reply(first, { + responseId: 'r1', + itemId: 'a1', + text: 'Sunny today.', + audioMs: 600, + }); + first.deliver({ type: 'input_audio_buffer.speech_started' }); + first.deliver({ type: 'input_audio_buffer.committed', item_id: 'u2' }); + first.deliver({ + type: 'conversation.item.added', + previous_item_id: 'a1', + item: { + id: 'u2', + type: 'message', + role: 'user', + status: 'completed', + content: [{ type: 'input_audio', transcript: null }], + }, + }); + first.deliver({ + type: 'conversation.item.input_audio_transcription.completed', + item_id: 'u2', + content_index: 0, + transcript: 'And tomorrow?', + }); + toolTurn(c, first, { responseId: 'r2', callId: 'call_1' }); + reply(first, { responseId: 'r3', itemId: 'a2', text: 'Rain.' }); + + let sentAtReady; + c.on('ready', () => { + sentAtReady = socket().sent.map((event) => event.type); + }); + first.close(1006); + await answerReconnect(); + expect(replayed()).toEqual([ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'What is the weather?' }], + }, + { + type: 'message', + role: 'assistant', + content: [{ type: 'output_text', text: 'Sunny today.' }], + }, + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'And tomorrow?' }], + }, + { + type: 'function_call', + call_id: 'call_1', + name: 'lookup', + arguments: '{"q":1}', + }, + { type: 'function_call_output', call_id: 'call_1', output: '42' }, + { + type: 'message', + role: 'assistant', + content: [{ type: 'output_text', text: 'Rain.' }], + }, + ]); + expect(sentAtReady).toEqual([ + 'session.update', + ...Array(6).fill('conversation.item.create'), + ]); + expect(c.state).toBe('connected'); + }); + + it('honors previous_item_id when the provider inserts an item mid-conversation', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + first.deliver(userText('u1', 'one')); + first.deliver(userText('u3', 'three', 'u1')); + first.deliver(userText('u2', 'two', 'u1')); + first.close(1006); + await answerReconnect(); + expect(replayed().map((item) => item.content[0].text)).toEqual([ + 'one', + 'two', + 'three', + ]); + }); + + it('leaves out what it cannot recreate faithfully', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + // Audio without a transcript, and one whose transcription failed. + for (const id of ['u1', 'u2']) + first.deliver({ + type: 'conversation.item.added', + item: { + id, + type: 'message', + role: 'user', + status: 'completed', + content: [{ type: 'input_audio', transcript: null }], + }, + }); + first.deliver({ + type: 'conversation.item.input_audio_transcription.failed', + item_id: 'u2', + content_index: 0, + error: { code: 'x' }, + }); + // An image message. + first.deliver({ + type: 'conversation.item.added', + item: { + id: 'u3', + type: 'message', + role: 'user', + status: 'completed', + content: [ + { type: 'input_text', text: 'What is this?' }, + { + type: 'input_image', + image_url: 'data:image/png;base64,', + }, + ], + }, + }); + // A function call cut off by a cancellation. + first.deliver({ + type: 'response.output_item.done', + response_id: 'r1', + item: { + id: 'fc1', + type: 'function_call', + call_id: 'call_1', + name: 'lookup', + arguments: '{"q":', + status: 'incomplete', + }, + }); + // A reply still generating when the connection drops. + reply(first, { + responseId: 'r2', + itemId: 'a1', + text: 'Half a', + finish: false, + }); + first.deliver(userText('u4', 'kept')); + first.close(1006); + await answerReconnect(); + expect(replayed()).toEqual([ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'kept' }], + }, + ]); + }); + + it('never restores items the provider or the app deleted', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + for (const id of ['u1', 'u2', 'u3']) first.deliver(userText(id, id)); + first.deliver({ type: 'conversation.item.deleted', item_id: 'u2' }); + c.send({ type: 'conversation.item.delete', item_id: 'u3' }); + // A late copy of a deleted item doesn't bring it back. + first.deliver({ + ...userText('u2', 'u2'), + type: 'conversation.item.done', + }); + first.close(1006); + await answerReconnect(); + expect(replayed().map((item) => item.content[0].text)).toEqual(['u1']); + }); + + it('cuts truncated assistant audio to the share that was heard', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + reply(first, { + responseId: 'r1', + itemId: 'a1', + text: 'one two three four', + audioMs: 1_000, + }); + first.deliver({ + type: 'conversation.item.truncated', + item_id: 'a1', + content_index: 0, + audio_end_ms: 500, + }); + reply(first, { + responseId: 'r2', + itemId: 'a2', + text: 'never heard', + audioMs: 400, + }); + first.deliver({ + type: 'conversation.item.truncated', + item_id: 'a2', + content_index: 0, + audio_end_ms: 0, + }); + first.close(1006); + await answerReconnect(); + expect(replayed()).toEqual([ + { + type: 'message', + role: 'assistant', + content: [{ type: 'output_text', text: 'one two' }], + }, + ]); + }); + + it('cuts audio still playing or queued when the connection drops', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + reply(first, { + responseId: 'r1', + itemId: 'a1', + text: 'one two three four', + audioMs: 1_000, + }); + reply(first, { + responseId: 'r2', + itemId: 'a2', + text: 'queued behind it', + audioMs: 500, + }); + vi.spyOn(c.speaker, 'unheard').mockReturnValue([ + { itemId: 'a1', contentIndex: 0, playedMs: 250 }, + { itemId: 'a2', contentIndex: 0, playedMs: 0 }, + ]); + first.close(1006); + await answerReconnect(); + expect(replayed().map((item) => item.content[0].text)).toEqual(['one']); + }); + + it('cuts a cancelled text reply to what the app was shown', async () => { + const c = await connected({ output: ['text'] }); + fakeTime(); + const first = socket(); + reply(first, { + responseId: 'r1', + itemId: 'a1', + text: 'Hello ', + finish: false, + }); + const interrupted = c.interrupt(); + first.deliver({ + type: 'response.output_text.delta', + response_id: 'r1', + item_id: 'a1', + delta: 'there', + }); + first.deliver({ + type: 'response.output_item.done', + response_id: 'r1', + item: { + id: 'a1', + type: 'message', + role: 'assistant', + status: 'incomplete', + content: [{ type: 'output_text', text: 'Hello there' }], + }, + }); + first.deliver({ + type: 'response.done', + response: { id: 'r1', status: 'cancelled' }, + }); + await interrupted; + first.close(1006); + await answerReconnect(); + expect(replayed()).toEqual([ + { + type: 'message', + role: 'assistant', + content: [{ type: 'output_text', text: 'Hello' }], + }, + ]); + }); + + it('restores only the newest items within the item and byte caps', async () => { + const c = await connected(); + fakeTime(); + let first = socket(); + for (let i = 1; i <= 50; i++) first.deliver(userText(`u${i}`, `m${i}`)); + first.close(1006); + await answerReconnect(); + const items = replayed(); + expect(items).toHaveLength(40); + expect(items[0].content[0].text).toBe('m11'); + expect(items.at(-1).content[0].text).toBe('m50'); + + const d = await connected(); + first = socket(); + for (let i = 1; i <= 20; i++) + first.deliver(userText(`v${i}`, `${i} `.padEnd(4_000, 'x'))); + first.close(1006); + await answerReconnect(); + const creates = socket().sent.filter( + (event) => event.type === 'conversation.item.create', + ); + const bytes = creates.reduce( + (sum, event) => + sum + + new TextEncoder().encode(JSON.stringify(event)).byteLength, + 0, + ); + expect(creates.length).toBe(11); + expect(bytes).toBeLessThanOrEqual(48 * 1024); + expect(creates.at(-1).item.content[0].text.startsWith('20 ')).toBe( + true, + ); + expect(d.state).toBe('connected'); + }); + + it('drops a function output whose call fell outside the window', async () => { + const c = await connected(); + fakeTime(); + const first = socket(); + toolTurn(c, first, { responseId: 'r1', callId: 'call_1' }); + for (let i = 1; i <= 39; i++) first.deliver(userText(`u${i}`, `m${i}`)); + first.close(1006); + await answerReconnect(); + const items = replayed(); + expect(items).toHaveLength(39); + expect(items.every((item) => item.type === 'message')).toBe(true); + }); + + it('reports a refused item once, stays connected and leaves it out later', async () => { + const c = await connected(); + fakeTime(); + const log = track(c); + let current = socket(); + for (const id of ['u1', 'u2', 'u3']) current.deliver(userText(id, id)); + current.close(1006); + await answerReconnect(); + current = socket(); + const ids = current.sent + .filter((event) => event.type === 'conversation.item.create') + .map((event) => event.event_id); + expect(ids).toHaveLength(3); + for (const id of ids.slice(0, 2)) + current.deliver({ + type: 'error', + error: { code: 'invalid_value', message: 'Nope', event_id: id }, + }); + // The provider's copy of the item it accepted, under a new ID. + current.deliver(userText('n3', 'u3')); + expect(log.errors()).toEqual([ + { + type: 'error', + code: 'context_restore_failed', + message: + 'Earlier conversation items could not be restored: Nope', + error: { + code: 'context_restore_failed', + message: + 'Earlier conversation items could not be restored: Nope', + }, + }, + ]); + expect(c.state).toBe('connected'); + + // Deleting the new ID removes the restored item; nothing is doubled. + current.deliver(userText('n4', 'new')); + current.deliver({ type: 'conversation.item.deleted', item_id: 'n3' }); + current.close(1006); + await answerReconnect(500); + expect(replayed().map((item) => item.content[0].text)).toEqual(['new']); + }); + + it("keeps the new session's copies of restored items from listeners", async () => { + const c = await connected(); + fakeTime(); + const log = track(c); + socket().deliver(userText('u1', 'hello')); + socket().close(1006); + await answerReconnect(); + const seen = log.events.length; + const copy = userText('n1', 'hello'); + socket().deliver(copy); + socket().deliver({ ...copy, type: 'conversation.item.done' }); + expect(log.events).toHaveLength(seen); + // Items after them are the conversation's own. + socket().deliver(userText('n2', 'hello')); + expect(log.events.slice(seen).map((e) => e.item?.id)).toEqual(['n2']); + }); + + it('keeps items whose restored copy never arrived for the next reconnect', async () => { + const c = await connected(); + fakeTime(); + socket().deliver(userText('u1', 'hello')); + socket().close(1006); + await answerReconnect(); + expect(replayed()).toHaveLength(1); + socket().close(1006); + await answerReconnect(); + expect(replayed().map((item) => item.content[0].text)).toEqual([ + 'hello', + ]); + expect(c.state).toBe('connected'); + }); + + it('reconnects even when the replay cannot be sent', async () => { + const c = await connected(); + fakeTime(); + const log = track(c); + socket().deliver(userText('u1', 'hello')); + socket().close(1006); + await vi.advanceTimersByTimeAsync(500); + socket().deliver({ type: 'session.created', session: {} }); + await vi.advanceTimersByTimeAsync(0); + socket().bufferedAmount = 192 * 1024; + socket().deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connected'); + expect(log.errors()).toMatchObject([ + { code: 'context_restore_failed' }, + ]); + expect(log.events.at(-1)).toMatchObject({ + type: 'ready', + reconnected: true, + }); + }); +}); + +describe('reconnect retries', () => { + it('retries after an attempt that dropped before its settings were acknowledged', async () => { + const c = await connected({ instructions: 'Be brief.' }); + fakeTime(); + socket().close(1006); + await vi.advanceTimersByTimeAsync(500); + socket().deliver({ type: 'session.created', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(socket().sent[0].type).toBe('session.update'); + socket().close(1006); + await answerReconnect(1_000); + expect(c.state).toBe('connected'); + }); + + it('retries a reconnect refused by a draining server', async () => { + const c = await connected(); + fakeTime(); + socket().close(1006); + await vi.advanceTimersByTimeAsync(500); + socket().deliver({ + type: 'error', + error: { + code: 'server_restarting', + message: 'Server is restarting; reconnect', + }, + }); + socket().close(1012, 'server_restarting'); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connecting'); + await answerReconnect(1_000); + expect(c.state).toBe('connected'); + }); + + describe('while the browser is offline', () => { + let target; + beforeEach(() => { + target = new EventTarget(); + vi.stubGlobal('navigator', { onLine: false }); + vi.stubGlobal( + 'addEventListener', + vi.fn((...args) => target.addEventListener(...args)), + ); + vi.stubGlobal( + 'removeEventListener', + vi.fn((...args) => target.removeEventListener(...args)), + ); + }); + + it('waits for the online event instead of spending attempts', async () => { + const c = await connected(); + fakeTime(); + socket().close(1006); + const sockets = FakeWebSocket.instances.length; + await vi.advanceTimersByTimeAsync(90_000); + expect(FakeWebSocket.instances).toHaveLength(sockets); + expect(c.state).toBe('connecting'); + navigator.onLine = true; + target.dispatchEvent(new Event('online')); + await vi.advanceTimersByTimeAsync(0); + expect(FakeWebSocket.instances).toHaveLength(sockets + 1); + socket().deliver({ type: 'session.created', session: {} }); + await vi.advanceTimersByTimeAsync(0); + socket().deliver({ type: 'session.updated', session: {} }); + await vi.advanceTimersByTimeAsync(0); + expect(c.state).toBe('connected'); + expect(removeEventListener).toHaveBeenCalledWith( + 'online', + expect.any(Function), + ); + }); + + it('gives up waiting after two minutes and tries anyway', async () => { + const c = await connected(); + fakeTime(); + socket().close(1006); + const sockets = FakeWebSocket.instances.length; + // Two minutes from the drop, backoff included. + await vi.advanceTimersByTimeAsync(119_999); + expect(FakeWebSocket.instances).toHaveLength(sockets); + await vi.advanceTimersByTimeAsync(1); + expect(FakeWebSocket.instances).toHaveLength(sockets + 1); + expect(c.state).toBe('connecting'); + }); + + it('close() stops the wait and removes its listener', async () => { + const c = await connected(); + fakeTime(); + socket().close(1006); + await vi.advanceTimersByTimeAsync(1_000); + c.close(); + expect(vi.getTimerCount()).toBe(0); + expect(removeEventListener).toHaveBeenCalledWith( + 'online', + expect.any(Function), + ); + const sockets = FakeWebSocket.instances.length; + target.dispatchEvent(new Event('online')); + await vi.advanceTimersByTimeAsync(10_000); + expect(FakeWebSocket.instances).toHaveLength(sockets); + }); + }); +}); diff --git a/src/puter-js/src/modules/ai/convoHistory.js b/src/puter-js/src/modules/ai/convoHistory.js new file mode 100644 index 000000000..2ab8b778b --- /dev/null +++ b/src/puter-js/src/modules/ai/convoHistory.js @@ -0,0 +1,524 @@ +/** @typedef {import('./types.js').RealtimeServerEvent} RealtimeServerEvent */ + +/** + * An item `conversation.item.create` accepts: a text message, a function call + * or a function call's output. + * + * @typedef {Record & { type: string }} ConvoReplayItem + */ + +/** + * @typedef {Object} HistoryRecord + * @property {string} id + * @property {'message' | 'function_call' | 'function_call_output'} type + * @property {string} [role] + * @property {string} responseId + * @property {boolean} final + * @property {boolean} unsupported A part that can't be recreated as text. + * @property {(string | null)[]} parts Text per content part; `null` for audio + * whose transcript hasn't arrived. + * @property {string} received Transcript and text deltas as they arrived. + * @property {string} shown The deltas the conversation delivered to the app. + * @property {boolean} cancelled Finished after the SDK cancelled its response. + * @property {number} audioMs Assistant audio received. + * @property {number} [playedMs] Assistant audio heard, once truncated. + * @property {string} [status] + * @property {string} [callId] + * @property {string} [name] + * @property {string} [args] + * @property {string} [output] + */ + +/** @typedef {{ record: HistoryRecord; item: ConvoReplayItem }} ReplayEntry */ + +const ITEM_EVENTS = new Set([ + 'conversation.item.added', + 'conversation.item.done', + 'response.output_item.added', + 'response.output_item.done', +]); +const DONE_EVENTS = new Set([ + 'conversation.item.done', + 'response.output_item.done', +]); +const TEXT_DELTAS = new Set([ + 'response.output_text.delta', + 'response.output_audio_transcript.delta', +]); +// PCM16 at 24 kHz mono. +const AUDIO_BYTES_PER_MS = 48; +// Twice the replay cap, leaving room for items that can't be replayed. +const MAX_RECORDS = 80; +const MAX_DELETED = 256; +const encoder = new TextEncoder(); + +/** + * @param {unknown} value + * @returns {value is Record} + */ +const isPlainObject = (value) => + value !== null && typeof value === 'object' && !Array.isArray(value); + +/** + * `text` cut to at most `limit` characters, back to the last word boundary when + * the cut falls inside a word. + * + * @param {string} text + * @param {number} limit + * @returns {string} + */ +function cutText(text, limit) { + if (limit >= text.length) return text; + if (limit <= 0) return ''; + const cut = text.slice(0, limit); + if (/\s/.test(text[limit])) return cut.trimEnd(); + const space = cut.search(/\s\S*$/); + return (space > 0 ? cut.slice(0, space) : cut).trimEnd(); +} + +/** + * The text of each message part, `null` for audio without a transcript, or + * `undefined` for a part that can't be recreated as text. + * + * @param {unknown} part + * @returns {string | null | undefined} + */ +function partText(part) { + if (!isPlainObject(part)) return undefined; + if (part.type === 'input_text' || part.type === 'output_text') + return typeof part.text === 'string' ? part.text : undefined; + if (part.type === 'input_audio' || part.type === 'output_audio') + return typeof part.transcript === 'string' ? part.transcript : null; + return undefined; +} + +/** + * Finished conversation items the SDK can recreate on a new provider session, + * kept so a reconnect can restore context. Mirrors what the provider holds, + * except that assistant output is cut to what the user saw or heard. + * + * @internal + */ +export class ConvoHistory { + /** @type {HistoryRecord[]} */ + #records = []; + /** @type {Map} */ + #byId = new Map(); + /** @type {Set} */ + #deleted = new Set(); + /** @type {(responseId: string) => boolean} */ + #isCancelled; + // Replayed entries whose copy on the new session hasn't arrived yet. + /** @type {ReplayEntry[]} */ + #echoes = []; + // IDs of those copies, until their `conversation.item.done`. + /** @type {Set} */ + #echoIds = new Set(); + + /** + * @param {(responseId: string) => boolean} isCancelled Whether the SDK + * cancelled a response. + */ + constructor(isCancelled) { + this.#isCancelled = isCancelled; + } + + /** + * @param {RealtimeServerEvent} event + * @returns {boolean} Whether the event announces the new session's copy of + * a replayed item. + */ + observe(event) { + if (ITEM_EVENTS.has(event.type)) + return this.#upsert( + event.item, + event.previous_item_id, + event.response_id, + DONE_EVENTS.has(event.type), + ); + const record = + typeof event.item_id === 'string' + ? this.#byId.get(event.item_id) + : undefined; + switch (event.type) { + case 'conversation.item.input_audio_transcription.completed': + case 'conversation.item.input_audio_transcription.failed': { + const index = Number(event.content_index ?? 0); + if (!record || !(index in record.parts)) break; + record.parts[index] = + event.type.endsWith('.completed') && + typeof event.transcript === 'string' + ? event.transcript + : ''; + break; + } + case 'conversation.item.truncated': + if (record && Number.isFinite(event.audio_end_ms)) + record.playedMs = Number(event.audio_end_ms); + break; + case 'conversation.item.deleted': + if (typeof event.item_id === 'string') + this.forget(event.item_id); + break; + case 'response.output_audio.delta': + if (record && typeof event.delta === 'string') + record.audioMs += + Math.floor((event.delta.length * 3) / 4) / + AUDIO_BYTES_PER_MS; + break; + default: + if ( + record && + TEXT_DELTAS.has(event.type) && + typeof event.delta === 'string' + ) { + record.received += event.delta; + if (!this.#isCancelled(String(event.response_id ?? ''))) + record.shown += event.delta; + } + } + return false; + } + + /** + * Drops an item the app deleted, and keeps it from coming back. + * + * @param {string} itemId + * @returns {void} + */ + forget(itemId) { + this.#deleted.add(itemId); + if (this.#deleted.size > MAX_DELETED) + this.#deleted.delete(this.#deleted.values().next().value); + const record = this.#byId.get(itemId); + if (record) this.#remove(record); + } + + /** + * Records that local playback of an item stopped after `playedMs`. + * + * @param {string} itemId + * @param {number} playedMs + * @returns {void} + */ + truncate(itemId, playedMs) { + const record = this.#byId.get(itemId); + if (!record) return; + record.playedMs = Math.min(record.playedMs ?? Infinity, playedMs); + } + + /** + * Whether a finished function call since the last user message has no + * output yet. Calls cut off by a cancellation don't count. + * + * @returns {boolean} + */ + get awaitingToolOutput() { + const answered = new Set(); + for (let i = this.#records.length - 1; i >= 0; i--) { + const record = this.#records[i]; + if (record.role === 'user') return false; + if (record.type === 'function_call_output') + answered.add(record.callId); + else if ( + record.type === 'function_call' && + record.final && + record.status !== 'incomplete' && + !answered.has(record.callId) + ) + return true; + } + return false; + } + + /** + * The newest recreatable items, oldest first, within `maxItems` and + * `maxBytes` of encoded `conversation.item.create` events. Stops at the + * first item that doesn't fit, so the window has no gaps from the budget. + * + * @param {number} maxItems + * @param {number} maxBytes + * @returns {ReplayEntry[]} + */ + replay(maxItems, maxBytes) { + /** @type {ReplayEntry[]} */ + const entries = []; + let bytes = 0; + for ( + let i = this.#records.length - 1; + i >= 0 && entries.length < maxItems; + i-- + ) { + const record = this.#records[i]; + const item = this.#replayItem(record); + if (!item) continue; + // The event wrapper and its ID add well under 128 bytes. + const size = encoder.encode(JSON.stringify(item)).byteLength + 128; + if (bytes + size > maxBytes) break; + bytes += size; + entries.push({ record, item }); + } + entries.reverse(); + const calls = new Set( + entries + .filter(({ item }) => item.type === 'function_call') + .map(({ item }) => item.call_id), + ); + return entries.filter( + ({ item }) => + item.type !== 'function_call_output' || calls.has(item.call_id), + ); + } + + /** + * Marks `entries` as sent to a new session, which answers with copies of + * them under new IDs. + * + * @param {ReplayEntry[]} entries + * @returns {void} + */ + replayed(entries) { + this.#echoes = [...entries]; + this.#echoIds.clear(); + } + + /** + * Drops a replayed item the provider refused. + * + * @param {ReplayEntry} entry + * @returns {void} + */ + refused(entry) { + this.#echoes = this.#echoes.filter((echo) => echo !== entry); + this.#remove(entry.record); + } + + /** + * @param {HistoryRecord} record + * @returns {void} + */ + #remove(record) { + const index = this.#records.indexOf(record); + if (index >= 0) this.#records.splice(index, 1); + if (this.#byId.get(record.id) === record) this.#byId.delete(record.id); + } + + /** + * @param {unknown} item + * @param {unknown} previousItemId + * @param {unknown} responseId + * @param {boolean} done + * @returns {boolean} Whether `item` is the new session's copy of a replayed + * item. + */ + #upsert(item, previousItemId, responseId, done) { + if (!isPlainObject(item) || typeof item.id !== 'string' || !item.id) + return false; + if (this.#echoIds.has(item.id)) { + if (done) this.#echoIds.delete(item.id); + return true; + } + if (this.#deleted.has(item.id)) return false; + let record = this.#byId.get(item.id); + if (!record) { + const fresh = this.#create(item); + if (!fresh) return false; + record = fresh; + const echo = this.#takeEcho(item); + if (echo) { + // The copy takes the original's place under its new ID. + this.#records[this.#records.indexOf(echo.record)] = record; + this.#byId.delete(echo.record.id); + this.#byId.set(record.id, record); + this.#update(record, item); + record.final = true; + if (!done) this.#echoIds.add(record.id); + return true; + } + const after = + typeof previousItemId === 'string' + ? this.#byId.get(previousItemId) + : undefined; + const index = after ? this.#records.indexOf(after) + 1 : -1; + if (index > 0) this.#records.splice(index, 0, record); + else this.#records.push(record); + if (this.#records.length > MAX_RECORDS) { + const [oldest] = this.#records.splice(0, 1); + this.#byId.delete(oldest.id); + } + this.#byId.set(record.id, record); + } + if (typeof responseId === 'string' && responseId) + record.responseId = responseId; + this.#update(record, item); + if (done || item.status === 'completed' || item.status === 'incomplete') + record.final = true; + if (record.final && record.role === 'assistant') + record.cancelled = this.#isCancelled(record.responseId); + return false; + } + + /** + * @param {Record} item + * @returns {HistoryRecord | null} + */ + #create(item) { + if ( + item.type !== 'message' && + item.type !== 'function_call' && + item.type !== 'function_call_output' + ) + return null; + if ( + item.type === 'message' && + !['user', 'assistant', 'system'].includes(item.role) + ) + return null; + return { + id: item.id, + type: item.type, + role: item.type === 'message' ? item.role : undefined, + responseId: '', + final: false, + unsupported: false, + parts: [], + received: '', + shown: '', + cancelled: false, + audioMs: 0, + }; + } + + /** + * @param {HistoryRecord} record + * @param {Record} item + * @returns {void} + */ + #update(record, item) { + if (typeof item.status === 'string') record.status = item.status; + if (record.type === 'message') { + if (!Array.isArray(item.content)) return; + record.unsupported = false; + record.parts = item.content.map((part, i) => { + const text = partText(part); + if (text === undefined) record.unsupported = true; + // A finished item can repeat its audio without the transcript. + return text ?? record.parts[i] ?? null; + }); + return; + } + if (typeof item.call_id === 'string') record.callId = item.call_id; + if (record.type === 'function_call') { + if (typeof item.name === 'string') record.name = item.name; + if (typeof item.arguments === 'string') + record.args = item.arguments; + } else if (typeof item.output === 'string') record.output = item.output; + } + + /** + * The `conversation.item.create` item that recreates `record`, or `null`. + * + * @param {HistoryRecord} record + * @returns {ConvoReplayItem | null} + */ + #replayItem(record) { + if (!record.final) return null; + if (record.type === 'function_call') { + if ( + record.status === 'incomplete' || + !record.callId || + !record.name || + typeof record.args !== 'string' + ) + return null; + return { + type: 'function_call', + call_id: record.callId, + name: record.name, + arguments: record.args, + }; + } + if (record.type === 'function_call_output') { + if (!record.callId || typeof record.output !== 'string') + return null; + return { + type: 'function_call_output', + call_id: record.callId, + output: record.output, + }; + } + if (record.unsupported || record.parts.includes(null)) return null; + if (record.role === 'assistant') { + const text = this.#assistantText(record); + if (!text.trim()) return null; + return { + type: 'message', + role: 'assistant', + content: [{ type: 'output_text', text }], + }; + } + const content = /** @type {string[]} */ (record.parts) + .filter((text) => text.trim()) + .map((text) => ({ type: 'input_text', text })); + if (!content.length) return null; + return { type: 'message', role: record.role, content }; + } + + /** + * Assistant text cut to what the user saw or heard: the deltas delivered + * before the SDK cancelled the response, and the share of the transcript + * matching the audio played before a truncation. + * + * @param {HistoryRecord} record + * @returns {string} + */ + #assistantText(record) { + const full = + /** @type {string[]} */ (record.parts).join('\n') || + record.received; + let limit = full.length; + if (record.cancelled) limit = Math.min(limit, record.shown.length); + if (record.playedMs !== undefined) + limit = Math.min( + limit, + record.audioMs > 0 + ? Math.floor( + (full.length * + Math.min(record.playedMs, record.audioMs)) / + record.audioMs, + ) + : 0, + ); + return cutText(full, limit); + } + + /** + * The pending replayed entry `item` is the new session's copy of. + * + * @param {Record} item + * @returns {ReplayEntry | undefined} + */ + #takeEcho(item) { + const index = this.#echoes.findIndex( + ({ item: sent }) => + sent.type === item.type && + (sent.type === 'message' + ? sent.role === item.role && + JSON.stringify( + /** @type {any[]} */ (sent.content).map( + (part) => part.text, + ), + ) === + JSON.stringify( + Array.isArray(item.content) + ? item.content.map(partText) + : null, + ) + : sent.call_id === item.call_id), + ); + if (index < 0) return undefined; + const [echo] = this.#echoes.splice(index, 1); + if (!this.#records.includes(echo.record)) return undefined; + return echo; + } +} diff --git a/src/puter-js/src/modules/ai/convoHistory.test.js b/src/puter-js/src/modules/ai/convoHistory.test.js new file mode 100644 index 000000000..fc18eb201 --- /dev/null +++ b/src/puter-js/src/modules/ai/convoHistory.test.js @@ -0,0 +1,134 @@ +import { describe, expect, it } from 'vitest'; +import { ConvoHistory } from './convoHistory.js'; + +const user = (id, content) => ({ + type: 'conversation.item.added', + item: { id, type: 'message', role: 'user', status: 'completed', content }, +}); +const texts = (history) => + history.replay(40, 48 * 1024).map(({ item }) => item.content[0].text); + +describe('ConvoHistory', () => { + it('cuts text without word breaks at the played share', () => { + const history = new ConvoHistory(() => false); + history.observe({ + type: 'response.output_item.added', + response_id: 'r', + item: { id: 'a', type: 'message', role: 'assistant', content: [] }, + }); + history.observe({ + type: 'response.output_audio.delta', + item_id: 'a', + delta: 'A'.repeat(64), // 1 ms + }); + history.observe({ + type: 'response.output_item.done', + item: { + id: 'a', + type: 'message', + role: 'assistant', + status: 'completed', + content: [ + { type: 'output_audio', transcript: '今日は晴れです' }, + ], + }, + }); + history.truncate('a', 0.5); + expect(texts(history)).toEqual(['今日は']); + }); + + it('keeps a known transcript when a later copy of the item lacks it', () => { + const history = new ConvoHistory(() => false); + history.observe(user('u', [{ type: 'input_audio', transcript: null }])); + expect(texts(history)).toEqual([]); + history.observe({ + type: 'conversation.item.input_audio_transcription.completed', + item_id: 'u', + content_index: 0, + transcript: 'hello', + }); + history.observe({ + ...user('u', [{ type: 'input_audio', transcript: null }]), + type: 'conversation.item.done', + }); + expect(texts(history)).toEqual(['hello']); + }); + + it('keeps a bounded number of the newest items', () => { + const history = new ConvoHistory(() => false); + for (let i = 0; i < 100; i++) + history.observe( + user(`u${i}`, [{ type: 'input_text', text: `${i}` }]), + ); + expect(history.replay(100, Infinity)).toHaveLength(80); + expect(texts(history).at(-1)).toBe('99'); + }); + + it('stops at the first item over the byte budget instead of leaving a gap', () => { + const history = new ConvoHistory(() => false); + history.observe(user('u1', [{ type: 'input_text', text: 'old' }])); + history.observe( + user('u2', [{ type: 'input_text', text: 'x'.repeat(50_000) }]), + ); + history.observe(user('u3', [{ type: 'input_text', text: 'new' }])); + expect(texts(history)).toEqual(['new']); + }); + + it('waits on every unanswered call since the last user message', () => { + const history = new ConvoHistory(() => false); + const call = (id, status = 'completed') => + history.observe({ + type: 'response.output_item.done', + item: { + id: `fc_${id}`, + type: 'function_call', + call_id: id, + name: 'f', + arguments: '{}', + status, + }, + }); + const output = (id) => + history.observe({ + type: 'conversation.item.added', + item: { + id: `out_${id}`, + type: 'function_call_output', + call_id: id, + output: '1', + status: 'completed', + }, + }); + call('a'); + call('b'); + call('cut', 'incomplete'); + output('b'); + expect(history.awaitingToolOutput).toBe(true); + output('a'); + expect(history.awaitingToolOutput).toBe(false); + call('c'); + history.observe(user('u', [{ type: 'input_text', text: 'moved on' }])); + expect(history.awaitingToolOutput).toBe(false); + }); + + 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)); + const copy = user('new', [{ type: 'input_text', text: 'hi' }]); + expect(history.observe(copy)).toBe(true); + expect( + history.observe({ ...copy, type: 'conversation.item.done' }), + ).toBe(true); + // A later item with the same text is the user's, not a copy. + expect( + history.observe( + user('again', [{ type: 'input_text', text: 'hi' }]), + ), + ).toBe(false); + expect(texts(history)).toEqual(['hi', 'hi']); + // The copy took the original's place under its new ID. + history.observe({ type: 'conversation.item.deleted', item_id: 'new' }); + expect(texts(history)).toEqual(['hi']); + }); +}); diff --git a/src/puter-js/src/modules/ai/convoMedia.js b/src/puter-js/src/modules/ai/convoMedia.js index c398c65b8..604674dad 100644 --- a/src/puter-js/src/modules/ai/convoMedia.js +++ b/src/puter-js/src/modules/ai/convoMedia.js @@ -344,6 +344,15 @@ export class ConvoSpeaker { get active() { return this.#context !== null; } + /** + * Whether audio is scheduled or playing. + * + * @returns {boolean} + * @internal + */ + get playing() { + return this.#sources.size > 0; + } /** * Starts playback without requesting microphone access. @@ -454,6 +463,7 @@ export class ConvoSpeaker { this.#last.playedMs, (offset + duration) * 1000, ); + if (!this.#sources.size) this.#convo.playbackEnded(); }; this.#nextAt = source.end; node.start(start); @@ -469,21 +479,8 @@ export class ConvoSpeaker { * @internal */ playbackPosition(responseId) { - const context = this.#context; - const heard = - (context?.currentTime ?? 0) - - (context?.outputLatency || context?.baseLatency || 0); - for (const source of this.#sources) { - if (source.end > heard) - return { - itemId: source.itemId, - contentIndex: source.contentIndex, - playedMs: Math.floor( - (source.offset + Math.max(0, heard - source.start)) * - 1000, - ), - }; - } + const [current] = this.unheard(); + if (current) return current; const last = this.#last; if (last && responseId && last.responseId === responseId) return { @@ -494,11 +491,39 @@ export class ConvoSpeaker { return undefined; } + /** + * Each item with scheduled audio the listener hasn't fully heard, in play + * order, with how much of it was heard. + * + * @returns {ConvoPlaybackPosition[]} + * @internal + */ + unheard() { + const context = this.#context; + const heard = + (context?.currentTime ?? 0) - + (context?.outputLatency || context?.baseLatency || 0); + /** @type {Map} */ + const items = new Map(); + for (const source of this.#sources) { + if (source.end <= heard || items.has(source.itemId)) continue; + items.set(source.itemId, { + itemId: source.itemId, + contentIndex: source.contentIndex, + playedMs: Math.floor( + (source.offset + Math.max(0, heard - source.start)) * 1000, + ), + }); + } + return [...items.values()]; + } + /** * @returns {void} * @internal */ clearPlayback() { + const playing = this.#sources.size > 0; for (const source of this.#sources) { source.node.onended = null; source.node.stop(); @@ -507,6 +532,7 @@ export class ConvoSpeaker { this.#sources.clear(); this.#last = null; this.#nextAt = 0; + if (playing) this.#convo.playbackEnded(); } /** diff --git a/src/puter-js/src/modules/ai/convoMedia.test.js b/src/puter-js/src/modules/ai/convoMedia.test.js index 6be1ae9da..4d15e74b5 100644 --- a/src/puter-js/src/modules/ai/convoMedia.test.js +++ b/src/puter-js/src/modules/ai/convoMedia.test.js @@ -50,6 +50,7 @@ const makeConvo = () => ({ }, appendAudio: vi.fn(), reportMediaError: vi.fn(), + playbackEnded: vi.fn(), interrupt: vi.fn(async () => {}), }); const track = () => ({ enabled: true, stop: vi.fn() }); @@ -329,6 +330,65 @@ it('subtracts output latency and covers gaps between chunks', async () => { expect(speaker.playbackPosition('other')).toBeUndefined(); }); +it('reports playing until the last scheduled chunk ends', async () => { + const c = makeConvo(); + const speaker = new ConvoSpeaker(c); + helpers.push(speaker); + await speaker.start(); + const context = FakeContext.instances[0]; + expect(speaker.playing).toBe(false); + for (let i = 0; i < 2; i++) + speaker.enqueue(new Int16Array(480), { + type: 'response.output_audio.delta', + item_id: 'item', + response_id: 'r', + }); + expect(speaker.playing).toBe(true); + context.nodes[0].onended(); + expect(speaker.playing).toBe(true); + expect(c.playbackEnded).not.toHaveBeenCalled(); + context.nodes[1].onended(); + expect(speaker.playing).toBe(false); + expect(c.playbackEnded).toHaveBeenCalledOnce(); + speaker.enqueue(new Int16Array(480), { + type: 'response.output_audio.delta', + item_id: 'item', + response_id: 'r', + }); + speaker.clearPlayback(); + expect(speaker.playing).toBe(false); + expect(c.playbackEnded).toHaveBeenCalledTimes(2); + // Nothing was playing, so nothing ended. + speaker.clearPlayback(); + expect(c.playbackEnded).toHaveBeenCalledTimes(2); +}); + +it('lists every item not fully heard, in play order', async () => { + const speaker = new ConvoSpeaker(makeConvo()); + helpers.push(speaker); + await speaker.start(); + const context = FakeContext.instances[0]; + const delta = (item_id) => ({ + type: 'response.output_audio.delta', + item_id, + response_id: 'r', + content_index: 0, + }); + // One second of a1 in two chunks, then half a second of a2. + speaker.enqueue(new Int16Array(12000), delta('a1')); + speaker.enqueue(new Int16Array(12000), delta('a1')); + speaker.enqueue(new Int16Array(12000), delta('a2')); + context.currentTime = 0.75; + expect(speaker.unheard()).toEqual([ + { itemId: 'a1', contentIndex: 0, playedMs: 750 }, + { itemId: 'a2', contentIndex: 0, playedMs: 0 }, + ]); + context.currentTime = 1.25; + expect(speaker.unheard()).toEqual([ + { itemId: 'a2', contentIndex: 0, playedMs: 250 }, + ]); +}); + it('a late canceled permission request does not stop a newer capture', async () => { let grantFirst; const oldTrack = track(); diff --git a/src/puter-js/tests/api/realtime/convo.test.ts b/src/puter-js/tests/api/realtime/convo.test.ts index 22ea591a0..5c79a58e6 100644 --- a/src/puter-js/tests/api/realtime/convo.test.ts +++ b/src/puter-js/tests/api/realtime/convo.test.ts @@ -1,17 +1,17 @@ import { chromium, type Browser } from '@playwright/test'; import { readFileSync } from 'node:fs'; +import net, { type Socket } from 'node:net'; import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import WebSocket, { WebSocketServer } from 'ws'; import type { AddressInfo } from 'node:net'; +import { loadNodePuter, closeNodePuter } from '../harness/nodeSdkLoader'; +import type { EnvManifest, PuterSDK } from '../harness/types'; +import type { Conversation } from '../../../types/modules/ai/convo'; +import type { RealtimeServerEvent } from '../../../types/modules/ai/types'; import { - loadNodePuter, - closeNodePuter, -} from '../harness/nodeSdkLoader'; -import type { - EnvManifest, - PuterSDK, -} from '../harness/types'; -import { setupPuterTestEnv, type PuterTestEnv } from '@heyputer/backend/testUtil.ts'; + setupPuterTestEnv, + type PuterTestEnv, +} from '@heyputer/backend/testUtil.ts'; import type { IConfig } from '@heyputer/backend/types.ts'; describe('convo built SDK and browser media', () => { @@ -19,10 +19,28 @@ describe('convo built SDK and browser media', () => { let upstream: WebSocketServer; let puter: PuterSDK; let browser: Browser | undefined; + let manifest: EnvManifest; + // A TCP hop in front of the server whose connections a test can cut, so + // the client sees an abnormal closure (1006). + let proxy: net.Server; + let proxyOrigin: string; + let proxyConnections = 0; + const proxied = new Set(); const received: { type: string; audio?: string }[] = []; + // Events each upstream connection received, in connection order. + const connections: { + socket: WebSocket; + received: { type: string; item?: unknown }[]; + }[] = []; + let ids = 0; beforeAll(async () => { upstream = new WebSocketServer({ port: 0 }); upstream.on('connection', (socket) => { + const connection = { + socket, + received: [] as { type: string; item?: unknown }[], + }; + connections.push(connection); socket.send( JSON.stringify({ type: 'session.created', session: {} }), ); @@ -30,6 +48,19 @@ describe('convo built SDK and browser media', () => { socket.on('message', (data) => { const event = JSON.parse(data.toString()); received.push(event); + connection.received.push(event); + const send = (payload: Record) => + socket.send(JSON.stringify(payload)); + if (event.type === 'conversation.item.create') { + const item = { + ...event.item, + id: `item_${++ids}`, + object: 'realtime.item', + status: 'completed', + }; + send({ type: 'conversation.item.added', item }); + send({ type: 'conversation.item.done', item }); + } if (event.type === 'session.update') socket.send( JSON.stringify({ @@ -38,28 +69,49 @@ describe('convo built SDK and browser media', () => { }), ); if (event.type === 'response.create') { - socket.send( - JSON.stringify({ - type: 'response.created', - response: { id: 'r1' }, - }), - ); - socket.send( - JSON.stringify({ - type: 'response.output_text.delta', - response_id: 'r1', - delta: 'Hello', - }), - ); - socket.send( - JSON.stringify({ - type: 'response.output_audio.delta', - response_id: 'r1', - item_id: 'item1', - content_index: 0, - delta: Buffer.alloc(960).toString('base64'), - }), - ); + const responseId = `resp_${++ids}`; + const item = { + id: `item_${++ids}`, + object: 'realtime.item', + type: 'message', + role: 'assistant', + }; + send({ + type: 'response.created', + response: { id: responseId }, + }); + send({ + type: 'response.output_item.added', + response_id: responseId, + item: { ...item, status: 'in_progress', content: [] }, + }); + send({ + type: 'response.output_text.delta', + response_id: responseId, + item_id: item.id, + content_index: 0, + delta: 'Hello', + }); + send({ + type: 'response.output_audio.delta', + response_id: responseId, + item_id: item.id, + content_index: 0, + delta: Buffer.alloc(960).toString('base64'), + }); + send({ + type: 'response.output_item.done', + response_id: responseId, + item: { + ...item, + status: 'completed', + content: [{ type: 'output_text', text: 'Hello' }], + }, + }); + send({ + type: 'response.done', + response: { id: responseId, status: 'completed' }, + }); } if (event.type === 'response.cancel') socket.send( @@ -88,16 +140,40 @@ describe('convo built SDK and browser media', () => { providers: { 'openai-realtime': { apiKey: 'test-provider-key' } }, unlimitedMetering: true, } as IConfig); - const manifest: EnvManifest = { + manifest = { origin: env.origin, apiOrigin: env.apiOrigin, users: env.users, }; puter = loadNodePuter(manifest, env.users.user.token); + const api = new URL(env.apiOrigin); + proxy = net.createServer((inbound) => { + proxyConnections++; + const outbound = net.connect({ + host: api.hostname, + port: Number(api.port), + }); + for (const [from, to] of [ + [inbound, outbound], + [outbound, inbound], + ]) { + proxied.add(from); + from.on('error', () => to.destroy()); + from.on('close', () => { + proxied.delete(from); + to.destroy(); + }); + from.pipe(to); + } + }); + await new Promise((resolve) => proxy.listen(0, resolve)); + proxyOrigin = `http://${api.hostname}:${(proxy.address() as AddressInfo).port}`; }, 60_000); afterAll(async () => { await browser?.close(); if (puter) closeNodePuter(puter); + for (const socket of proxied) socket.destroy(); + await new Promise((resolve) => proxy?.close(() => resolve())); for (const socket of upstream?.clients ?? []) socket.terminate(); await env?.shutdown(); await new Promise((resolve) => upstream?.close(() => resolve())); @@ -220,8 +296,31 @@ describe('convo built SDK and browser media', () => { } c.microphone.mute(); c.microphone.unmute(); - c.sendText('Hello'); - await new Promise((resolve) => setTimeout(resolve, 300)); + // Closes once captured audio went out and the reply came back. + await new Promise((resolve, reject) => { + let appended = false; + let replied = false; + const timer = setTimeout( + () => reject(new Error('no captured audio or reply')), + 10_000, + ); + const check = () => { + if (!appended || !replied) return; + clearTimeout(timer); + resolve(); + }; + const appendAudio = c.appendAudio.bind(c); + c.appendAudio = (pcm: Int16Array) => { + appendAudio(pcm); + appended = true; + check(); + }; + c.on('turn.done', () => { + replied = true; + check(); + }); + c.sendText('Hello'); + }); c.close(); return { state: c.state, @@ -247,4 +346,153 @@ describe('convo built SDK and browser media', () => { ).toBe(true), ); }, 30_000); + /** Collects a conversation's events and waits on them. */ + const watch = (conversation: Conversation) => { + const events: RealtimeServerEvent[] = []; + conversation.on('*', (event) => events.push(event)); + const count = (type: string) => + events.filter((event) => event.type === type).length; + const until = (check: () => boolean, timeout = 10_000) => + vi.waitFor(() => expect(check()).toBe(true), { timeout }); + return { events, count, until }; + }; + const restored = (connection: (typeof connections)[number]) => + connection.received + .filter((event) => event.type === 'conversation.item.create') + .map((event) => event.item); + const history = [ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'Remember the number 7.' }], + }, + { + type: 'message', + role: 'assistant', + content: [{ type: 'output_text', text: 'Hello' }], + }, + ]; + + it('restores context on a new session after the connection drops', async () => { + const dropping = loadNodePuter( + { ...manifest, apiOrigin: proxyOrigin }, + env.users.user.token, + ); + const conversation = await dropping.ai.convo({ output: ['text'] }); + const log = watch(conversation); + try { + await conversation.connect(); + const first = connections.at(-1)!; + conversation.sendText('Remember the number 7.'); + await log.until(() => log.count('turn.done') === 1); + + for (const socket of proxied) socket.destroy(); + await log.until(() => + log.events.some( + (event) => event.type === 'ready' && event.reconnected, + ), + ); + expect( + log.events.find((event) => event.type === 'reconnecting'), + ).toMatchObject({ code: 1006 }); + const second = connections.at(-1)!; + expect(second).not.toBe(first); + // Sent before ready, so they reach the provider ahead of the next turn. + await vi.waitFor(() => expect(restored(second)).toEqual(history)); + expect(second.received[0].type).toBe('session.update'); + // The relay ended the old provider session. + await vi.waitFor(() => + expect(first.socket.readyState).toBe(WebSocket.CLOSED), + ); + + conversation.sendText('Which number?'); + await log.until(() => log.count('turn.done') === 2); + expect(restored(second)).toEqual([ + ...history, + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'Which number?' }], + }, + ]); + expect(log.count('error')).toBe(0); + } finally { + conversation.close(); + closeNodePuter(dropping); + } + }); + + it('moves to a new session when a restart is announced, restoring context', async () => { + const conversation = await puter.ai.convo({ output: ['text'] }); + const log = watch(conversation); + try { + await conversation.connect(); + const first = connections.at(-1)!; + conversation.sendText('Remember the number 7.'); + await log.until(() => log.count('turn.done') === 1); + + env.server.clients.event.emit( + 'serverRestartAnnounced', + { closeInMs: 60_000 }, + {}, + ); + await log.until(() => + log.events.some( + (event) => event.type === 'ready' && event.reconnected, + ), + ); + expect( + log.events.find((event) => event.type === 'puter.reconnect'), + ).toMatchObject({ + reason: 'server_restarting', + close_in_ms: 60_000, + }); + expect( + log.events.find((event) => event.type === 'reconnecting'), + ).toMatchObject({ code: 1012, reason: 'server_restarting' }); + const second = connections.at(-1)!; + expect(second).not.toBe(first); + // Sent before ready, so they reach the provider ahead of the next turn. + await vi.waitFor(() => expect(restored(second)).toEqual(history)); + await vi.waitFor(() => + expect(first.socket.readyState).toBe(WebSocket.CLOSED), + ); + + conversation.sendText('Which number?'); + await log.until(() => log.count('turn.done') === 2); + expect(log.count('error')).toBe(0); + expect(log.count('close')).toBe(0); + } finally { + conversation.close(); + } + }); + + // 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( + { ...manifest, apiOrigin: proxyOrigin }, + env.users.user.token, + ); + const conversation = await draining.ai.convo({ output: ['text'] }); + const log = watch(conversation); + try { + await conversation.connect(); + const before = proxyConnections; + await env.server.services.openaiRealtime.onServerDrainStart({ + drainMs: 60_000, + }); + await log.until(() => log.count('puter.reconnect') === 1); + // One replacement within 3 seconds, the next about a second later. + await log.until(() => proxyConnections - before >= 2, 15_000); + conversation.sendText('Still there?'); + await log.until(() => log.count('turn.done') === 1); + expect(conversation.state).toBe('connected'); + expect(log.count('reconnecting')).toBe(0); + expect(log.count('error')).toBe(0); + } finally { + conversation.close(); + closeNodePuter(draining); + } + expect(conversation.state).toBe('closed'); + }); }); diff --git a/src/puter-js/tests/api/realtime/convoLive.test.ts b/src/puter-js/tests/api/realtime/convoLive.test.ts index 23fb9a0cc..e13638bb2 100644 --- a/src/puter-js/tests/api/realtime/convoLive.test.ts +++ b/src/puter-js/tests/api/realtime/convoLive.test.ts @@ -1,17 +1,14 @@ import WebSocket from 'ws'; import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; -import { - closeNodePuter, - loadNodePuter, -} from '../harness/nodeSdkLoader'; -import type { - EnvManifest, - PuterSDK, -} from '../harness/types'; +import { closeNodePuter, loadNodePuter } from '../harness/nodeSdkLoader'; +import type { EnvManifest, PuterSDK } from '../harness/types'; import type { Conversation } from '../../../types/modules/ai/convo'; import type { RealtimeServerEvent } from '../../../types/modules/ai/types'; import { optionalEnv } from '@heyputer/backend/drivers/integrationTestUtil.ts'; -import { setupPuterTestEnv, type PuterTestEnv } from '@heyputer/backend/testUtil.ts'; +import { + setupPuterTestEnv, + type PuterTestEnv, +} from '@heyputer/backend/testUtil.ts'; import type { IConfig } from '@heyputer/backend/types.ts'; const apiKey = optionalEnv('PUTER_TEST_AI_OPENAI_API_KEY'); @@ -84,16 +81,22 @@ describe.skipIf(!apiKey)('OpenAI realtime live relay', () => { let text = ''; try { await new Promise((resolve, reject) => { - const timer = setTimeout(() => reject(new Error('Realtime timeout')), 20_000); + const timer = setTimeout( + () => reject(new Error('Realtime timeout')), + 20_000, + ); const fail = (error: Error) => { clearTimeout(timer); reject(error); }; socket.on('error', fail); - socket.on('close', () => fail(new Error('Realtime closed before completion'))); + socket.on('close', () => + fail(new Error('Realtime closed before completion')), + ); socket.on('message', (data) => { const event = JSON.parse(data.toString()); - const send = (payload: Record) => socket.send(JSON.stringify(payload)); + const send = (payload: Record) => + socket.send(JSON.stringify(payload)); if (event.type === 'session.created') { send({ type: 'session.update', @@ -108,17 +111,23 @@ describe.skipIf(!apiKey)('OpenAI realtime live relay', () => { send({ type: 'conversation.item.create', item: { - type: 'message', role: 'user', - content: [{ type: 'input_text', text: 'Hello!' }], + type: 'message', + role: 'user', + content: [ + { type: 'input_text', text: 'Hello!' }, + ], }, }); send({ type: 'response.create' }); } - if (event.type === 'response.output_text.delta') text += event.delta; - if (event.type === 'error') fail(new Error(`Realtime error: ${event.error?.code}`)); + if (event.type === 'response.output_text.delta') + text += event.delta; + if (event.type === 'error') + fail(new Error(`Realtime error: ${event.error?.code}`)); if (event.type === 'response.done') { clearTimeout(timer); - if (event.response?.status !== 'completed') fail(new Error('Realtime response incomplete')); + if (event.response?.status !== 'completed') + fail(new Error('Realtime response incomplete')); else resolve(); } }); @@ -189,6 +198,88 @@ describe.skipIf(!apiKey)('OpenAI realtime live relay', () => { }, 45_000, ); + it('restores tool and text context on a new provider session after a restart notice', async () => { + const session = await puter.ai.convo({ + output: ['text'], + instructions: 'Answer in one short sentence.', + }); + sessions.add(session); + const errors: string[] = []; + session.on('error', (event) => + errors.push( + String(event.code ?? (event.error as { code?: string })?.code), + ), + ); + try { + await session.connect(); + let updated = waitFor(session, 'session.updated'); + session.send({ + type: 'session.update', + session: { + type: 'realtime', + tools: [ + { + type: 'function', + name: 'get_secret_word', + description: 'Returns the secret word.', + parameters: { type: 'object', properties: {} }, + }, + ], + tool_choice: 'required', + }, + }); + await updated; + let done = waitFor(session, 'response.done'); + session.sendText('What is the secret word?'); + const call = ( + (await done).response as { + output: { type: string; call_id: string }[]; + } + ).output.find((item) => item.type === 'function_call'); + expect(call).toBeDefined(); + updated = waitFor(session, 'session.updated'); + session.send({ + type: 'session.update', + session: { type: 'realtime', tool_choice: 'none' }, + }); + await updated; + session.send({ + type: 'conversation.item.create', + item: { + type: 'function_call_output', + call_id: call!.call_id, + output: JSON.stringify({ word: 'pineapple' }), + }, + }); + done = waitFor(session, 'response.done'); + session.send({ type: 'response.create' }); + await done; + + const ready = waitFor(session, 'ready'); + env.server.clients.event.emit('serverRestartAnnounced', {}, {}); + expect(await ready).toMatchObject({ reconnected: true }); + + let text = ''; + const off = session.on('text', (event) => { + text += event.delta; + }); + done = waitFor(session, 'response.done'); + session.sendText( + 'Repeat the secret word. Reply with just the word.', + ); + await done; + off(); + expect(text.toLowerCase()).toContain('pineapple'); + expect(errors).toEqual([]); + process.stdout.write( + `[live convo] context restored after reconnect, reply=${JSON.stringify(text)}\n`, + ); + } finally { + session.close(); + sessions.delete(session); + } + }, 60_000); + it('receives live PCM audio and interrupts the current response', async () => { const c = await puter.ai.convo({ output: ['audio'],