diff --git a/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts b/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts index b15a406b5..dafa11d8f 100644 --- a/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts +++ b/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts @@ -194,7 +194,11 @@ describe('OpenAIRealtimeRelayService relay', () => { let user: TestUserCredentials; let relayUrl: string; const upstreamRequests: IncomingMessage[] = []; - const upstreamEvents: { type: string; response_id?: string }[] = []; + const upstreamEvents: { + type: string; + response_id?: string; + event_id?: string; + }[] = []; const clients = new Set(); let responseCount = 0; @@ -294,6 +298,11 @@ describe('OpenAIRealtimeRelayService relay', () => { ); return; } + // Drops the connection, like a provider ending the session. + if (event.type === 'test.drop') { + ws.terminate(); + return; + } if (event.type === 'test.oversize') { ws.send( JSON.stringify({ @@ -611,9 +620,22 @@ describe('OpenAIRealtimeRelayService relay', () => { c.ws.close(); }); - it('closes on an oversized upstream message and alarms on the response it loses', async () => { + // 8,000 audio input and 4,096 audio output tokens at mini rates. + const MINI_HOLD = 8_000 * 1000 + 4_096 * 2000; + const unreportedRows = (costOverride: number) => [ + { + usageType: 'openai:gpt-realtime-mini:realtime:unreported_response', + usageAmount: 1, + costOverride, + }, + ]; + + it('closes on an oversized upstream message and bills the response it loses at its hold', async () => { const alarm = vi.spyOn(server.clients.alarm, 'create'); - const c = open(`${relayUrl}?auth_token=${user.token}`); + const spy = vi.spyOn(server.services.metering, 'batchIncrementUsages'); + const c = open( + `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, + ); await c.waitFor('session.created'); c.ws.send(JSON.stringify({ type: 'hold.response' })); await c.waitFor('response.created'); @@ -625,14 +647,107 @@ describe('OpenAIRealtimeRelayService relay', () => { ), ).toBe(false); await vi.waitFor(() => - expect(alarm).toHaveBeenCalledWith( - 'realtime_unbilled_response', - expect.any(String), - expect.objectContaining({ inFlight: 1 }), - 'warning', - { dedup: true }, + expect(spy.mock.calls.map(([, rows]) => rows)).toContainEqual( + unreportedRows(MINI_HOLD), ), ); + expect(alarm).toHaveBeenCalledWith( + 'realtime_unreported_response', + expect.any(String), + expect.objectContaining({ inFlight: 1 }), + 'warning', + { dedup: true }, + ); + }); + + it('bills a response at its hold through the cost factor when the upstream drops mid-response', async () => { + const dropped = await createTestUser(server, { + username: 'realtime-dropped', + password: 'realtime-dropped-password', + }); + const row = await server.stores.user.getByUsername(dropped.username); + const actor = { user: row } as never; + const metering = server.services.metering; + const spy = vi.spyOn(metering, 'batchIncrementUsages'); + await withAiCostFactorHook(server.clients.event, 2, async () => { + const c = open( + `${relayUrl}?model=gpt-realtime-mini&auth_token=${dropped.token}`, + ); + await c.waitFor('session.created'); + c.ws.send(JSON.stringify({ type: 'hold.response' })); + await c.waitFor('response.created'); + await vi.waitFor(async () => + expect(await metering.getOutstandingHolds(actor)).toBe( + MINI_HOLD * 2, + ), + ); + c.ws.send(JSON.stringify({ type: 'test.drop' })); + expect((await c.closed).code).toBe(1011); + await vi.waitFor(() => + expect(spy.mock.calls.map(([, rows]) => rows)).toContainEqual( + unreportedRows(MINI_HOLD * 2), + ), + ); + }); + await vi.waitFor(async () => + expect(await metering.getOutstandingHolds(actor)).toBe(0), + ); + }); + + const mcpTool = { + type: 'mcp', + server_label: 'docs', + server_url: 'https://mcp.example.com', + }; + it.each([ + [ + 'conversation.item.retrieve', + { type: 'conversation.item.retrieve', item_id: 'item_1' }, + ], + [ + 'an MCP tool in session.update', + { + type: 'session.update', + session: { type: 'realtime', tools: [mcpTool] }, + }, + ], + [ + 'an MCP tool in response.create', + { type: 'response.create', response: { tools: [mcpTool] } }, + ], + ])('refuses %s without forwarding it', async (_name, event) => { + const c = open(`${relayUrl}?auth_token=${user.token}`); + await c.waitFor('session.created'); + const eventId = `event_refused_${Date.now()}_${Math.random()}`; + c.ws.send(JSON.stringify({ ...event, event_id: eventId })); + expect(await c.waitFor('error')).toMatchObject({ + error: { code: 'invalid_event' }, + }); + expect((await c.closed).code).toBe(1008); + expect(upstreamEvents.some((e) => e.event_id === eventId)).toBe(false); + }); + + it('forwards function tools', async () => { + const c = open(`${relayUrl}?auth_token=${user.token}`); + await c.waitFor('session.created'); + c.ws.send( + JSON.stringify({ + type: 'session.update', + session: { + type: 'realtime', + tools: [ + { + type: 'function', + name: 'lookup', + parameters: { type: 'object' }, + }, + ], + }, + }), + ); + await c.waitForEcho('session.update'); + expect(c.events.some((e) => e.type === 'error')).toBe(false); + c.ws.close(); }); it('treats a null usage as none and keeps the session', async () => { @@ -654,7 +769,7 @@ describe('OpenAIRealtimeRelayService relay', () => { c.ws.close(); await c.closed; expect(alarm).not.toHaveBeenCalledWith( - 'realtime_unbilled_response', + 'realtime_unreported_response', expect.anything(), expect.anything(), expect.anything(), @@ -1065,11 +1180,8 @@ describe('OpenAIRealtimeRelayService relay', () => { expect(await metering.getOutstandingHolds(actor)).toBe(0); c.ws.send(JSON.stringify({ type: 'hold.response' })); await c.waitFor('response.created'); - // 8,000 audio input and 4,096 audio output tokens at mini rates. await vi.waitFor(async () => - expect(await metering.getOutstandingHolds(actor)).toBe( - 8_000 * 1000 + 4_096 * 2000, - ), + expect(await metering.getOutstandingHolds(actor)).toBe(MINI_HOLD), ); c.ws.close(); await vi.waitFor(async () => diff --git a/src/backend/services/realtime/OpenAIRealtimeRelayService.ts b/src/backend/services/realtime/OpenAIRealtimeRelayService.ts index e8f637a0f..c0bed662a 100644 --- a/src/backend/services/realtime/OpenAIRealtimeRelayService.ts +++ b/src/backend/services/realtime/OpenAIRealtimeRelayService.ts @@ -128,8 +128,8 @@ export interface RealtimeTranscriptionUsage { const OPENAI_KEY_PROTOCOL = 'openai-insecure-api-key.'; export const MAX_REALTIME_MESSAGE_BYTES = 64 * 1024; -// OpenAI events can carry whole items, such as a retrieved audio item or a -// long reply, so upstream messages get far more room than client ones. +// OpenAI events can carry whole items, such as a long reply, so upstream +// messages get far more room than client ones. export const MAX_UPSTREAM_MESSAGE_BYTES = 4 * 1024 * 1024; const MAX_PENDING_MESSAGES = 32; // Usage batches waiting on metering; past this the session closes, but every @@ -375,6 +375,39 @@ const responseHoldAmount = (model: string): number => { ); }; +/** Bills `count` responses that never reported usage at one hold each. */ +const unreportedResponseRows = (model: string, count: number): UsageInput[] => [ + { + usageType: `openai:${model}:realtime:unreported_response`, + usageAmount: count, + costOverride: count * responseHoldAmount(model), + }, +]; + +/** + * Why a client event is refused, or null. Retrieved items and MCP tool output + * can exceed the upstream message cap. + */ +const clientEventRefusal = (event: RealtimeEvent): string | null => { + if (event.type === 'conversation.item.retrieve') { + return 'conversation.item.retrieve is not supported'; + } + const config: unknown = + event.type === 'session.update' + ? event.session + : event.type === 'response.create' + ? event.response + : undefined; + if ( + isRecord(config) && + Array.isArray(config.tools) && + config.tools.some((tool) => isRecord(tool) && tool.type === 'mcp') + ) { + return 'MCP tools are not supported'; + } + return null; +}; + /** Whichever of two pricings of the same usage costs more. */ export const costlierRows = ( a: UsageInput[], @@ -613,8 +646,9 @@ export class OpenAIRealtimeRelayService extends PuterService { const token = realtimeTokenFrom(req, url); let actor: Actor | null = null; let rechecking = false; - // Responses started upstream and not yet done. A response only bills - // on `response.done`, so the upstream outlives the client until then. + // Responses started upstream and not yet done. A response bills its + // usage on `response.done`, so the upstream outlives the client until + // then; one the upstream closes on first is billed its hold. let inFlight = 0; const inFlightIds = new Set(); // Client `response.create` events whose response hasn't started; the @@ -836,12 +870,12 @@ export class OpenAIRealtimeRelayService extends PuterService { // draining after the session closes. const usageQueue: { rows: UsageInput[]; - hold?: Promise; + holds: Promise[]; }[] = []; let recording = false; const drainUsage = async () => { while (usageQueue.length) { - const { rows, hold } = usageQueue[0]; + const { rows, holds } = usageQueue[0]; if (rows.length) { await this.#record(actor!, rows, () => reject( @@ -850,18 +884,19 @@ export class OpenAIRealtimeRelayService extends PuterService { ), ); } - await releaseHold(hold); + await Promise.all(holds.map((hold) => releaseHold(hold))); usageQueue.shift(); } recording = false; }; - // `hold` is released once `rows` are recorded. + // `holds` are released once `rows` are recorded. const meter = ( rows: UsageInput[], - hold?: Promise, + ...maybeHolds: (Promise | undefined)[] ) => { - if (!rows.length && !hold) return; - usageQueue.push({ rows, hold }); + const holds = maybeHolds.filter((hold) => hold !== undefined); + if (!rows.length && !holds.length) return; + usageQueue.push({ rows, holds }); if (!recording) { recording = true; setImmediate(() => void drainUsage()); @@ -1028,6 +1063,11 @@ export class OpenAIRealtimeRelayService extends PuterService { ); return; } + const refusal = clientEventRefusal(event); + if (refusal) { + reject('invalid_event', refusal); + return; + } spend(size); if (closed) return; if (event.type === 'response.create') requested++; @@ -1247,27 +1287,30 @@ export class OpenAIRealtimeRelayService extends PuterService { }); up.on('close', (code, reason) => { if (settleTimer) clearTimeout(settleTimer); - if (inFlight > 0) { - this.clients.alarm.create( - 'realtime_unbilled_response', - 'Realtime upstream closed before a response reported usage', - { - userId: actor?.user?.uuid, - model: billingModel, - inFlight, - code, - }, - 'warning', - { dedup: true }, - ); - } - for (const id of [...responseHolds.keys()]) { - void releaseHold(takeHold(id)); - } session.close( code === 1005 || code === 1006 ? 1011 : code, reason.toString().slice(0, 120), ); + const holds = [...responseHolds.values()]; + responseHolds.clear(); + if (inFlight === 0) { + for (const hold of holds) void releaseHold(hold); + return; + } + this.clients.alarm.create( + 'realtime_unreported_response', + 'Realtime upstream closed before a response reported usage', + { + userId: actor?.user?.uuid, + model: billingModel, + inFlight, + code, + }, + 'warning', + { dedup: true }, + ); + // The holds are released once this charge is recorded. + meter(unreportedResponseRows(billingModel, inFlight), ...holds); }); } diff --git a/src/backend/services/realtime/README.md b/src/backend/services/realtime/README.md index b6b8857c6..c60cae893 100644 --- a/src/backend/services/realtime/README.md +++ b/src/backend/services/realtime/README.md @@ -47,13 +47,18 @@ backpressure bounds what a slow client can queue. Usage is priced as each event arrives, before any forwarding decision, and recorded serially off the forwarding path through the AI cost-factor hook. Pricing follows the model the session reports, falling back to the connection's; a client `session.update` -that changes the model is refused. +that changes the model is refused. So are `conversation.item.retrieve` and MCP +tools in `session.update` or `response.create`, since a retrieved item or MCP +output can exceed the upstream message cap. When a session closes with a response still generating, the relay cancels it and keeps the upstream open for up to five seconds so the cancelled response reports, and is billed for, what it generated. A response the client requested -that starts during that window is cancelled the same way. An upstream that -closes with a response unreported raises a warning alarm. +that starts during that window is cancelled the same way. When the upstream +closes, for any reason, with a started response that hasn't reported usage, +that response is billed its full hold through the same metering path and a +warning alarm is raised. A response the client requested but that never +started isn't billed. Each response holds a fixed amount of the account's budget (`RESPONSE_HOLD_*_TOKENS` at the model's audio rates) from `response.created` diff --git a/src/docs/src/AI/convo.md b/src/docs/src/AI/convo.md index 14bb54809..0cd8e7bd2 100644 --- a/src/docs/src/AI/convo.md +++ b/src/docs/src/AI/convo.md @@ -62,7 +62,7 @@ When Puter knows ahead of time that the server a session runs on will restart, i 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. +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. Puter refuses `conversation.item.retrieve` events and MCP tools (`"type": "mcp"`) in `session.update` and `response.create` events. Sending one fires an `error` event with code `invalid_event`, and the conversation closes with code 1008. Audio helpers expect the session to use PCM16 at 24 kHz. ## Convenience events diff --git a/src/docs/src/AI/realtime.md b/src/docs/src/AI/realtime.md index 3f3e02bc4..0fb17e7bb 100644 --- a/src/docs/src/AI/realtime.md +++ b/src/docs/src/AI/realtime.md @@ -82,6 +82,8 @@ socket.on('close', (code, reason) => console.log(code, reason.toString())); Each message contains one JSON object with a nonempty string `type` of at most 128 characters. Binary messages are rejected. Native events retain their provider fields. The endpoint doesn't deliver `rate_limits.updated`, which describes Puter's provider account rather than yours, and it removes provider organization IDs from error messages. The endpoint exposes the WebSocket event protocol. Your client manages capture, playback, acknowledgments, and timeouts. +The endpoint refuses `conversation.item.retrieve` events and MCP tools (`"type": "mcp"`) in `session.update` and `response.create` events. A refused event isn't sent to the provider. The session gets an `invalid_event` error and closes with 1008. + PCM16 audio configured at 24 kHz mono is sent as base64 chunks in `input_audio_buffer.append` events. Output arrives in `response.output_audio.delta` events. With manual turn detection, commit buffered input with `input_audio_buffer.commit`, then request a reply with `response.create`. To interrupt, stop local playback immediately and send `response.cancel` with the active `response_id`. If some generated audio was not played, send `conversation.item.truncate` with its `item_id`, `content_index`, and `audio_end_ms` set to the amount actually played. Wait for `response.done` and any requested `conversation.item.truncated` acknowledgment before submitting a replacement turn. @@ -111,4 +113,4 @@ API errors arrive as `{ "type": "error", "error": { "code": "...", "message": ". Policy failures generally close with 1008, oversized client messages with 1009, upstream or setup failures with 1011, and server restarts with 1012. Authentication and model failures can arrive after the WebSocket handshake succeeds. -The endpoint requires a paid Puter plan. Usage is charged after completed responses and input transcriptions. While a response generates, part of your balance is held for it, so a session opens only if your balance covers one response. Each account can open 20 sessions at once by default. Each JSON message you send is limited to 64 KiB, including reassembled fragments. Messages from the provider can be up to 4 MiB. Stream audio in small chunks. All message rates, queue caps, timeouts, and other bounds are published in [rate limits and quotas](/rate-limits-and-quotas/). +The endpoint requires a paid Puter plan. Usage is charged after completed responses and input transcriptions. While a response generates, part of your balance is held for it, so a session opens only if your balance covers one response. If the provider connection closes before a response reports its usage, the response is charged the full amount held for it. Each account can open 20 sessions at once by default. Each JSON message you send is limited to 64 KiB, including reassembled fragments. Messages from the provider can be up to 4 MiB. Stream audio in small chunks. All message rates, queue caps, timeouts, and other bounds are published in [rate limits and quotas](/rate-limits-and-quotas/). diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index 5c5942a1b..ba95b6b10 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -96,7 +96,7 @@ Realtime sessions ([`puter.ai.convo()`](/AI/convo/), `/puterai/openai/v1/realtim | Realtime transport limit | Value | Behavior at the limit | | --- | --- | --- | | Client JSON message size | 64 KiB, including reassembled fragments | Close 1009. | -| Upstream JSON message size | 4 MiB, including reassembled fragments | Close 1011. | +| Upstream JSON message size | 4 MiB, including reassembled fragments | Close 1011. A response that hasn't reported its usage is billed the credit held for it. | | Fragments per client message | 64 | Connection closes. | | Buffered receive chunks from the client | 128 | Connection closes. | | Event type length | 1–128 characters | Invalid client envelopes receive `invalid_event` and close with 1008. Invalid upstream envelopes close with 1011. | @@ -112,10 +112,10 @@ Realtime sessions ([`puter.ai.convo()`](/AI/convo/), `/puterai/openai/v1/realtim | Heartbeat interval | 30 seconds | A missing pong terminates the connection on the next heartbeat. | | 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. | +| Credit held per response | 8,000 audio input and 4,096 audio output tokens at the session model's rates, held until the response's usage is recorded | A session opens only if the account can cover one hold. A response the account can't cover gets `insufficient_funds`, close 1008, and is billed for what it generated. A response whose usage never arrives because the provider connection closed first is billed the full hold. | | Restart notice (`puter.reconnect`) | At most once a minute per session | Repeated announcements within the minute aren't sent. | | In-flight response drain at shutdown | 5 seconds | Close 1012. | -| 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. | +| 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. A response that doesn't report its usage within the 5 seconds is billed the credit held for it. | Send audio as a stream of small chunks. The message cap applies to text, configuration, tool output, and audio alike. diff --git a/src/puter-js/tests/api/realtime/convo.test.ts b/src/puter-js/tests/api/realtime/convo.test.ts index 5c79a58e6..81e0626a3 100644 --- a/src/puter-js/tests/api/realtime/convo.test.ts +++ b/src/puter-js/tests/api/realtime/convo.test.ts @@ -467,6 +467,33 @@ describe('convo built SDK and browser media', () => { } }); + it('closes with invalid_event when Puter refuses a native event', async () => { + const conversation = await puter.ai.convo({ output: ['text'] }); + const log = watch(conversation); + try { + await conversation.connect(); + conversation.send({ + type: 'conversation.item.retrieve', + item_id: 'item_1', + }); + await log.until(() => conversation.state === 'closed'); + expect( + log.events.find((event) => event.type === 'error'), + ).toMatchObject({ error: { code: 'invalid_event' } }); + expect( + log.events.find((event) => event.type === 'close'), + ).toMatchObject({ code: 1008 }); + expect(log.count('reconnecting')).toBe(0); + expect( + received.some( + (event) => event.type === 'conversation.item.retrieve', + ), + ).toBe(false); + } 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(