fix(ai): drain, transcription metering and credential re-check on the realtime relay

Shutdown now drains: a new onServerDrainStart hook (run when SIGTERM
starts the drain, before the listener closes) makes the relay refuse new
sessions with 1012, open ones keep running through the drain window, and
at onServerPrepareShutdown each gets up to 5 s to finish an in-flight
response before closing with 1012 so the client reconnects elsewhere.

Input transcription is metered. The relay reads the type from each text
frame's first 96 bytes and, after forwarding, parses response.done,
input_audio_transcription.completed and session.created/updated (for the
transcription model) on setImmediate, so metering never holds up frames.
Unknown transcription models bill at the highest known rate.

Credentials are re-checked: a successful auth is cached in the kv.js
singleton for 60 s under a hash of the token, and a 15 s sweep
re-authenticates sessions whose entry expired, closing revoked or
suspended ones with invalid_api_key. Lookup errors keep the session and
retry on the next sweep.

src/backend/server.ts

@@ -1633,7 +1633,11 @@ export class PuterServer {
      */
     async #runHook(
         order: 'bottomUp' | 'topDown',
-        hook: 'onServerStart' | 'onServerPrepareShutdown' | 'onServerShutdown',
+        hook:
+            | 'onServerStart'
+            | 'onServerDrainStart'
+            | 'onServerPrepareShutdown'
+            | 'onServerShutdown',
     ) {
         const layers = [
             this.clients,
@@ -1664,6 +1668,11 @@ export class PuterServer {
 
     async prepareShutdown() {
         if (this.#server) {
+            try {
+                await this.#runHook('bottomUp', 'onServerDrainStart');
+            } catch (err) {
+                console.error('[server] onServerDrainStart failed', err);
+            }
             this.#server.close(async () => {
                 console.log(
                     'PuterServer has stopped accepting new connections',

src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts

@@ -9,9 +9,12 @@ import {
     setupTestServer,
     type TestUserCredentials,
 } from '../../testUtil';
+import { kv } from '../../util/kvSingleton';
 import {
     REALTIME_PATH,
+    realtimeAuthKey,
     realtimeTokenFrom,
+    realtimeTranscriptionRows,
     realtimeUsageRows,
 } from './OpenAIRealtimeRelayService';
 
@@ -87,6 +90,58 @@ describe('realtimeUsageRows', () => {
     });
 });
 
+describe('realtimeTranscriptionRows', () => {
+    it('prices token usage by transcription model', () => {
+        const rows = realtimeTranscriptionRows('gpt-4o-mini-transcribe', {
+            type: 'tokens',
+            input_tokens: 30,
+            input_token_details: { text_tokens: 10, audio_tokens: 20 },
+            output_tokens: 5,
+        });
+        expect(rows.map((r) => [r.usageType, r.costOverride])).toEqual([
+            [
+                'openai:gpt-4o-mini-transcribe:realtime:transcription_text_input',
+                10 * 125,
+            ],
+            [
+                'openai:gpt-4o-mini-transcribe:realtime:transcription_audio_input',
+                20 * 300,
+            ],
+            [
+                'openai:gpt-4o-mini-transcribe:realtime:transcription_text_output',
+                5 * 500,
+            ],
+        ]);
+    });
+
+    it('prices duration usage per second', () => {
+        expect(
+            realtimeTranscriptionRows('whisper-1', {
+                type: 'duration',
+                seconds: 2.5,
+            }),
+        ).toEqual([
+            {
+                usageType: 'openai:whisper-1:realtime:transcription_seconds',
+                usageAmount: 2.5,
+                costOverride: 25_000,
+            },
+        ]);
+    });
+
+    it('bills an unknown model at the highest known rate', () => {
+        const [row] = realtimeTranscriptionRows(undefined, {
+            type: 'tokens',
+            input_tokens: 10,
+        });
+        expect(row).toEqual({
+            usageType: 'openai:unknown:realtime:transcription_audio_input',
+            usageAmount: 10,
+            costOverride: 10 * 600,
+        });
+    });
+});
+
 describe('OpenAIRealtimeRelayService relay', () => {
     let server: PuterServer;
     let upstream: WebSocketServer;
@@ -116,6 +171,47 @@ describe('OpenAIRealtimeRelayService relay', () => {
                 }
                 const event = JSON.parse(data.toString());
                 ws.send(JSON.stringify({ type: 'echo', event }));
+                if (event.type === 'transcribe') {
+                    ws.send(
+                        JSON.stringify({
+                            type: 'session.updated',
+                            session: {
+                                audio: {
+                                    input: {
+                                        transcription: {
+                                            model: 'gpt-4o-transcribe',
+                                        },
+                                    },
+                                },
+                            },
+                        }),
+                    );
+                    ws.send(
+                        JSON.stringify({
+                            type: 'conversation.item.input_audio_transcription.completed',
+                            usage: {
+                                type: 'tokens',
+                                input_tokens: 4,
+                                input_token_details: { audio_tokens: 4 },
+                                output_tokens: 2,
+                            },
+                        }),
+                    );
+                }
+                // Holds a response open so the drain has one in flight.
+                if (event.type === 'slow.response') {
+                    ws.send(JSON.stringify({ type: 'response.created' }));
+                    setTimeout(
+                        () =>
+                            ws.send(
+                                JSON.stringify({
+                                    type: 'response.done',
+                                    response: {},
+                                }),
+                            ),
+                        300,
+                    );
+                }
                 if (event.type === 'response.create') {
                     ws.send(
                         JSON.stringify({
@@ -279,6 +375,70 @@ describe('OpenAIRealtimeRelayService relay', () => {
         expect(relay.stats().upstreamFailed).toBe(before + 1);
     });
 
+    it('meters input transcription at the session transcription model', async () => {
+        const spy = vi.spyOn(server.services.metering, 'batchIncrementUsages');
+        spy.mockClear();
+        const { ws, waitFor } = open(
+            `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`,
+        );
+        await new Promise((r) => ws.once('open', r));
+        ws.send(JSON.stringify({ type: 'transcribe' }));
+        await waitFor('conversation.item.input_audio_transcription.completed');
+        await vi.waitFor(() => expect(spy).toHaveBeenCalled());
+        const [, rows] = spy.mock.calls.at(-1)!;
+        expect(rows.map((r) => [r.usageType, r.costOverride])).toEqual([
+            [
+                'openai:gpt-4o-transcribe:realtime:transcription_audio_input',
+                4 * 600,
+            ],
+            [
+                'openai:gpt-4o-transcribe:realtime:transcription_text_output',
+                2 * 1000,
+            ],
+        ]);
+        ws.close();
+    });
+
+    it('closes a session whose credential stops authenticating', async () => {
+        const relay = server.services.openaiRealtime;
+        const auth = vi.spyOn(server.services.auth, 'authenticate');
+        const { ws, closed, waitFor } = open(
+            `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`,
+        );
+        await waitFor('session.created');
+
+        // Within the cache window nothing is re-checked.
+        auth.mockClear();
+        await relay.recheckSessions();
+        expect(auth).not.toHaveBeenCalled();
+        expect(ws.readyState).toBe(WebSocket.OPEN);
+
+        auth.mockResolvedValue({ invalid: true } as never);
+        kv.del(realtimeAuthKey(user.token));
+        await relay.recheckSessions();
+        const error = (await waitFor('error')) as {
+            error: { code: string };
+        };
+        expect(error.error.code).toBe('invalid_api_key');
+        expect((await closed).code).toBe(1008);
+        auth.mockRestore();
+    });
+
+    it('keeps a session when the re-check lookup itself fails', async () => {
+        const relay = server.services.openaiRealtime;
+        const auth = vi.spyOn(server.services.auth, 'authenticate');
+        const { ws, waitFor } = open(
+            `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`,
+        );
+        await waitFor('session.created');
+        auth.mockRejectedValue(new Error('db down'));
+        kv.del(realtimeAuthKey(user.token));
+        await relay.recheckSessions();
+        expect(ws.readyState).toBe(WebSocket.OPEN);
+        auth.mockRestore();
+        ws.close();
+    });
+
     it('reports counters, event-loop delay and memory', async () => {
         const relay = server.services.openaiRealtime;
         await vi.waitFor(() => expect(relay.stats().active).toBe(0));
@@ -295,4 +455,30 @@ describe('OpenAIRealtimeRelayService relay', () => {
             expect(Number.isFinite(stats[key])).toBe(true);
         }
     });
+
+    // 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}`,
+        );
+        await live.waitFor('session.created');
+        live.ws.send(JSON.stringify({ type: 'slow.response' }));
+        await live.waitFor('response.created');
+
+        relay.onServerDrainStart();
+        const late = open(
+            `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`,
+        );
+        const error = (await late.waitFor('error')) as {
+            error: { code: string };
+        };
+        expect(error.error.code).toBe('server_restarting');
+        expect((await late.closed).code).toBe(1012);
+        expect(live.ws.readyState).toBe(WebSocket.OPEN);
+
+        await relay.onServerPrepareShutdown();
+        expect((await live.closed).code).toBe(1012);
+        expect(live.events.some((e) => e.type === 'response.done')).toBe(true);
+    });
 });

src/backend/services/realtime/OpenAIRealtimeRelayService.ts

@@ -17,6 +17,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+import { createHash } from 'node:crypto';
 import type { IncomingMessage, Server as HttpServer } from 'node:http';
 import type { Duplex } from 'node:stream';
 import {
@@ -31,7 +32,9 @@ import {
     CONCURRENT_SLOT_TTL_MS,
     acquireConcurrent,
 } from '../../core/http/middleware/rateLimit.js';
+import { SPEECH_TO_TEXT_COSTS } from '../../drivers/ai-speech2txt/costs.js';
 import { readProviderKey } from '../../drivers/util/providerRegistry.js';
+import { kv } from '../../util/kvSingleton.js';
 import type { UsageInput } from '../metering/types.js';
 import { decideSocketAuth } from '../socket/SocketService.js';
 import { PuterService } from '../types.js';
@@ -98,6 +101,29 @@ export interface RealtimeUsage {
     output_token_details?: { text_tokens?: number; audio_tokens?: number };
 }
 
+/** Micro-cents per token for input transcription billed in tokens. */
+const TRANSCRIPTION_TOKEN_PRICES: Record<
+    string,
+    { textIn: number; audioIn: number; textOut: number }
+> = {
+    'gpt-4o-transcribe': { textIn: 250, audioIn: 600, textOut: 1000 },
+    'gpt-4o-mini-transcribe': { textIn: 125, audioIn: 300, textOut: 500 },
+};
+/** Micro-cents per second for input transcription billed by duration. */
+const TRANSCRIPTION_SECOND_PRICES: Record<string, number> = {
+    'whisper-1': 10_000,
+    'gpt-transcribe': SPEECH_TO_TEXT_COSTS['openai:gpt-transcribe:second'],
+};
+
+/** The `usage` on `conversation.item.input_audio_transcription.completed`. */
+export interface RealtimeTranscriptionUsage {
+    type?: 'tokens' | 'duration';
+    input_tokens?: number;
+    output_tokens?: number;
+    input_token_details?: { text_tokens?: number; audio_tokens?: number };
+    seconds?: number;
+}
+
 const OPENAI_KEY_PROTOCOL = 'openai-insecure-api-key.';
 const MAX_CLIENT_PAYLOAD = 16 * 1024 * 1024;
 // Client reads pause past this much buffered before the upstream opens.
@@ -112,6 +138,16 @@ const HEARTBEAT_MS = 30_000;
 const UPSTREAM_HANDSHAKE_MS = 10_000;
 const STATS_INTERVAL_MS = 30_000;
 const DEFAULT_MAX_PER_USER = 20;
+// Credentials are re-authenticated once their kv entry expires.
+const AUTH_RECHECK_SEC = 60;
+const AUTH_SWEEP_MS = 15_000;
+// At shutdown, how long a session may finish its in-flight response.
+const DRAIN_RESPONSE_MS = 5_000;
+// RFC 6455 "Service Restart": the client should reconnect.
+const CLOSE_RESTART = 1012;
+const TYPE_PREFIX = /^\{\s*"type"\s*:\s*"([^"]+)"/;
+const TRANSCRIPTION_DONE =
+    'conversation.item.input_audio_transcription.completed';
 
 /**
  * The credential a realtime client presents, in the places OpenAI's own clients
@@ -179,6 +215,74 @@ export const realtimeUsageRows = (
         }));
 };
 
+/**
+ * Usage rows for one completed input transcription. A model missing from the
+ * tables is billed at the highest known rate, so it is never free.
+ */
+export const realtimeTranscriptionRows = (
+    model: string | undefined,
+    usage: RealtimeTranscriptionUsage | undefined,
+): UsageInput[] => {
+    if (!usage) return [];
+    const n = (v: number | undefined) =>
+        typeof v === 'number' && Number.isFinite(v) && v > 0 ? v : 0;
+    const known =
+        model &&
+        (TRANSCRIPTION_TOKEN_PRICES[model] ||
+            TRANSCRIPTION_SECOND_PRICES[model])
+            ? model
+            : 'unknown';
+    const usageType = (kind: string) =>
+        `openai:${known}:realtime:transcription_${kind}`;
+    if (usage.type === 'duration') {
+        const seconds = n(usage.seconds);
+        const ucents =
+            TRANSCRIPTION_SECOND_PRICES[known] ??
+            Math.max(...Object.values(TRANSCRIPTION_SECOND_PRICES));
+        if (seconds === 0) return [];
+        return [
+            {
+                usageType: usageType('seconds'),
+                usageAmount: seconds,
+                costOverride: Math.ceil(seconds * ucents),
+            },
+        ];
+    }
+    const price =
+        TRANSCRIPTION_TOKEN_PRICES[known] ??
+        TRANSCRIPTION_TOKEN_PRICES['gpt-4o-transcribe'];
+    const details = usage.input_token_details;
+    const audioIn = details ? n(details.audio_tokens) : n(usage.input_tokens);
+    const counts: [string, number, number][] = [
+        ['text_input', n(details?.text_tokens), price.textIn],
+        ['audio_input', audioIn, price.audioIn],
+        ['text_output', n(usage.output_tokens), price.textOut],
+    ];
+    return counts
+        .filter(([, amount]) => amount > 0)
+        .map(([kind, amount, ucents]) => ({
+            usageType: usageType(kind),
+            usageAmount: amount,
+            costOverride: amount * ucents,
+        }));
+};
+
+/** The transcription model a `session.created`/`session.updated` configures. */
+const transcriptionModelOf = (session: unknown): string | undefined => {
+    const s = session as {
+        audio?: { input?: { transcription?: { model?: unknown } | null } };
+        input_audio_transcription?: { model?: unknown } | null;
+    } | null;
+    const model =
+        s?.audio?.input?.transcription?.model ??
+        s?.input_audio_transcription?.model;
+    return typeof model === 'string' ? model : undefined;
+};
+
+/** The kv key whose presence means `token` was verified recently. */
+export const realtimeAuthKey = (token: string): string =>
+    `realtime:auth:${createHash('sha256').update(token).digest('base64url')}`;
+
 /** An error event in OpenAI's realtime shape. */
 const errorEvent = (code: string, message: string): string =>
     JSON.stringify({
@@ -212,8 +316,14 @@ export class OpenAIRealtimeRelayService extends PuterService {
     #sessions = new Set<{
         close: (code: number, reason: string) => void;
         heartbeat: () => void;
+        recheck: () => Promise<void>;
+        drain: () => Promise<void>;
     }>();
     #heartbeatTimer: ReturnType<typeof setInterval> | null = null;
+    #authTimer: ReturnType<typeof setInterval> | null = null;
+    // One re-authentication per token at a time, shared by its sessions.
+    #authChecks = new Map<string, Promise<Actor | null>>();
+    #draining = false;
     #stats = {
         opened: 0,
         closed: 0,
@@ -270,15 +380,37 @@ export class OpenAIRealtimeRelayService extends PuterService {
             for (const session of this.#sessions) session.heartbeat();
         }, HEARTBEAT_MS);
         this.#heartbeatTimer.unref?.();
+        this.#authTimer = setInterval(
+            () => void this.recheckSessions(),
+            AUTH_SWEEP_MS,
+        );
+        this.#authTimer.unref?.();
     }
 
-    override onServerPrepareShutdown(): void {
+    /** Re-authenticates every session whose token's kv entry has expired. */
+    async recheckSessions(): Promise<void> {
+        await Promise.all([...this.#sessions].map((s) => s.recheck()));
+    }
+
+    /** New sessions are refused; open ones keep running through the drain. */
+    override onServerDrainStart(): void {
+        this.#draining = true;
+        console.info(
+            `[realtime] draining; ${this.#sessions.size} sessions stay open`,
+        );
+    }
+
+    /**
+     * Each session gets a moment to finish its in-flight response, then closes
+     * with 1012 so the client reconnects to another node.
+     */
+    override async onServerPrepareShutdown(): Promise<void> {
+        this.#draining = true;
         if (this.#statsTimer) clearInterval(this.#statsTimer);
         if (this.#heartbeatTimer) clearInterval(this.#heartbeatTimer);
+        if (this.#authTimer) clearInterval(this.#authTimer);
         this.#loopDelay?.disable();
-        for (const session of this.#sessions) {
-            session.close(1001, 'server shutting down');
-        }
+        await Promise.all([...this.#sessions].map((s) => s.drain()));
         this.#wss?.close();
     }
 
@@ -336,6 +468,14 @@ export class OpenAIRealtimeRelayService extends PuterService {
         let renewTimer: ReturnType<typeof setInterval> | null = null;
         let budgetTimer: ReturnType<typeof setTimeout> | null = null;
         let stallTimer: ReturnType<typeof setTimeout> | null = null;
+        const token = realtimeTokenFrom(req, url);
+        let actor: Actor | null = null;
+        let rechecking = false;
+        // Responses started upstream and not yet done, for the drain.
+        let inFlight = 0;
+        let onIdle: (() => void) | null = null;
+        // Decoded upstream state the meter needs.
+        const decoded: { transcriptionModel?: string } = {};
 
         // Client reads stay paused while any reason holds.
         const holds = new Set<'pending' | 'upstream' | 'budget'>();
@@ -358,6 +498,7 @@ export class OpenAIRealtimeRelayService extends PuterService {
                 if (renewTimer) clearInterval(renewTimer);
                 if (budgetTimer) clearTimeout(budgetTimer);
                 if (stallTimer) clearTimeout(stallTimer);
+                onIdle?.();
                 void Promise.all(slots.map((s) => s.release()));
                 if (client.readyState === WebSocket.OPEN) {
                     client.close(code, reason);
@@ -380,19 +521,63 @@ export class OpenAIRealtimeRelayService extends PuterService {
                 alive = false;
                 if (client.readyState === WebSocket.OPEN) client.ping();
             },
+            // Called every AUTH_SWEEP_MS; re-authenticates once the token's kv
+            // entry has expired. A failed lookup is retried on the next sweep.
+            recheck: async () => {
+                if (!actor || !token || closed || rechecking) return;
+                if (kv.get(realtimeAuthKey(token)) !== undefined) return;
+                rechecking = true;
+                try {
+                    const next = await this.#authenticateCached(token, req);
+                    if (closed) return;
+                    if (!next || next.user?.uuid !== actor.user?.uuid) {
+                        reject(
+                            'invalid_api_key',
+                            'Credentials are no longer valid',
+                        );
+                        return;
+                    }
+                    actor = next;
+                } catch (err) {
+                    console.warn('[realtime] credential recheck failed', err);
+                } finally {
+                    rechecking = false;
+                }
+            },
+            drain: async () => {
+                if (inFlight > 0 && !closed) {
+                    await new Promise<void>((resolve) => {
+                        const timer = setTimeout(resolve, DRAIN_RESPONSE_MS);
+                        onIdle = () => {
+                            clearTimeout(timer);
+                            resolve();
+                        };
+                    });
+                }
+                session.close(CLOSE_RESTART, 'server restarting');
+            },
         };
         this.#sessions.add(session);
         this.#stats.opened++;
         client.on('pong', () => (alive = true));
 
-        const reject = (code: string, message: string) => {
+        const reject = (code: string, message: string, closeCode = 1008) => {
             this.#stats.rejected++;
             if (client.readyState === WebSocket.OPEN) {
                 client.send(errorEvent(code, message));
             }
-            session.close(1008, code);
+            session.close(closeCode, code);
         };
 
+        if (this.#draining) {
+            reject(
+                'server_restarting',
+                'Server is restarting; reconnect',
+                CLOSE_RESTART,
+            );
+            return;
+        }
+
         // Token bucket over client bytes: one max-size frame of burst, then
         // reads pause until the sustained rate catches up.
         const rate =
@@ -457,9 +642,11 @@ export class OpenAIRealtimeRelayService extends PuterService {
             return;
         }
 
-        const actor = await this.#authenticate(req, url).catch(
-            (): null => null,
-        );
+        try {
+            actor = token ? await this.#authenticateCached(token, req) : null;
+        } catch {
+            actor = null;
+        }
         if (closed) return;
         if (!actor) {
             reject('invalid_api_key', 'Authentication failed');
@@ -558,9 +745,31 @@ export class OpenAIRealtimeRelayService extends PuterService {
                     SLOW_CLIENT_MS,
                 );
             }
-            if (!isBinary && buf.subarray(0, 64).includes('"response.done"')) {
-                this.#meter(actor, model, buf, () =>
-                    reject('insufficient_funds', 'No usage left for request'),
+            if (isBinary) return;
+            // Audio deltas dominate; only the type prefix is read per frame.
+            const type = TYPE_PREFIX.exec(
+                buf.subarray(0, 96).toString('latin1'),
+            )?.[1];
+            if (type === 'response.created') inFlight++;
+            if (type === 'response.done') {
+                inFlight = Math.max(0, inFlight - 1);
+                if (inFlight === 0) onIdle?.();
+            }
+            if (
+                type === 'response.done' ||
+                type === TRANSCRIPTION_DONE ||
+                type === 'session.created' ||
+                type === 'session.updated'
+            ) {
+                // Parsed and billed after the frame is forwarded, in arrival
+                // order, so metering never holds up the relay.
+                setImmediate(() =>
+                    this.#decode(type, buf, model, decoded, actor!, () =>
+                        reject(
+                            'insufficient_funds',
+                            'No usage left for request',
+                        ),
+                    ),
                 );
             }
         });
@@ -594,34 +803,78 @@ export class OpenAIRealtimeRelayService extends PuterService {
         });
     }
 
-    #meter(
-        actor: Actor,
-        model: string,
+    async #decode(
+        type: string,
         buf: Buffer,
+        model: string,
+        decoded: { transcriptionModel?: string },
+        actor: Actor,
         onExhausted: () => void,
-    ): void {
-        let usage: RealtimeUsage | undefined;
+    ): Promise<void> {
+        let event: {
+            session?: unknown;
+            response?: { usage?: RealtimeUsage };
+            usage?: RealtimeTranscriptionUsage;
+        };
         try {
-            usage = JSON.parse(buf.toString('utf8'))?.response?.usage;
+            event = JSON.parse(buf.toString('utf8'));
         } catch {
             return;
         }
-        const rows = realtimeUsageRows(model, usage);
+        if (type === 'session.created' || type === 'session.updated') {
+            decoded.transcriptionModel = transcriptionModelOf(event?.session);
+            return;
+        }
+        const rows =
+            type === 'response.done'
+                ? realtimeUsageRows(model, event?.response?.usage)
+                : realtimeTranscriptionRows(
+                      decoded.transcriptionModel,
+                      event?.usage,
+                  );
         if (rows.length === 0) return;
-        void this.services.metering
-            .batchIncrementUsages(actor, rows)
-            .then(() => this.services.metering.hasEnoughCredits(actor, 1))
-            .then((ok) => {
-                if (!ok) onExhausted();
-            })
-            .catch((err: unknown) =>
-                console.warn('[realtime] metering failed', err),
-            );
+        try {
+            await this.services.metering.batchIncrementUsages(actor, rows);
+            if (!(await this.services.metering.hasEnoughCredits(actor, 1))) {
+                onExhausted();
+            }
+        } catch (err) {
+            console.warn('[realtime] metering failed', err);
+        }
     }
 
-    async #authenticate(req: IncomingMessage, url: URL): Promise<Actor | null> {
-        const token = realtimeTokenFrom(req, url);
-        if (!token) return null;
+    /**
+     * Authenticates `token`, caching success in kv for AUTH_RECHECK_SEC so
+     * sessions sharing a token re-check it once per expiry. Throws on lookup
+     * errors, which callers treat as "unknown" rather than "revoked".
+     */
+    #authenticateCached(
+        token: string,
+        req: IncomingMessage,
+    ): Promise<Actor | null> {
+        const key = realtimeAuthKey(token);
+        let check = this.#authChecks.get(key);
+        if (!check) {
+            check = (async () => {
+                try {
+                    const actor = await this.#authenticate(token, req);
+                    if (actor) {
+                        kv.set(key, actor.user!.uuid, { EX: AUTH_RECHECK_SEC });
+                    }
+                    return actor;
+                } finally {
+                    this.#authChecks.delete(key);
+                }
+            })();
+            this.#authChecks.set(key, check);
+        }
+        return check;
+    }
+
+    async #authenticate(
+        token: string,
+        req: IncomingMessage,
+    ): Promise<Actor | null> {
         const ua = req.headers['user-agent'];
         const result = await this.services.auth.authenticate(token, {
             ip: req.socket.remoteAddress,

src/backend/services/realtime/README.md

@@ -166,7 +166,7 @@ client ──ws──▶ api.<domain>/puterai/openai/v1/realtime?model=…
 2. The client socket is **accepted before authenticating** (see [Quirks](#quirks)).
 3. The relay checks the model, then the credential, then that the account has at least 1 cent of usage left, then takes a per-user concurrency slot. Any failure sends an OpenAI-shaped `error` event and closes with `1008`.
 4. It dials OpenAI with the server key. Client frames sent while that is in flight are buffered and replayed in order, keeping their text/binary type.
-5. From then on frames pass through untouched in both directions. Only `response.done` is parsed, to meter its `usage`.
+5. From then on frames pass through untouched in both directions. Only `response.done`, input transcription completions and `session.created`/`session.updated` are parsed, after forwarding, to meter usage.
 6. Either side closing tears down the other.
 
 ### Credentials
@@ -191,6 +191,8 @@ Prices are per 1M tokens, taken from OpenAI's pricing page on 2026-10-15. `REALT
 
 Each `response.done` records usage rows `openai:<model>:realtime:<kind>`, where `kind` is one of `text_input`, `text_cached`, `audio_input`, `audio_cached`, `image_input`, `image_cached`, `text_output` or `audio_output`. Each row carries a `costOverride`. Cached tokens are subtracted from their modality's input count, so nothing is billed twice.
 
+Each `conversation.item.input_audio_transcription.completed` records `openai:<transcription model>:realtime:transcription_<kind>` rows (`text_input`, `audio_input`, `text_output`, or `seconds` for duration-billed models). The transcription model is read from the session's `session.created`/`session.updated`. Token prices: `gpt-4o-transcribe` $2.50 / $6 / $10, `gpt-4o-mini-transcribe` $1.25 / $3 / $5 (text in / audio in / text out per 1M); `whisper-1` $0.006/min. A model not in these tables is billed as `unknown` at the highest known rate.
+
 ## Configuration
 
 ```jsonc
@@ -221,6 +223,8 @@ Each `response.done` records usage rows `openai:<model>:realtime:<kind>`, where
 | Client missing a ping                         | 30–60 s | socket terminated, upstream closed         |
 | Upstream handshake                            | 10 s   | close `1011`                                |
 | Credit needed to open a session               | 1 ¢    | `insufficient_funds` error event, `1008`    |
+| Credential re-check                           | every 60 s (kv.js cache, swept every 15 s) | `invalid_api_key` error event, `1008`; a failed lookup is retried, not fatal |
+| New sessions while the node drains            | refused | `server_restarting` error event, `1012`     |
 | Session length                                | none here | OpenAI ends sessions at 60 min           |
 
 The per-user cap is published in `src/docs/src/rate-limits-and-quotas.md`; keep the two in step.
@@ -234,15 +238,15 @@ The per-user cap is published in `src/docs/src/rate-limits-and-quotas.md`; keep
 
 **The model is checked before the credential.** An unauthenticated caller can learn which model names are valid. The keyless puter.js API test relies on this order (`ai.suite.ts`, "assistant.connect surfaces relay errors…").
 
-**Metering depends on OpenAI's field order.** To avoid parsing every audio delta, the relay only parses frames whose first 64 bytes contain `"response.done"`. OpenAI currently serialises `type` first. If that changes, usage silently stops being recorded. Check for that before trusting billing numbers.
+**Metering depends on OpenAI's field order.** To avoid parsing every audio delta, the relay only reads the `type` from each frame's first 96 bytes. OpenAI currently serialises `type` first. If that changes, usage silently stops being recorded. Check for that before trusting billing numbers.
 
-**Billing happens after the fact.** No credit is held while a response runs. A response is charged when it finishes, and only then does the relay check for a positive balance, closing the session (`insufficient_funds`) if there is none. An account can therefore go negative by one response's worth. Responses cancelled before `response.done`, and anything OpenAI bills outside `response.done` (for example input transcription), are not metered.
+**Billing happens after the fact.** No credit is held while a response runs. A response is charged when it finishes, and only then does the relay check for a positive balance, closing the session (`insufficient_funds`) if there is none. An account can therefore go negative by one response's worth. Responses cancelled before `response.done` are not metered.
 
 **It skips the shared AI gates.** Sessions don't count against `AI_RATE_LIMIT`/`AI_CONCURRENT` (`drivers/util/aiLimits.ts`), only against their own per-user cap. They also bypass the AI cost-factor hook (`withAiCostFactor`), so `ai.cost.factor.*` listeners don't apply.
 
 **It isn't paid-only.** Every other `/puterai/openai/v1/*` route requires a subscription and a full-access token. This one accepts session and app tokens on any plan. The docs say so.
 
-**The credential is never re-checked.** Unlike `SocketService`, there is no periodic re-auth sweep. A revoked session or newly suspended account keeps an open session until it ends (60 min at most, OpenAI's cap). A team's 2FA requirement is not enforced either (`teamRequires2fa` isn't passed).
+**The credential is re-checked every 60 s.** A successful auth is cached in the process kv.js singleton under a hash of the token for 60 s; a sweep every 15 s re-authenticates sessions whose entry expired (one lookup per token, shared by its sessions). A revoked token or suspended account is closed within about 75 s. A lookup that throws keeps the session and retries next sweep. A team's 2FA requirement is still not enforced (`teamRequires2fa` isn't passed).
 
 **Tokens can end up in logs.** The subprotocol header and `?auth_token=` query both carry the Puter token. Proxies or access logs that record them will capture it, so prefer the `Authorization` header from servers.
 
@@ -256,7 +260,7 @@ The per-user cap is published in `src/docs/src/rate-limits-and-quotas.md`; keep
 
 **It lives in the service layer.** Only services receive the raw `http.Server` (`attachHttpServer`), so this edge component, which applies auth gates, sits in `services/` instead of `controllers/`. Move it if the relay outlives the experiment.
 
-**Shutdown closes sessions immediately.** `onServerPrepareShutdown` closes every session with `1001`. They are not drained, so a deploy drops live conversations.
+**Shutdown drains.** On `SIGTERM` (`onServerDrainStart`) new sessions are refused with `1012` and open ones keep running for the drain window (90 s with a `serverId`). At `onServerPrepareShutdown` each session gets up to 5 s to finish an in-flight response, then closes with `1012` (Service Restart) so the client reconnects elsewhere. Conversation state lives upstream and is lost on reconnect.
 
 ## Observing it
 

src/backend/services/types.ts

@@ -62,7 +62,10 @@ export const PuterService = class PuterService implements WithLifecycle {
     public onServerStart() {
         return;
     }
-    public onServerPrepareShutdown() {
+    public onServerDrainStart(): Promise<void> | void {
+        return;
+    }
+    public onServerPrepareShutdown(): Promise<void> | void {
         return;
     }
     public onServerShutdown() {

src/backend/types.ts

@@ -1308,6 +1308,12 @@ export interface WithLifecycle extends Object {
     onServerStart?: () => Promise<void> | void;
     onServerShutdown?: () => Promise<void> | void;
     onServerPrepareShutdown?: () => Promise<void> | void;
+    /**
+     * When the drain starts, before `onServerPrepareShutdown`, which waits for
+     * every connection (upgraded sockets included) to close or the drain to
+     * end.
+     */
+    onServerDrainStart?: () => Promise<void> | void;
 }
 
 export interface WithCostsReporting extends WithLifecycle {