Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion examples/realtime-server-controlled/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,8 @@ This is not the Agents SDK's standard browser WebRTC transport with a data-chann

The included cookie session provides only a runnable localhost authentication example. It does not identify a real user and does not prevent an arbitrary local caller from creating its own demo principal. Before deploying this pattern, replace `DemoAuthStore` with your application's authentication and authorization, apply production rate limits, and derive `OpenAI-Safety-Identifier` from a stable privacy-preserving user identifier.

Anonymous demo authentication also has a 1,000-principal capacity limit. The server returns 503 for new principals at capacity and still renews existing cookies. Expiry sweeps run at most once per minute, so new principals may wait up to one sweep interval for expired entries to release capacity.

### Cleanup and retry behavior

The browser disables Start until the previous session's cleanup request succeeds or confirms that no session is accessible to the current principal. If that request fails, the browser keeps the application session ID and enables Stop for another attempt. An authentication rejection triggers one authentication refresh and Close retry. The complete cleanup request sequence has a 15-second deadline, so a stalled request also leaves Stop retryable.
Expand All @@ -68,7 +70,7 @@ The public speaking state follows the Realtime API's [WebRTC output-buffer event

### Cancellation records

The demo retains cancelled application session IDs in memory until the server stops, so a delayed request cannot recreate a cancelled session after the live-session expiry sweep. A production implementation must account for this storage growth and keep cancellation records for at least as long as it accepts delayed or replayed setup requests.
The demo retains cancelled application session IDs until the server stops, independently of the live-session expiry sweep. Live sessions and cancellation records share a capacity budget of 1,000 distinct application session IDs; the server returns 503 when a new setup or close cannot be admitted. Existing sessions can still close at capacity. A production implementation should set a bounded setup-acceptance window and matching record expiry, as well as rate limits.

### Tool authorization and prompt injection

