feat(voice): voice core — honest layer reach, spoken summary with on-screen card, narration, point-and-ask

- Voice layer manifest (src/voice/layerManifest.js) generates the layer enums,
  aliases and analyst field hints; a test fails when a catalog layer has no entry.
- analyst_query validates fields, operators, units, values and scopes, counts up
  to 250,000 loaded records per layer, reports complete/partial/unanswered, tells
  LAYER_OFF / NOT_READY / FEED_UNAVAILABLE apart from zero, refuses mismatched
  follow-ups and never lets a superseded query replace follow-up memory. Seven
  more layers expose analyst records.
- Tool results carry a code-built say line and a display card; the voice card in
  the mic control shows captions, plan steps and the result
  (src/voice/resultDisplay.js is the one adapter). Slow tools narrate their
  current step out of band; holding Space while the assistant talks cuts it off.
  Turn spans go to the debug log; caption transcription is metered; the log
  omits what was said unless GEV_VOICE_LOG_CONTENT=1.
- Point and ask: the Space-down pointer snapshot resolves "this", "that one" and
  "around here" through 'pointer' targets, and "the second one" through
  referent:n from the last numbered list.
- Prompt states each policy once (instructions + tools 48.5k -> 46.7k chars).
- Voice benchmark harness (scripts/qa-voice-bench.mjs, scripts/voice-bench/)
  shares its phrase table and grading with scripts/qa-voice-routing.mjs.
This commit is contained in:
Bilawal Sidhu committed 2026-09-28 21:19:23 -05:00
1 parent e7707d9a0f
commit a19897ad15
97 files changed
+14343 -755

No files matched your search

+179
View File
@@ -0,0 +1,179 @@
/**
* Gemini Live adapter — BidiGenerateContent over WebSocket.
*
* Same prompt as production (system instruction = realtimeInstructions(),
* tools = GEV_REALTIME_TOOLS as functionDeclarations with the production JSON
* schema passed verbatim via parametersJsonSchema). The native-audio Live
* model rejects a TEXT response modality (close 1007), so the session
* requests AUDIO with output transcription on; tool calls arrive as
* `toolCall` messages and the transcript stands in for the final words.
*
* Fresh session per phrase / dialogue. Each toolCall gets stub
* `functionResponses`; the model continues on its own. A turn ends at a
* `turnComplete` whose generation had no toolCall, or after 5 tool rounds
* (mirrors the Realtime adapter's continuation cap), or 12 s of silence after
* a tool round.
*/
import WebSocket from 'ws';
import { providerKey } from '../keys.mjs';
import { realtimeInstructions, GEV_REALTIME_TOOLS } from '../productionConfig.mjs';
import { toGeminiDeclarations } from '../toolFormats.mjs';
import { geminiUsageCostUsd } from '../pricing.mjs';
import { newTurnResult } from './openaiRealtime.mjs';
const ENDPOINT = 'wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent';
const MAX_TOOL_ROUNDS = 5;
const IDLE_AFTER_TOOL_MS = 12_000;
export function createGeminiLiveAdapter({ model, schemaMode = 'jsonSchema', turnTimeoutMs = 60_000 }) {
const instructions = realtimeInstructions();
const functionDeclarations = toGeminiDeclarations(GEV_REALTIME_TOOLS, { mode: schemaMode });
const adapter = {
provider: 'gemini-live',
model,
settings: {
responseModalities: ['AUDIO'],
outputAudioTranscription: true,
schemaMode,
temperature: 'server default (not set; production has no Gemini config)',
thinking: 'server default',
input: 'realtimeInput.text',
context: 'fresh session per phrase / dialogue',
},
async open() {},
async close() {},
async runTurn({ text, onCall, context }) {
return (await adapter.runDialogue({ turns: [text], onCall, contexts: [context] }))[0];
},
runDialogue({ turns, onCall, contexts = [] }) {
return new Promise((resolveAll) => {
const tSetup = Date.now();
const ws = new WebSocket(`${ENDPOINT}?key=${providerKey('gemini')}`);
const results = [];
let setupMs = null;
let cur = null;
let closed = false;
const send = (o) => { if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify(o)); };
const shutdown = (error) => {
if (closed) return;
closed = true;
if (cur) {
clearTimeout(cur.timer);
clearTimeout(cur.idle);
if (error) cur.r.errors.push(error);
if (cur.r.tDoneMs === null) cur.r.tDoneMs = Date.now() - cur.t0;
results.push(cur.r);
cur = null;
}
while (results.length < turns.length) {
const r = newTurnResult();
r.errors.push(error || 'dialogue aborted');
results.push(r);
}
try { ws.terminate(); } catch { /* noop */ }
resolveAll(results);
};
const startTurn = (turn) => {
const r = newTurnResult();
r.setupMs = turn === 0 ? setupMs : 0;
cur = { r, t0: Date.now(), turn, toolRounds: 0, toolCallThisGeneration: false, idle: null };
cur.timer = setTimeout(() => { cur.r.timeout = true; endTurn(); }, turnTimeoutMs);
// Live has no system-role mid-session item; a context item (pointer_context)
// rides as a JSON line ahead of the user's words in the same text turn.
const ctx = contexts[turn] ? `${JSON.stringify(contexts[turn])}\n` : '';
send({ realtimeInput: { text: `${ctx}${turns[turn]}` } });
};
const endTurn = () => {
if (!cur) return;
clearTimeout(cur.timer);
clearTimeout(cur.idle);
if (cur.r.tDoneMs === null) cur.r.tDoneMs = Date.now() - cur.t0;
results.push(cur.r);
const next = cur.turn + 1;
const timedOut = cur.r.timeout;
cur = null;
if (next < turns.length && !timedOut) startTurn(next);
else shutdown();
};
const mark = (key) => { if (cur && cur.r[key] === null) cur.r[key] = Date.now() - cur.t0; };
ws.on('open', () => {
send({
setup: {
model: `models/${model}`,
generationConfig: { responseModalities: ['AUDIO'] },
systemInstruction: { parts: [{ text: instructions }] },
tools: [{ functionDeclarations }],
outputAudioTranscription: {},
},
});
});
ws.on('unexpected-response', (_req, res) => shutdown(`HTTP ${res.statusCode}`));
ws.on('error', (e) => shutdown(`ws: ${e.message}`));
ws.on('close', (code, reason) => {
shutdown(results.length < turns.length ? `close ${code}: ${reason.toString().slice(0, 200)}` : undefined);
});
ws.on('message', (raw) => {
let m;
try { m = JSON.parse(raw.toString()); } catch { return; }
if (m.setupComplete) {
if (setupMs === null) {
setupMs = Date.now() - tSetup;
startTurn(0);
}
return;
}
if (!cur) return;
clearTimeout(cur.idle);
const r = cur.r;
const lastAt = Date.now();
const sc = m.serverContent;
if (sc?.modelTurn?.parts?.length) {
mark('tFirstTokenMs');
for (const p of sc.modelTurn.parts) if (typeof p.text === 'string' && !p.thought) r.text += p.text;
}
if (sc?.outputTranscription?.text) {
mark('tFirstTokenMs');
r.text += sc.outputTranscription.text;
}
if (m.toolCall?.functionCalls?.length) {
mark('tFirstTokenMs');
mark('tFirstCallMs');
cur.toolCallThisGeneration = true;
cur.toolRounds += 1;
const functionResponses = m.toolCall.functionCalls.map((fc) => {
const c = { name: fc.name, args: fc.args || {}, atMs: Date.now() - cur.t0 };
r.calls.push(c);
return { id: fc.id, name: fc.name, response: onCall(c, cur.turn) };
});
send({ toolResponse: { functionResponses } });
}
if (m.usageMetadata) {
const u = m.usageMetadata;
r.usage.inputTokens += u.promptTokenCount || 0;
r.usage.cachedTokens += u.cachedContentTokenCount || 0;
r.usage.outputTokens += u.responseTokenCount || 0;
r.usage.reasoningTokens += u.thoughtsTokenCount || 0;
r.costUsd += geminiUsageCostUsd(model, u);
}
if (m.goAway) r.errors.push('goAway');
if (sc?.turnComplete) {
r.responses += 1;
if (cur.toolCallThisGeneration && cur.toolRounds < MAX_TOOL_ROUNDS) {
cur.toolCallThisGeneration = false;
if (r.text && !r.text.endsWith(' | ')) r.text += ' | ';
// The follow-up generation normally starts within a second; if
// the model has nothing more to say, end the turn at its last
// message rather than charging the wait to its latency.
const c = cur;
c.idle = setTimeout(() => { if (cur === c) { c.r.tDoneMs = lastAt - c.t0; endTurn(); } }, IDLE_AFTER_TOOL_MS);
return;
}
endTurn();
}
});
});
},
};
return adapter;
}
+263
View File
@@ -0,0 +1,263 @@
/**
* Local-model adapters: stateless chat requests, so every phrase is a fresh
* context by construction. Same system prompt (realtimeInstructions()) and
* the production tools converted with toChatTools() (schema-normalized).
*
* openai-chat any OpenAI-compatible server (LM Studio :1234/v1, Ollama
* :11434/v1). Streams SSE; timing is client-side.
* ollama Ollama's native /api/chat. Same message and tool formats as
* Chat Completions, but it can set num_ctx (the OpenAI shim
* cannot — the ~15k-token prompt would be silently truncated
* at a small default context) and turn thinking off, and it
* reports server-side prefill/decode/load timings.
*
* Both mirror the Realtime loop: stub tool results, then continue until a
* response has no tool calls, up to 4 continuations.
*/
import { realtimeInstructions, GEV_REALTIME_TOOLS } from '../productionConfig.mjs';
import { toChatTools } from '../toolFormats.mjs';
const MAX_CONTINUATIONS = 4;
function emptyResult() {
return {
calls: [], text: '', errors: [], responses: 0,
usage: { inputTokens: 0, cachedTokens: 0, outputTokens: 0, reasoningTokens: 0 },
costUsd: 0, tFirstTokenMs: null, tFirstCallMs: null, tDoneMs: null, setupMs: 0, timeout: false,
local: { prefillTokS: null, decodeTokS: null, promptEvalCount: null, evalCount: null, loadMs: null, firstResponseMs: null },
};
}
function parseArgs(raw) {
if (raw && typeof raw === 'object') return raw;
try { return JSON.parse(raw || '{}'); } catch { return { __unparsed: String(raw).slice(0, 400) }; }
}
async function* lines(body) {
const decoder = new TextDecoder();
let buf = '';
for await (const chunk of body) {
buf += decoder.decode(chunk, { stream: true });
let i;
while ((i = buf.indexOf('\n')) >= 0) {
yield buf.slice(0, i);
buf = buf.slice(i + 1);
}
}
if (buf.trim()) yield buf;
}
/**
* Shared multi-response loop over one or more user turns in ONE growing
* message history. `request(messages, r, t0, first, deadline)` performs one
* streamed completion and returns `{ content, toolCalls:[{id,name,args}] }`.
*/
async function runChatDialogue({ turns, onCall, request, turnTimeoutMs, contexts = [] }) {
const messages = [{ role: 'system', content: realtimeInstructions() }];
const results = [];
for (let turn = 0; turn < turns.length; turn += 1) {
const r = emptyResult();
if (contexts[turn]) messages.push({ role: 'system', content: JSON.stringify(contexts[turn]) });
messages.push({ role: 'user', content: turns[turn] });
const t0 = Date.now();
const deadline = t0 + turnTimeoutMs;
try {
for (let round = 0; round <= MAX_CONTINUATIONS; round += 1) {
if (Date.now() > deadline) { r.timeout = true; break; }
const out = await request(messages, r, t0, round === 0, deadline);
r.responses += 1;
if (out.content) r.text += (r.text ? ' | ' : '') + out.content.trim();
if (!out.toolCalls.length) {
messages.push({ role: 'assistant', content: out.content || '' });
break;
}
messages.push({ role: 'assistant', content: out.content || '', tool_calls: out.toolCalls.map((tc) => tc.wire) });
for (const tc of out.toolCalls) {
const c = { name: tc.name, args: tc.args, atMs: tc.atMs };
r.calls.push(c);
messages.push({ role: 'tool', tool_call_id: tc.id, tool_name: tc.name, name: tc.name, content: JSON.stringify(onCall(c, turn)) });
}
}
} catch (e) {
if (e?.name === 'TimeoutError' || e?.name === 'AbortError') r.timeout = true;
else r.errors.push(String(e?.message || e).slice(0, 300));
}
r.tDoneMs = Date.now() - t0;
results.push(r);
if (r.timeout || r.errors.length) {
while (results.length < turns.length) {
const skipped = emptyResult();
skipped.errors.push('dialogue aborted after an earlier turn failed');
results.push(skipped);
}
break;
}
}
return results;
}
// ── OpenAI-compatible (SSE) ─────────────────────────────────────────────────
export function createOpenAiChatAdapter({ model, baseUrl, temperature = 0.8, extraBody = {}, turnTimeoutMs = 180_000, apiKey = 'local' }) {
const tools = toChatTools(GEV_REALTIME_TOOLS);
const request = async (messages, r, t0, first, deadline) => {
const res = await fetch(`${baseUrl.replace(/\/$/, '')}/chat/completions`, {
method: 'POST',
signal: AbortSignal.timeout(Math.max(1000, deadline - Date.now())),
headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${apiKey}` },
body: JSON.stringify({
model, messages, tools, tool_choice: 'auto', temperature, stream: true,
stream_options: { include_usage: true }, ...extraBody,
}),
});
if (!res.ok) throw new Error(`HTTP ${res.status}: ${(await res.text()).slice(0, 300)}`);
const tStart = Date.now();
let tFirst = null;
let content = '';
const byIndex = new Map();
let usage = null;
let firstCallDoneAt = null;
for await (const line of lines(res.body)) {
if (!line.startsWith('data:')) continue;
const data = line.slice(5).trim();
if (data === '[DONE]') break;
let m;
try { m = JSON.parse(data); } catch { continue; }
if (m.usage) usage = m.usage;
const choice = m.choices?.[0];
const d = choice?.delta || {};
const now = Date.now();
if ((d.content || d.reasoning_content || d.reasoning || d.tool_calls?.length) && tFirst === null) {
tFirst = now;
if (r.tFirstTokenMs === null) r.tFirstTokenMs = now - t0;
}
if (d.content) content += d.content;
for (const tc of d.tool_calls || []) {
const idx = tc.index ?? byIndex.size;
if (idx > 0 && byIndex.has(0) && firstCallDoneAt === null) firstCallDoneAt = now;
const cur = byIndex.get(idx) || { id: tc.id || `call_${idx}`, name: '', argText: '' };
if (tc.id) cur.id = tc.id;
if (tc.function?.name) cur.name += tc.function.name;
if (tc.function?.arguments) cur.argText += typeof tc.function.arguments === 'string' ? tc.function.arguments : JSON.stringify(tc.function.arguments);
byIndex.set(idx, cur);
}
if (choice?.finish_reason && byIndex.size && firstCallDoneAt === null) firstCallDoneAt = now;
}
const tEnd = Date.now();
if (byIndex.size && firstCallDoneAt === null) firstCallDoneAt = tEnd;
if (byIndex.size && r.tFirstCallMs === null) r.tFirstCallMs = firstCallDoneAt - t0;
if (usage) {
r.usage.inputTokens += usage.prompt_tokens || 0;
r.usage.cachedTokens += usage.prompt_tokens_details?.cached_tokens || 0;
r.usage.outputTokens += usage.completion_tokens || 0;
r.usage.reasoningTokens += usage.completion_tokens_details?.reasoning_tokens || 0;
}
if (first) {
r.local.firstResponseMs = tEnd - tStart;
if (usage && tFirst) {
r.local.promptEvalCount = usage.prompt_tokens ?? null;
r.local.evalCount = usage.completion_tokens ?? null;
// Client-side approximations: prefill ≈ prompt / (request → first token);
// decode ≈ (completion − 1) / (first token → end of stream).
r.local.prefillTokS = usage.prompt_tokens ? usage.prompt_tokens / Math.max(0.001, (tFirst - tStart) / 1000) : null;
r.local.decodeTokS = usage.completion_tokens > 1 && tEnd > tFirst ? (usage.completion_tokens - 1) / ((tEnd - tFirst) / 1000) : null;
r.local.timingSource = 'client';
}
}
const toolCalls = [...byIndex.values()].map((tc) => ({
id: tc.id, name: tc.name, args: parseArgs(tc.argText), atMs: firstCallDoneAt - t0,
wire: { id: tc.id, type: 'function', function: { name: tc.name, arguments: tc.argText || '{}' } },
}));
return { content, toolCalls };
};
return {
provider: 'openai-chat',
model,
settings: { baseUrl, temperature, tool_choice: 'auto', toolSchema: 'normalized (untyped → string|number|boolean)', context: 'stateless request per phrase', ...extraBody },
async open() {},
async close() {},
runDialogue: ({ turns, onCall, contexts }) => runChatDialogue({ turns, onCall, contexts, request, turnTimeoutMs }),
runTurn: async ({ text, onCall, context }) => (await runChatDialogue({ turns: [text], onCall, contexts: [context], request, turnTimeoutMs }))[0],
};
}
// ── Ollama native (/api/chat, NDJSON) ───────────────────────────────────────
export function createOllamaAdapter({ model, baseUrl = 'http://localhost:11434', temperature = 0.8, numCtx = 32768, think = false, keepAlive = '30m', turnTimeoutMs = 300_000 }) {
const tools = toChatTools(GEV_REALTIME_TOOLS);
let thinkParam = think;
const request = async (messages, r, t0, first, deadline) => {
const wireMessages = messages.map((m) => {
if (m.role === 'assistant' && m.tool_calls) {
return { role: 'assistant', content: m.content, tool_calls: m.tool_calls.map((tc) => ({ function: { name: tc.function.name, arguments: parseArgs(tc.function.arguments) } })) };
}
if (m.role === 'tool') return { role: 'tool', tool_name: m.tool_name, content: m.content };
return m;
});
const body = { model, messages: wireMessages, tools, stream: true, keep_alive: keepAlive, options: { temperature, num_ctx: numCtx } };
if (thinkParam !== null) body.think = thinkParam;
const res = await fetch(`${baseUrl}/api/chat`, {
method: 'POST',
signal: AbortSignal.timeout(Math.max(1000, deadline - Date.now())),
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(body),
});
if (!res.ok) {
const errText = await res.text();
if (/does not support thinking/i.test(errText) && thinkParam !== null) {
thinkParam = null;
return request(messages, r, t0, first, deadline);
}
throw new Error(`HTTP ${res.status}: ${errText.slice(0, 300)}`);
}
let content = '';
const toolCalls = [];
let final = null;
for await (const line of lines(res.body)) {
if (!line.trim()) continue;
let m;
try { m = JSON.parse(line); } catch { continue; }
if (m.error) throw new Error(m.error);
const now = Date.now();
const msg = m.message || {};
if ((msg.content || msg.thinking || msg.tool_calls?.length) && r.tFirstTokenMs === null) r.tFirstTokenMs = now - t0;
if (msg.content) content += msg.content;
for (const tc of msg.tool_calls || []) {
if (r.tFirstCallMs === null) r.tFirstCallMs = now - t0;
const i = `${messages.length}_${toolCalls.length}`;
const args = parseArgs(tc.function?.arguments);
toolCalls.push({
id: `call_${i}`, name: tc.function?.name, args, atMs: now - t0,
wire: { id: `call_${i}`, type: 'function', function: { name: tc.function?.name, arguments: JSON.stringify(args) } },
});
}
if (m.done) final = m;
}
if (final) {
r.usage.inputTokens += final.prompt_eval_count || 0;
r.usage.outputTokens += final.eval_count || 0;
if (first) {
const ns = 1e9;
r.local.promptEvalCount = final.prompt_eval_count ?? null;
r.local.evalCount = final.eval_count ?? null;
r.local.prefillTokS = final.prompt_eval_duration ? final.prompt_eval_count / (final.prompt_eval_duration / ns) : null;
r.local.decodeTokS = final.eval_duration ? final.eval_count / (final.eval_duration / ns) : null;
r.local.loadMs = final.load_duration ? final.load_duration / 1e6 : null;
r.local.firstResponseMs = final.total_duration ? final.total_duration / 1e6 : null;
r.local.timingSource = 'ollama';
}
}
return { content, toolCalls };
};
return {
provider: 'ollama',
model,
get settings() {
return { baseUrl, api: '/api/chat', temperature, num_ctx: numCtx, think: thinkParam === null ? 'unsupported by model (param omitted)' : thinkParam, keep_alive: keepAlive, toolSchema: 'normalized (untyped → string|number|boolean)', context: 'stateless request per phrase' };
},
async open() {},
async close() {},
runDialogue: ({ turns, onCall, contexts }) => runChatDialogue({ turns, onCall, contexts, request, turnTimeoutMs }),
runTurn: async ({ text, onCall, context }) => (await runChatDialogue({ turns: [text], onCall, contexts: [context], request, turnTimeoutMs }))[0],
};
}
@@ -0,0 +1,167 @@
/**
* OpenAI Realtime adapter — text turns over the Realtime WebSocket.
*
* Each dialogue (one phrase, or a multi-turn dialogue) opens a FRESH session
* configured with production's exact session body (captured from the token
* handler) and sends its user text items one turn at a time. Each turn
* mirrors qa-voice-routing's loop: every function call gets a stub output, and
* a continuation response is requested after `response.done` (never while a
* response is active), up to 4 continuations.
*/
import WebSocket from 'ws';
import { providerKey } from '../keys.mjs';
import { captureProductionSession } from '../productionConfig.mjs';
import { openAiUsageCostUsd } from '../pricing.mjs';
const MAX_CONTINUATIONS = 4;
export function newTurnResult() {
return {
calls: [], text: '', errors: [], responses: 0,
usage: { inputTokens: 0, cachedTokens: 0, outputTokens: 0, reasoningTokens: 0 },
costUsd: 0, tFirstTokenMs: null, tFirstCallMs: null, tDoneMs: null, setupMs: null, timeout: false,
};
}
export function createOpenAiRealtimeAdapter({ model, turnTimeoutMs = 45_000 }) {
let session = null;
const adapter = {
provider: 'openai-realtime',
model,
settings: {},
async open() {
session = await captureProductionSession(model);
adapter.settings = {
reasoning: session.reasoning,
truncation: session.truncation,
tool_choice: session.tool_choice,
temperature: 'not settable (GA Realtime session has no temperature field)',
output_modalities: ['text'],
context: 'fresh session per phrase / dialogue',
};
},
async close() {},
async runTurn({ text, onCall, context }) {
return (await adapter.runDialogue({ turns: [text], onCall, contexts: [context] }))[0];
},
runDialogue({ turns, onCall, contexts = [] }) {
return new Promise((resolveAll) => {
const tSetup = Date.now();
const ws = new WebSocket(`wss://api.openai.com/v1/realtime?model=${encodeURIComponent(model)}`, {
headers: { Authorization: `Bearer ${providerKey('openai')}` },
});
const results = [];
let setupMs = null;
let cur = null; // { r, t0, continuations, needContinuation, turn, timer }
let closed = false;
const send = (o) => { if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify(o)); };
const shutdown = (error) => {
if (closed) return;
closed = true;
if (cur) {
clearTimeout(cur.timer);
if (error) cur.r.errors.push(error);
cur.r.tDoneMs = Date.now() - cur.t0;
results.push(cur.r);
cur = null;
}
while (results.length < turns.length) {
const r = newTurnResult();
r.errors.push(error || 'dialogue aborted');
results.push(r);
}
try { ws.terminate(); } catch { /* noop */ }
resolveAll(results);
};
const startTurn = (turn) => {
const r = newTurnResult();
r.setupMs = turn === 0 ? setupMs : 0;
cur = { r, t0: Date.now(), continuations: 0, needContinuation: false, turn };
cur.timer = setTimeout(() => { cur.r.timeout = true; endTurn(); }, turnTimeoutMs);
// Optional inert system item first (the app's pointer_context shape).
if (contexts[turn]) send({ type: 'conversation.item.create', item: { type: 'message', role: 'system', content: [{ type: 'input_text', text: JSON.stringify(contexts[turn]) }] } });
send({ type: 'conversation.item.create', item: { type: 'message', role: 'user', content: [{ type: 'input_text', text: turns[turn] }] } });
send({ type: 'response.create', response: { output_modalities: ['text'] } });
};
const endTurn = () => {
if (!cur) return;
clearTimeout(cur.timer);
cur.r.tDoneMs = Date.now() - cur.t0;
results.push(cur.r);
const next = cur.turn + 1;
const timedOut = cur.r.timeout;
cur = null;
if (next < turns.length && !timedOut) startTurn(next);
else shutdown();
};
const mark = (key) => { if (cur && cur.r[key] === null) cur.r[key] = Date.now() - cur.t0; };
ws.on('unexpected-response', (_req, res) => shutdown(`HTTP ${res.statusCode}`));
ws.on('error', (e) => shutdown(`ws: ${e.message}`));
ws.on('close', () => shutdown(results.length < turns.length ? 'socket closed' : undefined));
ws.on('message', (raw) => {
let m;
try { m = JSON.parse(raw.toString()); } catch { return; }
if (m.type === 'session.created') {
send({ type: 'session.update', session });
return;
}
if (m.type === 'session.updated' && setupMs === null) {
setupMs = Date.now() - tSetup;
startTurn(0);
return;
}
if (m.type === 'error') {
if (!cur) { shutdown(m.error?.message || 'realtime error'); return; }
cur.r.errors.push(m.error?.message || 'realtime error');
return;
}
if (!cur) return;
const r = cur.r;
if (
m.type === 'response.output_text.delta' ||
m.type === 'response.function_call_arguments.delta' ||
m.type === 'response.output_audio_transcript.delta'
) mark('tFirstTokenMs');
if (m.type === 'response.output_text.delta' || m.type === 'response.output_audio_transcript.delta') {
r.text += m.delta || '';
}
if (m.type === 'response.output_item.done' && m.item?.type === 'function_call') {
mark('tFirstTokenMs');
mark('tFirstCallMs');
let args = {};
try { args = JSON.parse(m.item.arguments || '{}'); } catch { args = { __unparsed: m.item.arguments }; }
const c = { name: m.item.name, args, atMs: Date.now() - cur.t0 };
r.calls.push(c);
send({
type: 'conversation.item.create',
item: { type: 'function_call_output', call_id: m.item.call_id, output: JSON.stringify(onCall(c, cur.turn)) },
});
cur.needContinuation = true;
}
if (m.type === 'response.done') {
r.responses += 1;
const u = m.response?.usage;
if (u) {
r.usage.inputTokens += u.input_tokens || 0;
r.usage.cachedTokens += u.input_token_details?.cached_tokens || 0;
r.usage.outputTokens += u.output_tokens || 0;
r.usage.reasoningTokens += u.output_token_details?.reasoning_tokens || 0;
r.costUsd += openAiUsageCostUsd(model, u);
}
if (m.response?.status === 'failed') r.errors.push(JSON.stringify(m.response?.status_details).slice(0, 200));
if (cur.needContinuation && cur.continuations < MAX_CONTINUATIONS) {
cur.needContinuation = false;
cur.continuations += 1;
if (r.text) r.text += ' | ';
send({ type: 'response.create', response: { output_modalities: ['text'] } });
return;
}
endTurn();
}
});
});
},
};
return adapter;
}