Expand Down
1 change: 1 addition & 0 deletions examples/realtime-server-controlled/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ const auth = new DemoAuthStore({
const server = createApiServer({ appOrigin, auth, controller, sessions });

const cleanupTimer = setInterval(() => {
auth.removeExpired();
void sessions.closeExpired(30 * 60 * 1000);
}, 60_000);
cleanupTimer.unref();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,11 @@ import {
type IncomingMessage,
type ServerResponse,
} from 'node:http';
import type { DemoAuthStore, DemoPrincipal } from './demoAuth';
import {
DemoAuthCapacityError,
type DemoAuthStore,
type DemoPrincipal,
} from './demoAuth';
import { MAX_SDP_BYTES, SdpPolicyError, assertAudioOnlySdp } from './sdpPolicy';
import {
SessionCapacityError,
Expand Down Expand Up @@ -259,6 +263,8 @@ export function createApiServer(options: {
sendJson(response, 409, { error: error.message });
} else if (error instanceof SessionCapacityError) {
sendJson(response, 503, { error: error.message });
} else if (error instanceof DemoAuthCapacityError) {
sendJson(response, 503, { error: error.message });
} else {
sendJson(response, 502, {
error: 'Could not complete the voice session request.',
Expand Down
30 changes: 30 additions & 0 deletions examples/realtime-server-controlled/src/server/demoAuth.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,15 @@ import type { IncomingMessage } from 'node:http';
const COOKIE_NAME = 'realtime_demo_session';
const SESSION_TTL_SECONDS = 60 * 60;
const SESSION_TTL_MS = SESSION_TTL_SECONDS * 1000;
const MAX_DEMO_PRINCIPALS = 1_000;
const EXPIRY_SWEEP_INTERVAL_MS = 60_000;

export class DemoAuthCapacityError extends Error {
constructor() {
super('The demo authentication capacity has been reached.');
this.name = 'DemoAuthCapacityError';
}
}

export type DemoPrincipal = {
csrfToken: string;
Expand Down Expand Up @@ -34,6 +43,7 @@ export class DemoAuthStore {
readonly #principals = new Map<string, StoredPrincipal>();
readonly #secureCookie: boolean;
readonly #now: () => number;
#nextExpirySweepAt = 0;

constructor(options: { secureCookie: boolean; now?: () => number }) {
this.#secureCookie = options.secureCookie;
Expand All @@ -53,6 +63,13 @@ export class DemoAuthStore {
};
}

if (this.#principals.size >= MAX_DEMO_PRINCIPALS) {
this.removeExpired();
if (this.#principals.size >= MAX_DEMO_PRINCIPALS) {
throw new DemoAuthCapacityError();
}
}

const token = randomBytes(32).toString('base64url');
const ownerId = randomUUID();
const principal: StoredPrincipal = {
Expand All @@ -74,6 +91,19 @@ export class DemoAuthStore {
return this.#findStoredPrincipal(request);
}

removeExpired(): void {
const now = this.#now();
if (now < this.#nextExpirySweepAt) {
return;
}
this.#nextExpirySweepAt = now + EXPIRY_SWEEP_INTERVAL_MS;
for (const [token, principal] of this.#principals) {
if (principal.expiresAt <= now) {
this.#principals.delete(token);
}
}
}

#findStoredPrincipal(request: IncomingMessage): StoredPrincipal | null {
const token = parseCookies(request.headers.cookie).get(COOKIE_NAME);
if (!token) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ export type PublicEventClient = {
};

type SessionState = 'active' | 'closing' | 'creating';
const SESSION_RECORD_CAPACITY = 1_000;

export class SessionConflictError extends Error {
constructor(message = 'This user already has an active voice session.') {
Expand Down Expand Up @@ -79,7 +80,10 @@ export class SessionManager {
'This voice session identifier is already in use.',
);
}
if (this.#sessions.size >= this.#maxSessions) {
if (
this.#sessions.size >= this.#maxSessions ||
!this.#hasRecordCapacity()
) {
throw new SessionCapacityError();
}

Expand All @@ -95,6 +99,17 @@ export class SessionManager {
return id;
}

#hasRecordCapacity(): boolean {
// A cancelled session keeps its live record until hangup succeeds.
let count = this.#cancelledSetups.size;
for (const id of this.#sessions.keys()) {
if (!this.#cancelledSetups.has(id)) {
count++;
}
}
return count < SESSION_RECORD_CAPACITY;
}

async cancel(id: string, ownerId: string): Promise<boolean> {
const session = this.#sessions.get(id);
if (session && session.ownerId !== ownerId) {
Expand All @@ -106,6 +121,9 @@ export class SessionManager {
return false;
}

if (!session && !cancelledOwner && !this.#hasRecordCapacity()) {
throw new SessionCapacityError();
}
this.#cancelledSetups.set(id, ownerId);
if (session) {
await this.close(id, ownerId);
Expand Down
63 changes: 61 additions & 2 deletions examples/realtime-server-controlled/test/apiServer.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import { randomUUID } from 'node:crypto';
import type { AddressInfo } from 'node:net';
import { request as httpRequest } from 'node:http';
import { request as httpRequest, type IncomingMessage } from 'node:http';
import { afterEach, describe, expect, it, vi } from 'vitest';
import { createApiServer } from '../src/server/apiServer';
import { DemoAuthStore } from '../src/server/demoAuth';
Expand Down Expand Up @@ -39,6 +39,7 @@ afterEach(async () => {

async function createHarness(hangup = vi.fn(async () => {}), now = Date.now) {
const sessions = new SessionManager({ hangup, now });
const auth = new DemoAuthStore({ secureCookie: false, now });
const start = vi.fn(
async ({ ownerId, sessionId }: { ownerId: string; sessionId: string }) => {
sessions.reserve(ownerId, sessionId);
Expand All @@ -50,7 +51,7 @@ async function createHarness(hangup = vi.fn(async () => {}), now = Date.now) {
const controller = { start } as unknown as VoiceController;
const server = createApiServer({
appOrigin,
auth: new DemoAuthStore({ secureCookie: false, now }),
auth,
controller,
sessions,
});
Expand All @@ -67,6 +68,7 @@ async function createHarness(hangup = vi.fn(async () => {}), now = Date.now) {
}

return {
auth,
baseUrl,
cookie,
csrfToken: authBody.csrfToken,
Expand All @@ -77,6 +79,42 @@ async function createHarness(hangup = vi.fn(async () => {}), now = Date.now) {
}

describe('createApiServer', () => {
it('limits cookie-free bootstrap requests without denying an existing principal', async () => {
let now = 0;
const harness = await createHarness(
vi.fn(async () => {}),
() => now,
);
for (let index = 1; index < 1_000; index++) {
harness.auth.getOrCreate({ headers: {} } as IncomingMessage);
}

const overflow = await fetch(`${harness.baseUrl}/api/auth/session`);
expect(overflow.status).toBe(503);
expect(await overflow.json()).toEqual({
error: 'The demo authentication capacity has been reached.',
});
expect(overflow.headers.get('set-cookie')).toBeNull();

const unknownCookie = await fetch(`${harness.baseUrl}/api/auth/session`, {
headers: { Cookie: 'realtime_demo_session=unknown' },
});
expect(unknownCookie.status).toBe(503);

const renewed = await fetch(`${harness.baseUrl}/api/auth/session`, {
headers: { Cookie: harness.cookie },
});
expect(renewed.status).toBe(200);
expect((await renewed.json()) as { csrfToken: string }).toEqual({
csrfToken: harness.csrfToken,
});

now = 60 * 60 * 1000 + 1;
const afterExpiry = await fetch(`${harness.baseUrl}/api/auth/session`);
expect(afterExpiry.status).toBe(200);
expect(afterExpiry.headers.get('set-cookie')).toBeTruthy();
});

it('recovers Stop and Start after cleanup failure and demo authentication expiry', async () => {
let now = 1_000;
const hangup = vi
Expand Down Expand Up @@ -324,6 +362,27 @@ describe('createApiServer', () => {
});
});

it('returns a capacity error for new close-first IDs after the record budget fills', async () => {
const harness = await createHarness();
for (let index = 0; index < 1_000; index++) {
await harness.sessions.cancel(randomUUID(), 'owner-1');
}

const response = await fetch(
`${harness.baseUrl}/api/realtime/sessions/${randomUUID()}/close`,
{
method: 'POST',
headers: {
Cookie: harness.cookie,
Origin: appOrigin,
'X-CSRF-Token': harness.csrfToken,
},
},
);

expect(response.status).toBe(503);
});

it('rejects malformed cancellation identifiers', async () => {
const harness = await createHarness();
const response = await fetch(
Expand Down
23 changes: 23 additions & 0 deletions examples/realtime-server-controlled/test/demoAuth.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,29 @@ function request(cookie?: string): IncomingMessage {
}

describe('DemoAuthStore', () => {
it('recovers expired capacity within one sweep interval', () => {
let now = 0;
const auth = new DemoAuthStore({ now: () => now, secureCookie: false });
const initial = auth.getOrCreate(request());
const cookie = initial.setCookie!.split(';', 1)[0]!;
for (let index = 1; index < 1_000; index++) {
auth.getOrCreate(request());
}
now = 60 * 60 * 1000 - 1;
expect(() => auth.getOrCreate(request())).toThrow('capacity');
// Renewal remains available even while anonymous admission is full.
expect(auth.getOrCreate(request(cookie)).principal.ownerId).toBe(
initial.principal.ownerId,
);
now += 1;
expect(() => auth.getOrCreate(request())).toThrow('capacity');
now += 60_000;
expect(() => auth.getOrCreate(request())).not.toThrow();
expect(auth.authenticate(request(cookie))?.ownerId).toBe(
initial.principal.ownerId,
);
});

it('renews the same principal when a voice session is about to start', () => {
let now = 0;
const auth = new DemoAuthStore({
Expand Down
52 changes: 52 additions & 0 deletions examples/realtime-server-controlled/test/sessionManager.test.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { randomUUID } from 'node:crypto';
import { describe, expect, it, vi } from 'vitest';
import { NotFoundError } from 'openai';
import {
Expand Down Expand Up @@ -157,6 +158,57 @@ describe('SessionManager', () => {
);
});

it('bounds close-first cancellation records while preserving admitted cleanup', async () => {
const sessions = new SessionManager({ hangup: vi.fn(async () => {}) });
const activeId = sessions.reserve('active-owner');
for (let index = 0; index < 999; index++) {
const id = `00000000-0000-4000-8000-${index.toString(16).padStart(12, '0')}`;
await expect(sessions.cancel(id, 'owner-1')).resolves.toBe(true);
}

const rejectedId = '00000000-0000-4000-8000-000000001000';
await expect(sessions.cancel(rejectedId, 'owner-1')).rejects.toThrow(
'capacity',
);
expect(() => sessions.reserve('owner-1', rejectedId)).toThrow('capacity');
await expect(sessions.cancel(activeId, 'active-owner')).resolves.toBe(true);
await expect(sessions.cancel(activeId, 'active-owner')).resolves.toBe(true);
expect(() => sessions.reserve('active-owner', activeId)).toThrow(
'voice session setup was cancelled',
);
});

it('counts a cancelled session once while its hangup is pending', async () => {
let finishHangup!: () => void;
const hangup = vi.fn(
() =>
new Promise<void>((resolve) => {
finishHangup = resolve;
}),
);
const sessions = new SessionManager({ hangup });
const activeId = sessions.reserve('active-owner');
sessions.attachCall(activeId, 'rtc_pending');
for (let index = 0; index < 998; index++) {
await sessions.cancel(randomUUID(), 'owner-1');
}

const closing = sessions.cancel(activeId, 'active-owner');
await vi.waitFor(() => expect(hangup).toHaveBeenCalledOnce());
const finalId = sessions.reserve('last-owner');
expect(() => sessions.reserve('overflow-owner')).toThrow('capacity');
await expect(sessions.cancel(randomUUID(), 'owner-1')).rejects.toThrow(
'capacity',
);
await expect(sessions.cancel(finalId, 'last-owner')).resolves.toBe(true);
finishHangup();
await expect(closing).resolves.toBe(true);
expect(() => sessions.reserve('active-owner', activeId)).toThrow(
'setup was cancelled',
);
expect(() => sessions.reserve('overflow-owner')).toThrow('capacity');
});

it('requires ownership for subscription and close', async () => {
const sessions = new SessionManager({ hangup: vi.fn(async () => {}) });
const id = sessions.reserve('owner-1');
Expand Down
Loading