1897 lines
80 KiB
JavaScript
1897 lines
80 KiB
JavaScript
// @ts-check
|
|
/**
|
|
* Minimal WebSocket client for the Hugging Face speech-to-speech load balancer.
|
|
*
|
|
* Two-step handshake (same /session route as the WebRTC client):
|
|
*
|
|
* 1. POST `<lb_url>/session` -> JSON `{ connect_url: wss://<compute>/v1/realtime?session_token=<JWT>, ... }`
|
|
* 2. Open a WebSocket directly on `connect_url` (no rewrite, unlike the WebRTC client).
|
|
*
|
|
* Once the socket is open we follow the OpenAI Realtime GA WebSocket
|
|
* protocol:
|
|
*
|
|
* - Server pushes `session.created` immediately after upgrade.
|
|
* - We send `session.update` (GA schema: `session.audio.{input,output}`,
|
|
* `session.output_modalities`, ...).
|
|
* - We stream mic audio as PCM16 16 kHz mono base64 chunks via
|
|
* `input_audio_buffer.append`.
|
|
* - The server pushes `response.output_audio.delta` (PCM16 24 kHz mono
|
|
* base64) and transcript deltas.
|
|
*
|
|
* Audio is handled internally via two AudioWorklet processors so the
|
|
* client owns the full mic-in / speaker-out pipeline. The main app only
|
|
* sees high-level lifecycle events (`status`, `transcript`, `error`,
|
|
* `session`), the same shape as the WebRTC client.
|
|
*
|
|
* @typedef {"idle" | "creating-session" | "queued" | "your-turn" | "connecting" |
|
|
* "connected" | "user-speaking" | "processing" | "ai-speaking" |
|
|
* "closed" | "error"
|
|
* } WsStatus
|
|
*
|
|
* @typedef {Object} WsSessionInfo
|
|
* @property {string} sessionId
|
|
* @property {string} connectUrl
|
|
* @property {string} websocketUrl
|
|
* @property {string} sessionToken
|
|
* @property {number} pendingTimeoutS
|
|
* @property {string} [tier] Login tier from the session proxy ("anon"|"free"|"pro").
|
|
* @property {boolean} [limited] Whether this session is metered (heartbeat needed).
|
|
* @property {number} [heartbeatSec] Suggested heartbeat cadence in seconds.
|
|
* @property {number} [remainingSec] Daily budget left after this grant (display).
|
|
*/
|
|
|
|
/**
|
|
* @param {string} hypothesisId
|
|
* @param {string} location
|
|
* @param {string} msg
|
|
* @param {Record<string, unknown>} [data]
|
|
*/
|
|
function reportWsDebug(hypothesisId, location, msg, data = {}) {
|
|
void hypothesisId;
|
|
void location;
|
|
void msg;
|
|
void data;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 🔎 WS DEBUG TRACER (always active, 2026-09-24)
|
|
// Shared banner and label styles for all [S2S-WS:*] logs. Lifecycle events
|
|
// (connect, close, error) and the initial handshake are always logged.
|
|
// Count bulk input_audio_buffer.append messages to avoid console spam.
|
|
// ---------------------------------------------------------------------------
|
|
(function _wsDebugBanner() {
|
|
console.info(
|
|
"%c🟡 [S2S-WS] client loaded (ws/s2s-ws-client.js v2026-09-24 23:12) — logging handshake, status, key messages and tool calls; PREFETCH APIPILOT tools AT EVERY SESSION START + after phone login (per-user cache, phone changes trigger refresh=true; global invalidation __s2sInvalidateApipilotCache(); token masking; audio log throttling).",
|
|
"background:#fef3c7;color:#92400e;font-weight:700;padding:4px 8px;border-radius:4px;"
|
|
);
|
|
})();
|
|
/** @param {number} status 0=connect,1=ok,2=warn,3=err @param {string} tag @param {string} msg @param {unknown[]} extra */
|
|
function _wsLog(status, tag, msg, ...extra) {
|
|
const palette = [
|
|
["color:#1d4ed8;font-weight:700;", "background:#dbeafe;color:#1e3a8a;font-weight:800;padding:1px 6px;border-radius:3px;"], // CONNECT info
|
|
["color:#047857;font-weight:700;", "background:#d1fae5;color:#065f46;font-weight:800;padding:1px 6px;border-radius:3px;"], // OK
|
|
["color:#b45309;font-weight:700;", "background:#fef3c7;color:#92400e;font-weight:800;padding:1px 6px;border-radius:3px;"], // WARN
|
|
["color:#b91c1c;font-weight:700;", "background:#fee2e2;color:#7f1d1d;font-weight:800;padding:1px 6px;border-radius:3px;"], // ERR
|
|
];
|
|
const [labelFg, badgeBg] = palette[Math.max(0, Math.min(3, status))];
|
|
const badge = `%c[S2S-WS:${tag}]`;
|
|
console.log(`${badge}%c ${msg}`, badgeBg, labelFg, ...extra);
|
|
}
|
|
/** Bulk audio chunk counter for throttling logs. */
|
|
let _wsMicChunkCount = 0;
|
|
let _wsMicChunkLastLogTs = 0;
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 🔧 APIPILOT PREFETCH AT SESSION START (2026-09-24)
|
|
// Each connect() warms the server-side ApiPilot cache via /api/apipilot/tools.
|
|
// Client cache TTL is 240 s (server TTL 300 s minus a 60 s margin).
|
|
// Prefetch failures, including proxy 5xx responses, do not fail connect().
|
|
// ---------------------------------------------------------------------------
|
|
/** @type {{at:number, data:any, key:string|null}|null} */
|
|
let _apipilotClientCache = null;
|
|
const _APIPILOT_CLIENT_TTL_MS = 240_000; // 4 minutes
|
|
/** Session counter for logging, incremented by each connect(). */
|
|
let _wsSessionCounter = 0;
|
|
/** Per-user cache key: last 8 phoneNumber digits or a token fingerprint. */
|
|
function _apipilotUserKey() {
|
|
try {
|
|
// @ts-ignore
|
|
const phone = (typeof window !== "undefined" && window.voceroUserPhoneNumber) || "";
|
|
if (phone && phone.length >= 8) return `P:${phone.slice(-8)}`;
|
|
// @ts-ignore
|
|
const tk = typeof window !== "undefined" && typeof window.__s2sDemoSessionToken === "string" ? window.__s2sDemoSessionToken : "";
|
|
if (tk && tk.length >= 6) return `T:${tk.slice(-6)}`;
|
|
return "anon";
|
|
} catch (_) { return "anon"; }
|
|
}
|
|
/** Invalidate the client ApiPilot cache. Exported globally for main.js after phone login. */
|
|
function _invalidateApipilotClientCache() {
|
|
_apipilotClientCache = null;
|
|
console.info(
|
|
"%c🔧 [S2S-WS:APIPILOT] Client cache invalidated (new user login or forced refresh=true)",
|
|
"background:#fde68a;color:#92400e;font-weight:700;padding:1px 6px;border-radius:3px;"
|
|
);
|
|
}
|
|
|
|
// Match the Python server's marker and rules for consistent deduplication and
|
|
// AI marker syntax. effectiveReasoningInstructions checks this marker so the
|
|
// client and server do not append the same tools list twice.
|
|
const APIPILOT_PASS1_LIST_MARKER_JS = "<<<APIPILOT_TOOLS_LIST_BEGIN>>>";
|
|
|
|
/**
|
|
* Build the PASS-1 block injected into the LLM instructions prompt.
|
|
* Match Python _resolve_apipilot_reasoning_context_block(tools=[...]).
|
|
* - Start with APIPILOT_PASS1_LIST_MARKER_JS
|
|
* - End with <<<APIPILOT_TOOLS_LIST_END>>>
|
|
* - Instructions in English; assistant responses remain Spanish
|
|
* - Cap descriptions at 500 characters and tools at 100
|
|
* @param {any[]|null|undefined} tools Cached list, or null to use the current cache
|
|
* @returns {string} PASS-1 text block, empty when no tools or cache are available
|
|
*/
|
|
function buildApipilotPass1PromptBlock(tools) {
|
|
let list = Array.isArray(tools) ? tools : null;
|
|
if (!list) {
|
|
// Fall back to the current client cache, if available.
|
|
const c = _apipilotClientCache;
|
|
if (c && c.data && Array.isArray(c.data.tools)) list = c.data.tools;
|
|
}
|
|
if (!Array.isArray(list) || list.length === 0) return "";
|
|
const capped = list.slice(0, 100);
|
|
/** @type {string[]} */
|
|
const entries = [];
|
|
for (const t of capped) {
|
|
if (!t || typeof t !== "object") continue;
|
|
const name = typeof t.name === "string" ? t.name : "";
|
|
if (!name) continue;
|
|
const desc = typeof t.description === "string" ? t.description : "";
|
|
const params = (t.parameters && typeof t.parameters === "object") ? t.parameters : null;
|
|
const requiresAuth = Boolean(t.requires_auth);
|
|
const req = (params && Array.isArray(params.required))
|
|
? params.required.filter((x) => typeof x === "string")
|
|
: [];
|
|
let descCut = desc.slice(0, 500);
|
|
if (desc.length > 500) descCut += "…";
|
|
let snippet = `- name: ${name}\n requires_auth: ${requiresAuth ? "true" : "false"}`;
|
|
if (req.length) snippet += `\n required_parameters: [${req.slice(0, 40).join(", ")}]`;
|
|
if (descCut) snippet += `\n description: ${descCut}`;
|
|
entries.push(snippet);
|
|
}
|
|
if (!entries.length) return "";
|
|
console.warn("APIPILOT_PASS1_LIST_MARKER_JS", APIPILOT_PASS1_LIST_MARKER_JS);
|
|
const markers = (
|
|
`${APIPILOT_PASS1_LIST_MARKER_JS}\n` +
|
|
"### Available dynamic tools (ApiPilot AquaPortal)\n" +
|
|
"\n" +
|
|
"Each tool has a `name` field (a unique identifier: DO NOT change, camelCase " +
|
|
"or translate it) and a `description` field explaining its use case.\n" +
|
|
"\n" +
|
|
"MANDATORY EXECUTION RULES:\n" +
|
|
"1. If the USER REQUEST (transcript) clearly matches the `description` of " +
|
|
"EXACTLY one (1) tool below, output ONLY one line as the FIRST line of YOUR " +
|
|
"RESPONSE, using this exact syntax:\n" +
|
|
' <<<APIPILOT_CALL:NAME=tool_name,ARGUMENTS={"parameter1":"value1","parameter2":"value2"}>>>\n' +
|
|
"2. ARGUMENTS={…} MUST be valid JSON (double quotes, commas, no comments). " +
|
|
"Include ONLY fields listed in `required_parameters:`, if any.\n" +
|
|
"3. Do NOT add text after the marker line (no explanations or previews). " +
|
|
"The engine executes the tool and supplies its result in a second call.\n" +
|
|
"4. If NO `description` matches, do NOT emit a marker; respond normally in Spanish.\n" +
|
|
"5. If multiple descriptions match, choose the most specific and use the marker.\n" +
|
|
"\n" +
|
|
"Tools list (name / description):\n" +
|
|
`${entries.join("\n")}\n` +
|
|
"<<<APIPILOT_TOOLS_LIST_END>>>"
|
|
);
|
|
console.warn("markers", markers);
|
|
return markers;
|
|
}
|
|
|
|
/**
|
|
* Check whether the input instructions already contain the PASS1 marker.
|
|
* @param {string|undefined|null} text
|
|
* @returns {boolean}
|
|
*/
|
|
function instructionsAlreadyHasApipilotPass1(text) {
|
|
return typeof text === "string" && text.includes(APIPILOT_PASS1_LIST_MARKER_JS);
|
|
}
|
|
|
|
/**
|
|
* Expose a global helper for main.js to append tools to
|
|
* effectiveReasoningInstructions, and for other modules to read the list.
|
|
* The caller checks for duplicates before appending.
|
|
* @returns {{block: string, count: number, marker: string, alreadyPresent:(t:string)=>boolean}}
|
|
*/
|
|
function getApipilotPass1ContextInfo() {
|
|
const c = _apipilotClientCache;
|
|
const tools = (c && c.data && Array.isArray(c.data.tools)) ? c.data.tools : [];
|
|
const block = buildApipilotPass1PromptBlock(tools);
|
|
return {
|
|
block,
|
|
count: tools.length,
|
|
marker: APIPILOT_PASS1_LIST_MARKER_JS,
|
|
alreadyPresent: instructionsAlreadyHasApipilotPass1,
|
|
};
|
|
}
|
|
try {
|
|
if (typeof window !== "undefined") {
|
|
Object.defineProperty(window, "__s2sInvalidateApipilotCache", {
|
|
value: _invalidateApipilotClientCache,
|
|
writable: true, configurable: true,
|
|
});
|
|
Object.defineProperty(window, "__s2sGetApipilotPass1ContextInfo", {
|
|
value: getApipilotPass1ContextInfo,
|
|
writable: true, configurable: true,
|
|
});
|
|
}
|
|
} catch (_) { /* noop */ }
|
|
|
|
/**
|
|
* @typedef {Object} WsClientOptions
|
|
* @property {string} [sessionUrl] URL to POST for the session handshake (returns
|
|
* `{ connect_url, ... }`). Usually a same-origin proxy like `api/session` so the
|
|
* load-balancer address stays server-side. Provide this OR `directUrl`.
|
|
* @property {string} [loadBalancerUrl] Load-balancer base URL. Legacy/direct
|
|
* alternative to `sessionUrl`: the client POSTs `<lb>/session` itself. Prefer
|
|
* `sessionUrl` so the LB address isn't exposed to the browser.
|
|
* @property {string} [directUrl] Full WebSocket URL of an s2s realtime endpoint
|
|
* (e.g. `ws://localhost:8080/v1/realtime`). When set, the client skips the
|
|
* session POST and dials it directly — no load balancer in between.
|
|
* @property {string} voice
|
|
* @property {string} instructions
|
|
* @property {MediaStream} [micStream] Live mic stream. Provide this OR `acquireMic`.
|
|
* @property {() => Promise<MediaStream>} [acquireMic] Lazily obtain the mic stream,
|
|
* called only once a session is actually granted (after any queue wait). Lets the
|
|
* caller prime mic permission up front but not hold the mic 'in use' indicator on
|
|
* while waiting in line. Ignored if `micStream` is already set.
|
|
* @property {AudioContext} [audioContext] Pre-created (and resumed) context.
|
|
* iOS Safari only lets an AudioContext start from within a user gesture, so
|
|
* the caller creates/resumes it synchronously on the orb tap and hands it
|
|
* here; otherwise it stays suspended (silent) after the mic/session awaits.
|
|
* @property {ToolDef[]} [tools] Function tools declared to the backend in the
|
|
* initial `session.update`. The model decides when to call them; the caller
|
|
* executes and replies via `sendToolOutput` + `requestResponse`.
|
|
* @property {NoiseGate} [noiseGate] Client-side noise gate applied to the mic
|
|
* before it's sent. Tunable live via `setNoiseGate`.
|
|
*
|
|
* @typedef {Object} NoiseGate
|
|
* @property {boolean} enabled
|
|
* @property {number} thresholdDb Open threshold in dBFS (e.g. -45).
|
|
*
|
|
* @typedef {Object} ToolDef
|
|
* @property {"function"} type
|
|
* @property {string} name
|
|
* @property {string} description
|
|
* @property {object} parameters JSON Schema for the call arguments.
|
|
*
|
|
* @typedef {Object} TranscriptEvent
|
|
* @property {"user" | "assistant"} role
|
|
* @property {string} text
|
|
* @property {boolean} partial
|
|
*/
|
|
|
|
import {
|
|
base64FromArrayBuffer,
|
|
base64ToBytes,
|
|
extractResponseTranscript,
|
|
trimTrailingSlash,
|
|
} from "./codec.js?v=20260825003";
|
|
import { OrbVisualiser, VIS_FFT_SIZE } from "./orb-visualizer.js?v=20260825003";
|
|
|
|
/** Build an Error carrying a `code` (and optional extra fields) so callers can
|
|
* branch on the failure kind: "limit" | "queue-full" | "queue-expired" | "aborted".
|
|
* @param {string} message @param {string} code @param {object} [extra] */
|
|
function _codedError(message, code, extra) {
|
|
const err = /** @type {Error & { code?: string }} */ (new Error(message));
|
|
err.code = code;
|
|
if (extra) Object.assign(err, extra);
|
|
return err;
|
|
}
|
|
|
|
// The s2s pipeline runs internally at 16 kHz mono PCM. The WebRTC transport
|
|
// resamples to 48 kHz for Opus, but the WebSocket transport emits the
|
|
// native pipeline rate. We don't (can't) override it via `audio.output.format`
|
|
// because the server's pydantic validator rejects the whole `session.update`
|
|
// as soon as a sub-field shape it doesn't know about appears.
|
|
const OUTPUT_SAMPLE_RATE = 16000;
|
|
const MIC_CHUNK_MS = 40;
|
|
|
|
export class S2sWsRealtimeClient extends EventTarget {
|
|
/** @param {WsClientOptions} options */
|
|
constructor(options) {
|
|
super();
|
|
/** @type {WsClientOptions} */
|
|
this.options = options;
|
|
/** @type {ToolDef[]} Function tools declared to the backend. */
|
|
this._tools = options.tools ?? [];
|
|
/** @type {string} Direct realtime WS URL (set => skip the LB session POST). */
|
|
this._directUrl = options.directUrl ?? "";
|
|
/** @type {string} Where to POST for the session handshake. Prefer the
|
|
* explicit `sessionUrl`; fall back to `<loadBalancerUrl>/session` for callers
|
|
* that still pass the LB address directly. */
|
|
this._sessionUrl = options.sessionUrl
|
|
? options.sessionUrl
|
|
: options.loadBalancerUrl
|
|
? `${trimTrailingSlash(options.loadBalancerUrl)}/session`
|
|
: "";
|
|
/** @type {(() => Promise<MediaStream>) | null} Lazy mic acquisition (post-grant). */
|
|
this._acquireMic = options.acquireMic ?? null;
|
|
/** @type {boolean} Set by close() to abort a queue wait in progress. */
|
|
this._closed = false;
|
|
/** @type {string} The active queue ticket id while waiting (else ""). */
|
|
this._queueId = "";
|
|
/** @type {(() => void) | null} Wakes the queue poll sleep early on close(). */
|
|
this._queueWake = null;
|
|
/** @type {ReturnType<typeof setTimeout> | 0} */
|
|
this._queueTimer = 0;
|
|
// Join gate: after waiting in line the caller must explicitly `join()` before
|
|
// we dial, so a slot isn't spent on someone who walked away. Resolved by
|
|
// join(), rejected on timeout (the LB reclaims the slot) or close().
|
|
/** @type {(() => void) | null} */
|
|
this._joinResolve = null;
|
|
/** @type {((err: Error) => void) | null} */
|
|
this._joinReject = null;
|
|
/** @type {ReturnType<typeof setTimeout> | 0} */
|
|
this._joinTimer = 0;
|
|
/** @type {NoiseGate} Mic noise gate; off by default. */
|
|
this._noiseGate = options.noiseGate ?? { enabled: false, thresholdDb: -45 };
|
|
/** @type {boolean} If true, right after the initial session.updated ack we inject
|
|
* the wake-word-style bootstrap phrase (default "hola vocero") as if the user
|
|
* had spoken it, then trigger `requestResponse()` — so the assistant
|
|
* behaves EXACTLY as it does when the "Hola Vocero" hot-word is detected
|
|
* (same code path, same greeting style). Used when S2S_START_SESSION_MODE=TALK. */
|
|
this._startWithResponse = options.startWithResponse === true;
|
|
/** @type {string} Phrase injected into the first turn (mirrors the wake-word
|
|
* confirmation flow. Defaults to "hola vocero". Used ONLY on first session start.
|
|
* Overridable via options.startWithResponseText for test/CI flows. */
|
|
this._startWithResponseText = (options.startWithResponseText || "").trim() || "hola vocero";
|
|
/** @type {boolean} Guards the initial bootstrap turn from firing more than once
|
|
* (e.g. after a live `updateSession()` that re-emits session.updated later
|
|
* while the session is already running). */
|
|
this._initialResponseSent = false;
|
|
/** @type {WebSocket | null} */
|
|
this._ws = null;
|
|
/** @type {AudioContext | null} */
|
|
this._ctx = null;
|
|
/** @type {MediaStreamAudioSourceNode | null} */
|
|
this._micSrc = null;
|
|
/** @type {AudioWorkletNode | null} */
|
|
this._captureNode = null;
|
|
/** @type {AudioWorkletNode | null} */
|
|
this._playbackNode = null;
|
|
/** @type {GainNode | null} */
|
|
this._captureSink = null;
|
|
/** @type {AnalyserNode | null} */
|
|
this._micAnalyser = null;
|
|
/** @type {AnalyserNode | null} */
|
|
this._outAnalyser = null;
|
|
/** @type {OrbVisualiser | null} */
|
|
this._visualiser = null;
|
|
/** @type {WsStatus} */
|
|
this._status = "idle";
|
|
this._aiSpeaking = false;
|
|
/** @type {Set<string>} response_ids that have actually played audio, so the
|
|
* UI can tell a barge-in cut (keep it) from a never-heard speculative
|
|
* response (drop it). */
|
|
this._audibleResponses = new Set();
|
|
/** @type {Map<string, string>} The CURRENT assistant transcript segment per
|
|
* response, accumulated from streamed deltas (reset on each segment's done). */
|
|
this._asstTranscriptByResp = new Map();
|
|
/** @type {Map<string, string>} Completed assistant transcript segments per
|
|
* response, space-joined. A single response can emit several
|
|
* `*.transcript.done` events; we concatenate them until response.done. */
|
|
this._asstFullByResp = new Map();
|
|
this._muted = false;
|
|
// ── Response lock ────────────────────────────────────────────────────
|
|
// The backend allows only ONE response in flight: creating a second while
|
|
// one is active fails with `conversation_already_has_active_response`. So
|
|
// we serialize response.create. `_openResponses` counts responses the
|
|
// server has confirmed (response.created) but not yet finished
|
|
// (response.done) — it's cumulative, so every create maps to one done.
|
|
// `_createInFlight` covers the window after we send a create but before its
|
|
// response.created echo. Any requestResponse() made while locked is queued
|
|
// and replayed, one at a time, as each response.done frees the slot.
|
|
this._openResponses = 0;
|
|
this._createInFlight = false;
|
|
/** @type {{ image?: string }[]} Pending response.create payloads, one per
|
|
* queued requestResponse(). A payload may carry an image to send just
|
|
* before its create (so the frame travels with the create, not eagerly). */
|
|
this._createQueue = [];
|
|
/** @type {string[]} Pending user texts to inject only once the active
|
|
* assistant response has fully closed, preventing barge-in style audio cuts. */
|
|
this._deferredUserTextQueue = [];
|
|
/** @type {boolean} Silence the next assistant response opened after an MCP
|
|
* fallback wins the turn; covers the race before `response.created` lands. */
|
|
this._suppressNextAssistantResponse = false;
|
|
/** @type {string} Response id currently being silenced because it belongs
|
|
* to the speculative pre-tool reply. */
|
|
this._suppressedResponseId = "";
|
|
/** @type {string} Latest response id known to be active. */
|
|
this._activeResponseId = "";
|
|
/** @type {Promise<void> | null} */
|
|
this._readyPromise = null;
|
|
this._sessionConfigured = false;
|
|
this._debug = (() => { try { return localStorage.getItem("s2s.debug") === "1"; } catch { return false; } })();
|
|
}
|
|
|
|
get status() {
|
|
return this._status;
|
|
}
|
|
|
|
/** @param {WsStatus} status */
|
|
_setStatus(status) {
|
|
if (this._status === status) return;
|
|
this._status = status;
|
|
this.dispatchEvent(new CustomEvent("status", { detail: { status } }));
|
|
}
|
|
|
|
/** Full assistant transcript so far for a response: the completed segments
|
|
* plus the in-progress one, all space-joined.
|
|
* @param {string} rid @returns {string} */
|
|
_asstDisplay(rid) {
|
|
const full = this._asstFullByResp.get(rid) || "";
|
|
const seg = this._asstTranscriptByResp.get(rid) || "";
|
|
if (!seg) return full;
|
|
return full ? `${full} ${seg}` : seg;
|
|
}
|
|
|
|
_markAudible() {
|
|
if (this._status === "ai-speaking") return;
|
|
if (this._status === "closed" || this._status === "error") return;
|
|
this._setStatus("ai-speaking");
|
|
}
|
|
|
|
/**
|
|
* Full handshake. Resolves once the WS is open AND the audio pipeline is
|
|
* ready to send/receive samples.
|
|
* @returns {Promise<void>}
|
|
*/
|
|
async connect() {
|
|
if (this._ws) throw new Error("Already connected");
|
|
// Correlate [S2S-WS:*] events using an increasing session number.
|
|
_wsSessionCounter += 1;
|
|
const sessionIdShort = `#${_wsSessionCounter}`;
|
|
this._sessionSeq = _wsSessionCounter;
|
|
_wsLog(0, "SESSION", `${sessionIdShort} handshake started — status: creating-session`);
|
|
|
|
let connectUrl;
|
|
if (this._directUrl) {
|
|
// Direct mode: no load balancer, no /session POST — dial the realtime
|
|
// endpoint straight away (e.g. a local s2s server).
|
|
connectUrl = this._directUrl;
|
|
this._setStatus("connecting");
|
|
} else {
|
|
if (!this._sessionUrl) {
|
|
throw new Error("No session endpoint or direct URL configured");
|
|
}
|
|
this._setStatus("creating-session");
|
|
const { grant, waited } = await this._createSessionOrQueue();
|
|
if (this._closed) throw _codedError("connect aborted", "aborted");
|
|
// If we waited in line, don't dial until the user explicitly joins — this
|
|
// keeps a freed slot from being spent on someone who stepped away, and the
|
|
// click is a fresh gesture (re-arms the AudioContext on iOS).
|
|
if (waited) {
|
|
_wsLog(1, "QUEUE", `${sessionIdShort} your turn — waiting for a join click to confirm the slot`);
|
|
await this._awaitJoin(grant);
|
|
if (this._closed) throw _codedError("connect aborted", "aborted");
|
|
}
|
|
this.dispatchEvent(new CustomEvent("session", { detail: { info: grant } }));
|
|
connectUrl = grant.connectUrl;
|
|
this._setStatus("connecting");
|
|
}
|
|
|
|
// Acquire the mic now — only once a slot is actually ours. The caller primed
|
|
// permission up front, so this is silent and the 'in use' indicator lights
|
|
// only for a real, connecting session (never during a queue wait).
|
|
if (!this.options.micStream && this._acquireMic) {
|
|
this.options.micStream = await this._acquireMic();
|
|
}
|
|
|
|
// Spin up the AudioContext + worklets in parallel with the WS dial,
|
|
// Also warm ApiPilot (dynamic AquaPortal tools) in parallel.
|
|
// A failed prefetch does not fail connect().
|
|
const audioReady = this._setupAudio();
|
|
const wsReady = this._openWebSocket(connectUrl);
|
|
const apipilotReady = this._prefetchApipilotTools(sessionIdShort);
|
|
await Promise.all([audioReady, wsReady, apipilotReady.catch(() => void 0)]);
|
|
_wsLog(1, "SESSION", `${sessionIdShort} connection established + audio pipeline + apipilot OK`);
|
|
}
|
|
|
|
/**
|
|
* 🔧 Pre-fetch ApiPilot tools at every Realtime session start.
|
|
* - Server proxy: GET api/apipilot/tools?include_unauthorized=true
|
|
* - The global FETCH wrapper logs [FETCH:APIPILOT:TOOLS] in the console.
|
|
* - Local TTL 240 s avoids redundant calls during rapid session restarts
|
|
* (server TTL=300 s).
|
|
* - Skip immediately when window.__s2sApipilot=false.
|
|
* - Catch failures internally.
|
|
* @param {string} sessionIdShort Log tag "session #N"
|
|
* @returns {Promise<void>}
|
|
*/
|
|
async _prefetchApipilotTools(sessionIdShort) {
|
|
try {
|
|
// @ts-ignore: main.js sets __s2sApipilot after fetchConfig().
|
|
const enabled = typeof window !== "undefined" && !!window.__s2sApipilot;
|
|
if (!enabled) {
|
|
_wsLog(2, "APIPILOT", `${sessionIdShort} server feature flag apipilot=false — skipping tools prefetch`);
|
|
return;
|
|
}
|
|
const currentUser = _apipilotUserKey();
|
|
const now = Date.now();
|
|
const cached = _apipilotClientCache;
|
|
// Use the cache only within its TTL and for the same user key.
|
|
if (cached && (now - cached.at <= _APIPILOT_CLIENT_TTL_MS) && cached.key === currentUser) {
|
|
const ageSec = Math.round((now - cached.at) / 1000);
|
|
const n =
|
|
cached.data && typeof cached.data.count === "number"
|
|
? cached.data.count
|
|
: Array.isArray(cached.data && cached.data.tools)
|
|
? cached.data.tools.length
|
|
: "?";
|
|
_wsLog(1, "APIPILOT", `${sessionIdShort} HIT client tools cache (user=${currentUser}, age=${ageSec}s, N=${n}, TTL=${Math.round(_APIPILOT_CLIENT_TTL_MS/1000)}s)`);
|
|
return;
|
|
}
|
|
// Forward x-demo-session when available for the S2S server auth fingerprint.
|
|
/** @type {Record<string,string>} */
|
|
const extraHeaders = {};
|
|
try {
|
|
// @ts-ignore
|
|
if (typeof window.__s2sDemoSessionToken === "string" && window.__s2sDemoSessionToken.length > 4) {
|
|
// @ts-ignore
|
|
extraHeaders["x-demo-session"] = window.__s2sDemoSessionToken;
|
|
}
|
|
} catch (_) { /* noop */ }
|
|
// Refresh when the user key changes or the cache is missing/expired.
|
|
const needRefresh = !cached || cached.key !== currentUser || (now - cached.at > _APIPILOT_CLIENT_TTL_MS);
|
|
const qsRefresh = needRefresh ? "&refresh=true" : "";
|
|
const t0 = performance.now();
|
|
_wsLog(0, "APIPILOT", `${sessionIdShort} MISS cache (user=${currentUser}, refresh=${needRefresh ? "Y" : "N"}) → GET api/apipilot/tools (server proxy to S2S_APIPILOT_TOOLS_URL)...`);
|
|
const res = await fetch(`api/apipilot/tools?include_unauthorized=true${qsRefresh}`, {
|
|
method: "GET",
|
|
cache: "no-store",
|
|
headers: Object.keys(extraHeaders).length ? extraHeaders : undefined,
|
|
});
|
|
const dt = performance.now() - t0;
|
|
if (!res.ok) {
|
|
throw new Error(`HTTP ${res.status}`);
|
|
}
|
|
const j = await res.json().catch(() => ({}));
|
|
const ok = j && j.success && j.data;
|
|
if (!ok) {
|
|
throw new Error(`invalid envelope: success=${Boolean(j && j.success)}`);
|
|
}
|
|
const tools = Array.isArray(j.data.tools) ? j.data.tools : [];
|
|
const count = typeof j.data.count === "number" ? j.data.count : tools.length;
|
|
// Update the local cache so a subsequent session can reuse it.
|
|
_apipilotClientCache = { at: Date.now(), data: { count, tools: tools.slice(0, 100) }, key: currentUser };
|
|
const names = tools
|
|
.slice(0, 5)
|
|
.map((t) => (typeof t.name === "string" ? t.name : "?"))
|
|
.join(", ");
|
|
_wsLog(
|
|
1,
|
|
"APIPILOT",
|
|
`${sessionIdShort} READY APIPILOT tools loaded: N=${count}, t=${Math.round(dt)}ms, user=${currentUser} [first 5: ${names || "(none)"}]`
|
|
);
|
|
} catch (err) {
|
|
const msg = err instanceof Error ? err.message : String(err);
|
|
_wsLog(
|
|
3,
|
|
"APIPILOT",
|
|
`${sessionIdShort} prefetch FAILED: ${msg} — check server logs [APIPILOT-DISCOVERY]`
|
|
);
|
|
// Do not rethrow: connect() must still succeed.
|
|
}
|
|
}
|
|
|
|
/**
|
|
* POST the session handshake; if the pool is busy, wait in the queue (polling
|
|
* position) until a slot is claimed. Resolves to a grant plus whether we had to
|
|
* wait (which decides if an explicit join is required before dialing).
|
|
* @returns {Promise<{ grant: WsSessionInfo, waited: boolean }>}
|
|
*/
|
|
async _createSessionOrQueue() {
|
|
const first = await this._postSession();
|
|
if (first.state === "queued") {
|
|
this._setStatus("queued");
|
|
const grant = await this._pollQueue(first);
|
|
return { grant, waited: true };
|
|
}
|
|
return { grant: first.grant, waited: false };
|
|
}
|
|
|
|
/**
|
|
* Hold at the front of the line until the user clicks join (resolves the gate)
|
|
* or the grant lapses. Announces "your-turn" + a deadline the UI counts down.
|
|
* @param {WsSessionInfo} grant
|
|
* @returns {Promise<void>}
|
|
*/
|
|
_awaitJoin(grant) {
|
|
// The LB reclaims an unclaimed slot at its pending timeout; expire the gate a
|
|
// touch earlier so we never dial a session the LB just reaped.
|
|
const windowS = Math.max(3, (grant.pendingTimeoutS || 60) - 3);
|
|
this._setStatus("your-turn");
|
|
this.dispatchEvent(
|
|
new CustomEvent("ready-to-join", { detail: { info: grant, expiresSec: windowS } }),
|
|
);
|
|
return new Promise((resolve, reject) => {
|
|
this._joinResolve = resolve;
|
|
this._joinReject = reject;
|
|
this._joinTimer = setTimeout(() => {
|
|
this._joinResolve = null;
|
|
this._joinReject = null;
|
|
reject(_codedError("Your spot expired", "join-expired"));
|
|
}, windowS * 1000);
|
|
});
|
|
}
|
|
|
|
/** Accept the held slot and let connect() proceed to dial. Called from the
|
|
* "Join now" click, so it's a user gesture: re-resume the AudioContext, which
|
|
* iOS may have suspended while we waited. */
|
|
join() {
|
|
if (this._joinTimer) {
|
|
clearTimeout(this._joinTimer);
|
|
this._joinTimer = 0;
|
|
}
|
|
try {
|
|
void this.options.audioContext?.resume();
|
|
} catch {
|
|
// best-effort; _setupAudio resumes again
|
|
}
|
|
const resolve = this._joinResolve;
|
|
this._joinResolve = null;
|
|
this._joinReject = null;
|
|
resolve?.();
|
|
}
|
|
|
|
/**
|
|
* POST /session once. Returns either a granted session or a queue ticket.
|
|
* @returns {Promise<{ state: "granted", grant: WsSessionInfo } | { state: "queued", queueId: string, position: number, pollIntervalS: number }>}
|
|
*/
|
|
async _postSession() {
|
|
const url = this._sessionUrl;
|
|
console.log("[ws] POST", url);
|
|
_wsLog(0, "SESSION", `POST /session → ${url}`);
|
|
const t0 = performance.now();
|
|
const response = await fetch(url, {
|
|
method: "POST",
|
|
headers: { "Content-Type": "application/json" },
|
|
body: "{}",
|
|
});
|
|
const dur = (performance.now() - t0).toFixed(1);
|
|
if (response.status === 402) {
|
|
const body = await response.json().catch(() => ({}));
|
|
_wsLog(3, "SESSION", `/session HTTP 402: daily limit exhausted (tier=${body?.tier ?? "?"}) dur=${dur} ms`);
|
|
throw _codedError("Daily conversation limit reached", "limit", { tier: body?.tier });
|
|
}
|
|
if (response.status === 503) {
|
|
const body = await response.json().catch(() => ({}));
|
|
if (body?.state === "at_capacity") {
|
|
_wsLog(2, "QUEUE:FULL", `/session HTTP 503 state=at_capacity (queue full) dur=${dur} ms`);
|
|
throw _codedError("The queue is full — try again shortly.", "queue-full");
|
|
}
|
|
_wsLog(3, "SESSION", `/session HTTP 503 (unknown fallback) dur=${dur} ms`);
|
|
throw new Error("/session failed (503)");
|
|
}
|
|
if (!response.ok) {
|
|
const text = await response.text().catch(() => "");
|
|
_wsLog(3, "SESSION", `/session failed HTTP ${response.status}: ${text.slice(0, 180)} dur=${dur} ms`);
|
|
throw new Error(`/session failed (${response.status}): ${text}`);
|
|
}
|
|
const json = await response.json();
|
|
if (json.state === "queued") {
|
|
_wsLog(
|
|
2,
|
|
"QUEUE:WAIT",
|
|
`/session → state=queued queueId=${json.queue_id ?? "?"} position=${json.position ?? "?"} pollEvery=${json.poll_interval_s ?? "?"}s dur=${dur} ms`
|
|
);
|
|
return {
|
|
state: "queued",
|
|
queueId: json.queue_id,
|
|
position: json.position,
|
|
pollIntervalS: json.poll_interval_s,
|
|
};
|
|
}
|
|
const grant = this._parseGrant(json);
|
|
_wsLog(
|
|
1,
|
|
"SESSION",
|
|
`/session OK → GRANT sessionId=${grant.sessionId?.slice(0, 10)}… pendingTimeout=${grant.pendingTimeoutS}s dur=${dur} ms`
|
|
);
|
|
return { state: "granted", grant };
|
|
}
|
|
|
|
/**
|
|
* Poll the waiting queue until this ticket claims a slot. Emits `queue` events
|
|
* ({ position }) as the line advances. Throws on limit (402), expiry (404), or
|
|
* close(). Transient network/5xx blips are ignored and retried next tick.
|
|
* @param {{ queueId: string, position: number, pollIntervalS: number }} ticket
|
|
* @returns {Promise<WsSessionInfo>}
|
|
*/
|
|
async _pollQueue(ticket) {
|
|
const intervalMs = Math.max(1, ticket.pollIntervalS || 2) * 1000;
|
|
this._queueId = ticket.queueId;
|
|
this._emitQueue(ticket.position);
|
|
_wsLog(
|
|
2,
|
|
"QUEUE:POLL",
|
|
`Start queue polling queueId=${ticket.queueId.slice(0, 10)}… initialPos=${ticket.position} intervalMs=${intervalMs}`
|
|
);
|
|
let pollNum = 0;
|
|
|
|
while (true) {
|
|
pollNum += 1;
|
|
await this._queueSleep(intervalMs);
|
|
if (this._closed) throw _codedError("queue wait aborted", "aborted");
|
|
|
|
let response;
|
|
try {
|
|
response = await fetch(`api/queue/${encodeURIComponent(this._queueId)}`, {
|
|
headers: { "Content-Type": "application/json" },
|
|
});
|
|
} catch {
|
|
continue; // network blip — keep our place, retry next tick
|
|
}
|
|
|
|
if (response.status === 402) {
|
|
const body = await response.json().catch(() => ({}));
|
|
_wsLog(3, "QUEUE:LIMIT", `poll #${pollNum} HTTP 402 — daily limit (tier=${body?.tier ?? "?"})`);
|
|
throw _codedError("Daily conversation limit reached", "limit", { tier: body?.tier });
|
|
}
|
|
if (response.status === 404) {
|
|
_wsLog(3, "QUEUE:EXPIRED", `poll #${pollNum} HTTP 404 — ticket expired queueId=${this._queueId.slice(0, 10)}…`);
|
|
throw _codedError("Queue timed out", "queue-expired");
|
|
}
|
|
if (!response.ok) continue; // 502/503 — transient, retry
|
|
|
|
const json = await response.json().catch(() => null);
|
|
if (!json) continue;
|
|
if (json.state === "queued") {
|
|
if (this._debug || pollNum === 1 || pollNum % 10 === 0) {
|
|
_wsLog(2, "QUEUE:POLL", `poll #${pollNum} position=${json.position ?? "?"}`);
|
|
}
|
|
this._emitQueue(json.position);
|
|
continue;
|
|
}
|
|
// Reached the front and claimed a slot.
|
|
_wsLog(1, "QUEUE:FRONT", `poll #${pollNum} — REACHED the front. Assigning compute slot.`);
|
|
this._queueId = "";
|
|
return this._parseGrant(json);
|
|
}
|
|
}
|
|
|
|
/** @param {number} position */
|
|
_emitQueue(position) {
|
|
this.dispatchEvent(
|
|
new CustomEvent("queue", { detail: { position, queueId: this._queueId } }),
|
|
);
|
|
}
|
|
|
|
/** A sleep that close() can cut short so a queued client tears down promptly.
|
|
* @param {number} ms */
|
|
_queueSleep(ms) {
|
|
return new Promise((resolve) => {
|
|
this._queueWake = resolve;
|
|
this._queueTimer = setTimeout(() => {
|
|
this._queueWake = null;
|
|
resolve();
|
|
}, ms);
|
|
});
|
|
}
|
|
|
|
/** @param {any} json @returns {WsSessionInfo} */
|
|
_parseGrant(json) {
|
|
return {
|
|
sessionId: json.session_id,
|
|
connectUrl: json.connect_url,
|
|
websocketUrl: json.websocket_url,
|
|
sessionToken: json.session_token,
|
|
pendingTimeoutS: json.pending_timeout_s,
|
|
tier: json.tier,
|
|
limited: json.limited,
|
|
heartbeatSec: json.heartbeatSec,
|
|
remainingSec: json.remainingSec,
|
|
};
|
|
}
|
|
|
|
async _setupAudio() {
|
|
// Prefer a context the caller already created + resumed inside the tap
|
|
// gesture (required on iOS). Fall back to creating one here for callers
|
|
// that don't (desktop is lenient about the gesture timing).
|
|
// Most desktops give us 48 kHz, mobiles can give 44.1/24/16 kHz; the
|
|
// capture worklet handles any rate (linear interp fallback).
|
|
const ctx = this.options.audioContext ?? new AudioContext({ latencyHint: "interactive" });
|
|
this._ctx = ctx;
|
|
|
|
// Resume if still suspended. This is best-effort here — on iOS the resume
|
|
// that actually counts is the one the caller did synchronously on tap.
|
|
if (ctx.state === "suspended") {
|
|
try {
|
|
await ctx.resume();
|
|
} catch (err) {
|
|
console.warn("[ws] AudioContext resume failed:", err);
|
|
}
|
|
}
|
|
|
|
// The worklets live at the repo root, one level up from this module.
|
|
const base = new URL("../worklets/", import.meta.url);
|
|
await ctx.audioWorklet.addModule(new URL("mic-capture.js", base).href);
|
|
await ctx.audioWorklet.addModule(new URL("audio-playback.js", base).href);
|
|
|
|
const captureNode = new AudioWorkletNode(ctx, "mic-capture", {
|
|
numberOfInputs: 1,
|
|
numberOfOutputs: 0,
|
|
processorOptions: { chunkMs: MIC_CHUNK_MS },
|
|
});
|
|
captureNode.port.onmessage = (e) => {
|
|
const data = e.data;
|
|
if (data instanceof ArrayBuffer) {
|
|
this._onMicChunk(data);
|
|
} else if (data?.kind === "level") {
|
|
// Raw pre-gate mic RMS for the Settings meter.
|
|
this.dispatchEvent(new CustomEvent("input-level", { detail: { rms: data.rms } }));
|
|
}
|
|
};
|
|
// Push the initial gate config now that the worklet exists.
|
|
captureNode.port.postMessage({ kind: "gate", ...this._noiseGate });
|
|
this._captureNode = captureNode;
|
|
|
|
const micSrc = ctx.createMediaStreamSource(this.options.micStream);
|
|
micSrc.connect(captureNode);
|
|
this._micSrc = micSrc;
|
|
|
|
// Mic analyser: tap the mic in parallel with the worklet so we get the
|
|
// raw (un-resampled, un-clipped) signal for the visualiser.
|
|
const micAnalyser = ctx.createAnalyser();
|
|
micAnalyser.fftSize = VIS_FFT_SIZE;
|
|
micAnalyser.smoothingTimeConstant = 0;
|
|
micSrc.connect(micAnalyser);
|
|
this._micAnalyser = micAnalyser;
|
|
|
|
const playbackNode = new AudioWorkletNode(ctx, "audio-playback", {
|
|
numberOfInputs: 0,
|
|
numberOfOutputs: 1,
|
|
outputChannelCount: [1],
|
|
});
|
|
playbackNode.port.postMessage({ kind: "config", inputRate: OUTPUT_SAMPLE_RATE });
|
|
playbackNode.port.onmessage = (e) => this._onPlaybackMessage(e.data);
|
|
|
|
// Output analyser sits between the playback worklet and the speakers.
|
|
const outAnalyser = ctx.createAnalyser();
|
|
outAnalyser.fftSize = VIS_FFT_SIZE;
|
|
outAnalyser.smoothingTimeConstant = 0.3;
|
|
playbackNode.connect(outAnalyser);
|
|
outAnalyser.connect(ctx.destination);
|
|
this._outAnalyser = outAnalyser;
|
|
this._playbackNode = playbackNode;
|
|
|
|
this._visualiser = new OrbVisualiser(micAnalyser, outAnalyser, () => this._aiSpeaking);
|
|
this._visualiser.start();
|
|
}
|
|
|
|
/** @param {string} connectUrl */
|
|
_openWebSocket(connectUrl) {
|
|
return new Promise((resolve, reject) => {
|
|
// Mask the full URL, which may contain a plaintext session_token.
|
|
// Show host + path and secret fingerprints for auditing without leaks.
|
|
let maskedUrl = connectUrl;
|
|
try {
|
|
const u = new URL(connectUrl, window.location.href);
|
|
const secretPortions = [];
|
|
u.searchParams.forEach((v, k) => {
|
|
const lower = k.toLowerCase();
|
|
const isSecret = /token|session|auth|jwt|secret|key/i.test(lower) || String(v).length > 16;
|
|
secretPortions.push(`${k}=${isSecret ? `✂️…${String(v).slice(-4)} (${String(v).length} chars)` : v}`);
|
|
});
|
|
maskedUrl = `${u.protocol}//${u.host}${u.pathname} ? ${secretPortions.join(" & ") || "(no query parameters)"}`;
|
|
} catch {
|
|
// ignore
|
|
}
|
|
_wsLog(0, "CONNECT", `new WebSocket() → ${maskedUrl}`);
|
|
const ws = new WebSocket(connectUrl);
|
|
ws.binaryType = "arraybuffer";
|
|
this._ws = ws;
|
|
|
|
const onceOpen = () => {
|
|
ws.removeEventListener("open", onceOpen);
|
|
ws.removeEventListener("error", onceErr);
|
|
_wsLog(1, "CONNECT", `WebSocket OPEN at ${maskedUrl} readyState=${ws.readyState}`);
|
|
resolve();
|
|
};
|
|
const onceErr = (e) => {
|
|
ws.removeEventListener("open", onceOpen);
|
|
ws.removeEventListener("error", onceErr);
|
|
_wsLog(3, "CONNECT", `WebSocket FAILED to open: event.type=${e?.type ?? "error"} url=${maskedUrl}`);
|
|
reject(new Error(`WebSocket failed to open: ${e?.type ?? "error"}`));
|
|
};
|
|
ws.addEventListener("open", onceOpen);
|
|
ws.addEventListener("error", onceErr);
|
|
|
|
ws.addEventListener("message", (e) => this._onWsMessage(e.data));
|
|
ws.addEventListener("close", (e) => this._onWsClose(e));
|
|
ws.addEventListener("error", (e) => {
|
|
console.error("[ws] socket error", e);
|
|
_wsLog(3, "ERR", `WebSocket error: event.type=${e?.type ?? "error"} wasClean=${"wasClean" in e ? String(/** @type {CloseEvent} */ (e).wasClean) : "?"}`, e);
|
|
});
|
|
});
|
|
}
|
|
|
|
/**
|
|
* @param {{ kind: string; queuedMs?: number; played?: number }} data
|
|
*/
|
|
_onPlaybackMessage(data) {
|
|
if (data?.kind === "underrun") {
|
|
// Server stopped sending audio mid-response. Most likely the turn
|
|
// ended cleanly (a response.done usually arrives just before/after
|
|
// this). We let the state machine fall back to "connected" via the
|
|
// response.done event handler.
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Mic worklet just sent us a ~40 ms PCM16 16 kHz mono chunk.
|
|
* Base64-encode and forward via the WS.
|
|
* @param {ArrayBuffer} pcm16Buffer
|
|
*/
|
|
_onMicChunk(pcm16Buffer) {
|
|
if (!this._ws || this._ws.readyState !== WebSocket.OPEN) return;
|
|
if (!this._sessionConfigured) return; // Server rejects audio before session.update.
|
|
if (this._muted) return;
|
|
const b64 = base64FromArrayBuffer(pcm16Buffer);
|
|
this._send({ type: "input_audio_buffer.append", audio: b64 });
|
|
}
|
|
|
|
/**
|
|
* @param {string | ArrayBuffer | Blob} raw
|
|
*/
|
|
async _onWsMessage(raw) {
|
|
let text;
|
|
if (typeof raw === "string") {
|
|
text = raw;
|
|
} else if (raw instanceof ArrayBuffer) {
|
|
text = new TextDecoder("utf-8").decode(raw);
|
|
} else if (raw instanceof Blob) {
|
|
text = await raw.text();
|
|
} else {
|
|
return;
|
|
}
|
|
|
|
let event;
|
|
try {
|
|
event = JSON.parse(text);
|
|
} catch {
|
|
return;
|
|
}
|
|
|
|
const type = event?.type;
|
|
if (typeof type !== "string") return;
|
|
// Always-on tracing for high-signal events (lifecycle, transcript/tool calls,
|
|
// errors). For bulk streams (audio/text deltas) we log only count / first
|
|
// + last so the console doesn't scroll into uselessness.
|
|
const isHighSignal = /^(session\.|error$|conversation\.item\.created|conversation\.item\.input_audio_transcription\.completed|response\.created|response\.output_item\.done|response\.audio_transcript\.done|response\.output_audio_transcript\.done|response\.done|response\.function_call_arguments\.done|rate_limits\.updated)/i.test(
|
|
type,
|
|
);
|
|
if (isHighSignal) {
|
|
const isError = type === "error";
|
|
const extra = type.startsWith("conversation.item.input_audio_transcription")
|
|
? ` item=${event.item_id} ci=${event.content_index} — ${String(event.delta ?? event.transcript ?? "").slice(0, 180)}`
|
|
: type.startsWith("response.")
|
|
? ` resp=${String(event.response_id ?? event.response?.id ?? "").slice(0, 10)} status=${event.response?.status ?? "-"} ${String(event.transcript ?? event.text ?? event.delta ?? "").slice(0, 200)}`
|
|
: type === "response.function_call_arguments.done"
|
|
? ` tool=${event.name} callId=${String(event.call_id ?? "").slice(0, 10)} argsPreview=${String(event.arguments ?? "").slice(0, 220)}`
|
|
: type === "error"
|
|
? ` code=${String(event.code ?? "-")} message=${String(event.message ?? "").slice(0, 200)}`
|
|
: type === "rate_limits.updated"
|
|
? ` ${String(JSON.stringify(event.rate_limits ?? [])).slice(0, 220)}`
|
|
: "";
|
|
_wsLog(isError ? 3 : (type.startsWith("response.") && /(text_done|audio_transcript\.done|output_item\.done|output_audio_transcript\.done|function_call_arguments)/i.test(type) ? 1 : 0),
|
|
"MSG",
|
|
`← ${type}${extra}`
|
|
);
|
|
}
|
|
// Opt-in event tracing for diagnosing turn/transcript issues. Enable with
|
|
// `localStorage.setItem("s2s.debug", "1")` in the browser console.
|
|
if (this._debug && !isHighSignal) {
|
|
const extra = type.startsWith("conversation.item.input_audio_transcription")
|
|
? ` item=${event.item_id} ci=${event.content_index} ${event.delta ?? event.transcript ?? ""}`
|
|
: type.startsWith("response.")
|
|
? ` resp=${event.response_id ?? event.response?.id ?? ""} status=${event.response?.status ?? ""} ${event.transcript ?? ""}`
|
|
: "";
|
|
console.debug(`[ws] ${type}${extra}`);
|
|
}
|
|
|
|
switch (type) {
|
|
case "session.created":
|
|
// Server-side defaults for the s2s pipeline are already what we
|
|
// want (server_vad, whisper-1 transcription, PCM16 16k in / 24k
|
|
// out). We only push the user-tunable bits: voice + instructions.
|
|
this._sendSessionUpdate();
|
|
this._sessionConfigured = true;
|
|
if (this._status === "connecting") this._setStatus("connected");
|
|
break;
|
|
|
|
case "session.updated":
|
|
if (this._debug) {
|
|
let toolNames = "";
|
|
if (Array.isArray(event.session?.tools)) {
|
|
/** @type {string[]} */
|
|
const names = [];
|
|
for (const tool of event.session.tools) {
|
|
if (tool && typeof tool.name === "string" && tool.name) names.push(tool.name);
|
|
}
|
|
toolNames = names.join(",");
|
|
}
|
|
console.debug(
|
|
`[Vocero WS] session.updated tool_choice=${event.session?.tool_choice ?? ""} ` +
|
|
`tool_count=${Array.isArray(event.session?.tools) ? event.session.tools.length : 0} ` +
|
|
`tools=${toolNames || "none"}`
|
|
);
|
|
}
|
|
// On the first server acknowledgment, TALK startup should behave
|
|
// exactly like hearing the "Hola Vocero" wake word:
|
|
// 1. Inject "hola vocero" as a user turn through sendUserText().
|
|
// 2. Call requestResponse() without custom greeting instructions,
|
|
// matching the wake-word path in main.js.
|
|
if (this._startWithResponse && !this._initialResponseSent) {
|
|
this._initialResponseSent = true;
|
|
const phrase = this._startWithResponseText.trim() || "hola vocero";
|
|
// #region debug-point B:start-session-initial-response
|
|
reportWsDebug("B", "s2s-ws-client.js:session.updated",
|
|
"emitting initial turn via wake-word style sendUserText + requestResponse() (S2S_START_SESSION_MODE=TALK)",
|
|
{
|
|
toolCount: Array.isArray(event.session?.tools) ? event.session.tools.length : 0,
|
|
phrase,
|
|
});
|
|
// #endregion
|
|
if (this._debug) {
|
|
console.debug(`[Vocero WS] bootstrap turn: sendUserText(${JSON.stringify(phrase)}) → requestResponse()`);
|
|
}
|
|
this.sendUserText(phrase);
|
|
this.requestResponse();
|
|
}
|
|
// Acknowledged by server, nothing to do.
|
|
break;
|
|
|
|
case "input_audio_buffer.speech_started":
|
|
// User started speaking — stop any audio still playing OR queued, every
|
|
// time. We clear unconditionally (not just when `_aiSpeaking`): after a
|
|
// reply or a tool result the worklet's ring buffer can still be draining
|
|
// even though we already flipped `_aiSpeaking` off, and that tail would
|
|
// otherwise keep playing over the user's barge-in.
|
|
this._playbackNode?.port.postMessage({ kind: "clear" });
|
|
this._aiSpeaking = false;
|
|
this._setStatus("user-speaking");
|
|
break;
|
|
|
|
case "input_audio_buffer.speech_stopped":
|
|
if (this._status === "user-speaking") this._setStatus("processing");
|
|
break;
|
|
|
|
case "response.created":
|
|
this._activeResponseId = event.response?.id ?? event.response_id ?? "";
|
|
if (this._suppressNextAssistantResponse && this._activeResponseId && !this._suppressedResponseId) {
|
|
this._suppressedResponseId = this._activeResponseId;
|
|
this._suppressNextAssistantResponse = false;
|
|
this._playbackNode?.port.postMessage({ kind: "clear" });
|
|
// #region debug-point B:suppress-next-response
|
|
reportWsDebug("B", "s2s-ws-client.js:response.created", "marked next speculative response for audio suppression", {
|
|
responseId: this._suppressedResponseId,
|
|
});
|
|
// #endregion
|
|
}
|
|
// #region debug-point B:response-created
|
|
reportWsDebug("B", "s2s-ws-client.js:response.created", "response slot opened", {
|
|
responseId: event.response?.id ?? event.response_id ?? "",
|
|
status: event.response?.status ?? "",
|
|
openResponsesBeforeIncrement: this._openResponses,
|
|
createInFlight: this._createInFlight,
|
|
});
|
|
// #endregion
|
|
if (this._debug) {
|
|
console.debug(
|
|
`[Vocero WS] response.created id=${event.response?.id ?? event.response_id ?? ""} ` +
|
|
`status=${event.response?.status ?? ""}`
|
|
);
|
|
}
|
|
// A response now owns the slot — count it and clear our create guard
|
|
// (this confirms either our create or a server-initiated one).
|
|
this._openResponses++;
|
|
this._createInFlight = false;
|
|
if (this._status === "connected" || this._status === "user-speaking") {
|
|
this._setStatus("processing");
|
|
}
|
|
break;
|
|
|
|
case "response.output_item.added":
|
|
if (this._status === "connected" || this._status === "user-speaking") {
|
|
this._setStatus("processing");
|
|
}
|
|
break;
|
|
|
|
case "response.audio.delta":
|
|
case "response.output_audio.delta": {
|
|
const rid = event.response_id ?? event.response?.id;
|
|
if (rid && this._suppressedResponseId && rid === this._suppressedResponseId) break;
|
|
this._pushAudioDelta(event.delta);
|
|
if (rid) this._audibleResponses.add(rid);
|
|
if (!this._aiSpeaking) {
|
|
this._aiSpeaking = true;
|
|
this._markAudible();
|
|
}
|
|
break;
|
|
}
|
|
|
|
case "response.content_part.added": {
|
|
const part = event.part;
|
|
if (part?.type === "audio" || part?.type === "output_audio") {
|
|
this._markAudible();
|
|
}
|
|
break;
|
|
}
|
|
|
|
case "response.done": {
|
|
// #region debug-point B:response-done
|
|
reportWsDebug("B", "s2s-ws-client.js:response.done", "response slot closed", {
|
|
responseId: event.response?.id ?? "",
|
|
status: event.response?.status ?? "completed",
|
|
openResponsesBeforeDecrement: this._openResponses,
|
|
});
|
|
// #endregion
|
|
this._aiSpeaking = false;
|
|
// This response freed the slot (completion OR cancellation both arrive
|
|
// as response.done). Decrement and, if a create was waiting, replay it.
|
|
this._openResponses = Math.max(0, this._openResponses - 1);
|
|
if (this._status === "ai-speaking" || this._status === "processing") {
|
|
this._setStatus("connected");
|
|
}
|
|
// A response closes here for BOTH normal completion and cancellation
|
|
// (the s2s server signals a speculative-turn interrupt as
|
|
// `response.done` with status "cancelled" — there is no separate
|
|
// `response.cancelled` event). Surface the id + status so the UI can
|
|
// drop a cancelled response's transcript and commit a completed one.
|
|
const status = event.response?.status ?? "completed";
|
|
const responseId = event.response?.id ?? "";
|
|
if (responseId && responseId === this._activeResponseId) this._activeResponseId = "";
|
|
if (responseId && responseId === this._suppressedResponseId) this._suppressedResponseId = "";
|
|
// Did this response ever play audio? Distinguishes a barge-in cut (the
|
|
// user heard part of it) from a speculative response that never played.
|
|
const audible = responseId ? this._audibleResponses.has(responseId) : false;
|
|
this._audibleResponses.delete(responseId);
|
|
// Pull whatever transcript the response carries, falling back to the
|
|
// segments we concatenated from the `*.transcript.done` events (plus any
|
|
// in-progress delta). For an interrupted reply the response payload may
|
|
// be empty, so this is the last chance to capture the text.
|
|
const transcript =
|
|
extractResponseTranscript(event.response) ||
|
|
this._asstDisplay(responseId) ||
|
|
"";
|
|
// Response finished — clear both transcript accumulators for it.
|
|
this._asstTranscriptByResp.delete(responseId);
|
|
this._asstFullByResp.delete(responseId);
|
|
this.dispatchEvent(new CustomEvent("response-finished", {
|
|
detail: { responseId, status, audible, transcript },
|
|
}));
|
|
this._flushDeferredUserText();
|
|
// The slot is free now — replay a queued create (e.g. a tool follow-up
|
|
// that arrived while this response was still running).
|
|
this._flushQueuedCreate();
|
|
break;
|
|
}
|
|
|
|
case "response.function_call_arguments.done": {
|
|
const name = typeof event.name === "string" ? event.name : "";
|
|
const args = typeof event.arguments === "string" ? event.arguments : "{}";
|
|
const callId = typeof event.call_id === "string" ? event.call_id : "";
|
|
// #region debug-point D:function-call-done
|
|
reportWsDebug("D", "s2s-ws-client.js:response.function_call_arguments.done", "received function_call_arguments.done", {
|
|
name,
|
|
callId,
|
|
args,
|
|
});
|
|
// #endregion
|
|
console.debug(
|
|
`[Vocero WS] function_call_arguments.done name=${name || "unknown"} ` +
|
|
`callId=${JSON.stringify(callId)} args=${args}`
|
|
);
|
|
if (name) {
|
|
this.dispatchEvent(new CustomEvent("toolcall", {
|
|
detail: { name, arguments: args, callId },
|
|
}));
|
|
} else {
|
|
// A nameless call can't be executed, so no function_call_output is
|
|
// ever sent and the model would wait forever for a result. The
|
|
// backend shouldn't emit these; warn loudly rather than stall silently.
|
|
console.warn(`[ws] function_call_arguments.done with no name (call_id=${callId}); cannot run tool — turn may stall`);
|
|
}
|
|
break;
|
|
}
|
|
|
|
case "conversation.item.input_audio_transcription.delta": {
|
|
const delta = typeof event.delta === "string" ? event.delta : "";
|
|
if (delta) {
|
|
// `itemId` is REUSED across a speculative continuation, so the UI
|
|
// groups both segments into one message. The delta carries the full
|
|
// cumulative transcript so far (not an increment).
|
|
this.dispatchEvent(
|
|
new CustomEvent("transcript", {
|
|
detail: {
|
|
role: "user",
|
|
text: delta,
|
|
partial: true,
|
|
itemId: typeof event.item_id === "string" ? event.item_id : "",
|
|
},
|
|
}),
|
|
);
|
|
}
|
|
break;
|
|
}
|
|
|
|
case "conversation.item.input_audio_transcription.completed": {
|
|
const transcript = typeof event.transcript === "string" ? event.transcript : "";
|
|
if (transcript) {
|
|
this.dispatchEvent(
|
|
new CustomEvent("transcript", {
|
|
detail: {
|
|
role: "user",
|
|
text: transcript,
|
|
partial: false,
|
|
itemId: typeof event.item_id === "string" ? event.item_id : "",
|
|
},
|
|
}),
|
|
);
|
|
}
|
|
break;
|
|
}
|
|
|
|
case "response.audio_transcript.delta":
|
|
case "response.output_audio_transcript.delta": {
|
|
// Stream the assistant transcript live: accumulate the incremental
|
|
// deltas and push the running text to the UI. Every transcribe event we
|
|
// receive reaches the conversation, so an interrupted reply already has
|
|
// its partial text even if the `.done` never fires.
|
|
this._markAudible();
|
|
const rid = typeof event.response_id === "string" ? event.response_id : "";
|
|
const delta = typeof event.delta === "string" ? event.delta : "";
|
|
if (delta) {
|
|
this._asstTranscriptByResp.set(rid, (this._asstTranscriptByResp.get(rid) || "") + delta);
|
|
// Show completed segments + the segment streaming in right now.
|
|
this.dispatchEvent(
|
|
new CustomEvent("transcript", {
|
|
detail: { role: "assistant", text: this._asstDisplay(rid), partial: true, responseId: rid },
|
|
}),
|
|
);
|
|
}
|
|
break;
|
|
}
|
|
|
|
case "response.audio_transcript.done":
|
|
case "response.output_audio_transcript.done": {
|
|
const rid = typeof event.response_id === "string" ? event.response_id : "";
|
|
// This is ONE completed segment. A response can emit several; concatenate
|
|
// them, space-separated, until response.done clears the accumulator.
|
|
const segment =
|
|
(typeof event.transcript === "string" && event.transcript) ||
|
|
this._asstTranscriptByResp.get(rid) ||
|
|
"";
|
|
this._asstTranscriptByResp.delete(rid); // segment finished; next one starts fresh
|
|
if (segment) {
|
|
const prev = this._asstFullByResp.get(rid) || "";
|
|
this._asstFullByResp.set(rid, prev ? `${prev} ${segment}` : segment);
|
|
}
|
|
const full = this._asstFullByResp.get(rid) || "";
|
|
if (full) {
|
|
this.dispatchEvent(
|
|
new CustomEvent("transcript", {
|
|
detail: { role: "assistant", text: full, partial: false, responseId: rid },
|
|
}),
|
|
);
|
|
}
|
|
break;
|
|
}
|
|
|
|
case "error": {
|
|
const err = event.error;
|
|
console.error("[ws] server error:", err);
|
|
// The "another response is already active" race: our optimistic create
|
|
// collided with a still-running response. Don't surface it — clear the
|
|
// in-flight guard and re-queue, so the create replays on the next
|
|
// response.done (never retried immediately, which would just collide
|
|
// again).
|
|
if (err?.type === "conversation_already_has_active_response" ||
|
|
err?.code === "conversation_already_has_active_response") {
|
|
if (this._createInFlight) {
|
|
this._createInFlight = false;
|
|
// Re-queue a BARE create: any image on the original payload was
|
|
// already sent before this (rejected) create, so don't resend it.
|
|
this._createQueue.push({});
|
|
}
|
|
break;
|
|
}
|
|
// Every other server error is non-fatal: surface it for logging but
|
|
// NEVER tear the socket down. Only transport failures (close / failed
|
|
// open) are fatal, and those come through their own paths.
|
|
this.dispatchEvent(
|
|
new CustomEvent("server-error", { detail: { error: new Error(err?.message ?? "Server error") } }),
|
|
);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
/** @param {string} b64 */
|
|
_pushAudioDelta(b64) {
|
|
if (!this._playbackNode) return;
|
|
if (!b64) return;
|
|
const startedAt = performance.now();
|
|
const bytes = base64ToBytes(b64);
|
|
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
|
|
const samples = new Float32Array(bytes.byteLength / 2);
|
|
for (let i = 0; i < samples.length; i++) {
|
|
const s = view.getInt16(i * 2, true);
|
|
samples[i] = s < 0 ? s / 0x8000 : s / 0x7fff;
|
|
}
|
|
this._playbackNode.port.postMessage({ kind: "audio", samples }, [samples.buffer]);
|
|
const durationMs = performance.now() - startedAt;
|
|
if (durationMs >= 8) {
|
|
// #region debug-point E:audio-delta-slow
|
|
reportWsDebug("E", "s2s-ws-client.js:_pushAudioDelta", "slow output audio delta decode/post", {
|
|
durationMs: Math.round(durationMs),
|
|
base64Length: b64.length,
|
|
sampleCount: samples.length,
|
|
});
|
|
// #endregion
|
|
}
|
|
}
|
|
|
|
/** @param {CloseEvent} ev */
|
|
_onWsClose(ev) {
|
|
console.log("[ws] socket closed:", ev.code, ev.reason);
|
|
_wsLog(
|
|
(ev.code === 1000 || ev.code === 1001) ? 1 : (ev.code ? 3 : 2),
|
|
"CLOSE",
|
|
`WebSocket CLOSED code=${ev.code ?? "?"} reason=${String(ev.reason ?? "(none)").slice(0, 160)} wasClean=${ev.wasClean}`
|
|
);
|
|
if (this._status === "closed" || this._status === "error") return;
|
|
if (ev.code === 1000) {
|
|
this._setStatus("closed");
|
|
} else {
|
|
this.dispatchEvent(
|
|
new CustomEvent("error", {
|
|
detail: { error: new Error(`WebSocket closed (${ev.code}) ${ev.reason || ""}`.trim()) },
|
|
}),
|
|
);
|
|
this._setStatus("error");
|
|
}
|
|
}
|
|
|
|
_sendSessionUpdate() {
|
|
// Minimal payload: only the bits the user is allowed to configure.
|
|
// The s2s server already defaults to server_vad, whisper-1
|
|
// transcription, 16 kHz PCM input and 24 kHz PCM output, so we don't
|
|
// need (and must not send) `audio.input.format`, `audio.input.transcription`,
|
|
// `audio.input.turn_detection` or `audio.output.format`: the pydantic
|
|
// validator on the server rejects the whole event if any unknown or
|
|
// future-shaped sub-field shows up.
|
|
/** @type {Record<string, any>} */
|
|
const session = {
|
|
type: "realtime",
|
|
instructions: this.options.instructions,
|
|
audio: {
|
|
output: { voice: this.options.voice },
|
|
},
|
|
};
|
|
// ───────────────────────────────────────────────────────────────────────────
|
|
// 2026-09-25 INJECT API-PILOT DYNAMIC TOOLS (PASS 1) in session.update.
|
|
// Three safeguards, deduplicated by marker:
|
|
// (A) main.js effectiveReasoningInstructions() appends the list.
|
|
// (B) The server.py Realtime proxy injects it when routing passes there.
|
|
// (C) This client is the final safeguard before sending instructions,
|
|
// including after a live updateSession() replaces the prompt.
|
|
{
|
|
const currentInstr = (typeof session.instructions === "string") ? session.instructions : "";
|
|
if (!instructionsAlreadyHasApipilotPass1(currentInstr)) {
|
|
try {
|
|
const info = typeof window !== "undefined" && typeof window.__s2sGetApipilotPass1ContextInfo === "function"
|
|
? window.__s2sGetApipilotPass1ContextInfo()
|
|
: null;
|
|
const blk = (info && typeof info.block === "string") ? info.block : buildApipilotPass1PromptBlock(null);
|
|
if (blk && blk.trim()) {
|
|
session.instructions = currentInstr
|
|
? `${currentInstr}\n\n${blk}`
|
|
: blk;
|
|
const lenBlk = blk.length;
|
|
const cntTools = (info && typeof info.count === "number") ? info.count : "?";
|
|
_wsLog(
|
|
1, "APIPILOT",
|
|
`_sendSessionUpdate: APPENDED PASS-1 instructions (N=${cntTools}, +${lenBlk} chars). origLen=${currentInstr.length} newLen=${session.instructions.length}`,
|
|
);
|
|
}
|
|
} catch (_) {
|
|
/* never-fail */
|
|
}
|
|
}
|
|
}
|
|
// Tools are declared here; the backend already accepts them in
|
|
// session.update and emits response.function_call_arguments.done when the
|
|
// model decides to call one. Only include the keys when we actually have
|
|
// tools — the server's pydantic validator is strict about shapes.
|
|
if (this._tools.length) {
|
|
session.tools = this._tools;
|
|
session.tool_choice = "auto";
|
|
}
|
|
// #region debug-point B:send-session-update
|
|
reportWsDebug("B", "s2s-ws-client.js:_sendSessionUpdate", "sending initial session.update", {
|
|
toolCount: this._tools.length,
|
|
toolNames: this._tools.map((tool) => tool.name),
|
|
hasInstructions: !!session.instructions,
|
|
instructionLength: (session.instructions || "").length,
|
|
apipilotPass1Present: instructionsAlreadyHasApipilotPass1(session.instructions),
|
|
});
|
|
// #endregion
|
|
this._send({ type: "session.update", session });
|
|
}
|
|
|
|
/** Update voice/instructions on a live session without tearing down. */
|
|
/** @param {{ voice?: string; instructions?: string }} patch */
|
|
updateSession(patch) {
|
|
/** @type {Record<string, any>} */
|
|
const session = { type: "realtime" };
|
|
if (patch.instructions) {
|
|
let instr = patch.instructions;
|
|
// ── 2026-09-25 ApiPilot PASS-1 safeguard for live patches too. ──
|
|
if (typeof instr === "string" && !instructionsAlreadyHasApipilotPass1(instr)) {
|
|
try {
|
|
const info = typeof window !== "undefined" && typeof window.__s2sGetApipilotPass1ContextInfo === "function"
|
|
? window.__s2sGetApipilotPass1ContextInfo()
|
|
: null;
|
|
const blk = (info && typeof info.block === "string") ? info.block : buildApipilotPass1PromptBlock(null);
|
|
if (blk && blk.trim()) {
|
|
const origLen = instr.length;
|
|
instr = `${instr}\n\n${blk}`;
|
|
_wsLog(
|
|
1, "APIPILOT",
|
|
`updateSession: APPENDED PASS-1 on live patch (N=${(info && typeof info.count === "number") ? info.count : "?"}, +${blk.length} chars). origLen=${origLen} newLen=${instr.length}`,
|
|
);
|
|
}
|
|
} catch (_) { /* never-fail */ }
|
|
}
|
|
session.instructions = instr;
|
|
}
|
|
if (patch.voice) session.audio = { output: { voice: patch.voice } };
|
|
if (Object.keys(session).length > 1) {
|
|
// #region debug-point C:update-session
|
|
reportWsDebug("C", "s2s-ws-client.js:updateSession", "sending patch session.update", {
|
|
hasInstructions: !!session.instructions,
|
|
instructionLength: (session.instructions || "").length,
|
|
hasVoice: !!patch.voice,
|
|
apipilotPass1Present: typeof session.instructions === "string" && instructionsAlreadyHasApipilotPass1(session.instructions),
|
|
});
|
|
// #endregion
|
|
this._send({ type: "session.update", session });
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Replace the declared tool set on a live session (e.g. the user flipped a
|
|
* tool switch mid-conversation). Always sends `tools` — an empty array
|
|
* clears them — so toggling the last tool off actually removes it.
|
|
* @param {ToolDef[]} tools
|
|
* @param {"auto"|"required"|"none"} [toolChoice]
|
|
*/
|
|
setTools(tools, toolChoice = tools.length ? "auto" : "none") {
|
|
this._tools = tools;
|
|
// #region debug-point B:set-tools
|
|
reportWsDebug("B", "s2s-ws-client.js:setTools", "replacing live tool set", {
|
|
toolCount: tools.length,
|
|
toolNames: tools.map((tool) => tool.name),
|
|
toolChoice,
|
|
});
|
|
// #endregion
|
|
this._send({
|
|
type: "session.update",
|
|
session: { type: "realtime", tools, tool_choice: toolChoice },
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Return a tool's result to the model. Pairs with the `toolcall` event's
|
|
* `callId`. Caller follows this with `requestResponse()` so the model speaks.
|
|
* @param {string} callId
|
|
* @param {string} output Plain text / JSON string the model will read.
|
|
*/
|
|
sendToolOutput(callId, output) {
|
|
if (!callId) return; // Can't target a result without the call id.
|
|
if (this._debug) {
|
|
console.debug(
|
|
`[Vocero WS] function_call_output callId=${JSON.stringify(callId)} ` +
|
|
`outputLength=${(output || "").length}`
|
|
);
|
|
}
|
|
this._send({
|
|
type: "conversation.item.create",
|
|
item: { type: "function_call_output", call_id: callId, output },
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Add an image to the conversation as user content, so the vision-language
|
|
* model can see it (used by the camera tool). `dataUrl` is a
|
|
* `data:image/jpeg;base64,...` string.
|
|
* @param {string} dataUrl
|
|
*/
|
|
sendUserImage(dataUrl) {
|
|
this._send({
|
|
type: "conversation.item.create",
|
|
item: {
|
|
type: "message",
|
|
role: "user",
|
|
content: [{ type: "input_image", image_url: dataUrl }],
|
|
},
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Add a user text message to the live conversation.
|
|
* @param {string} text
|
|
*/
|
|
sendUserText(text) {
|
|
const content = (text || "").trim();
|
|
if (!content) return;
|
|
if (this._responseActive()) {
|
|
this._deferredUserTextQueue.push(content);
|
|
// #region debug-point B:send-user-text-deferred
|
|
reportWsDebug("B", "s2s-ws-client.js:sendUserText", "deferred MCP prompt until active response ends", {
|
|
textLength: content.length,
|
|
responseActive: true,
|
|
openResponses: this._openResponses,
|
|
createInFlight: this._createInFlight,
|
|
deferredCount: this._deferredUserTextQueue.length,
|
|
});
|
|
// #endregion
|
|
return;
|
|
}
|
|
// #region debug-point B:send-user-text
|
|
reportWsDebug("B", "s2s-ws-client.js:sendUserText", "queueing MCP prompt into live conversation", {
|
|
textLength: content.length,
|
|
preview: content.slice(0, 180),
|
|
responseActive: this._responseActive(),
|
|
openResponses: this._openResponses,
|
|
createInFlight: this._createInFlight,
|
|
});
|
|
// #endregion
|
|
this._send({
|
|
type: "conversation.item.create",
|
|
item: {
|
|
type: "message",
|
|
role: "user",
|
|
content: [{ type: "input_text", text: content }],
|
|
},
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Ask the model to generate a response now (after feeding tool results).
|
|
* Serialized: if a response is already in flight we queue this request and
|
|
* replay it once the active response finishes, so we never trip the
|
|
* backend's `conversation_already_has_active_response` guard.
|
|
*
|
|
* @param {{ image?: string }} [opts] Optional `image` (a data URL) sent as a
|
|
* user `input_image` immediately before this response.create — so the frame
|
|
* travels with the create (and is deferred together with it if queued),
|
|
* rather than being added to the conversation eagerly. Used by the camera
|
|
* tool so the model sees the snapshot in the response it's about to speak.
|
|
*/
|
|
requestResponse(opts = {}) {
|
|
if (this._responseActive()) {
|
|
// #region debug-point B:response-create-queued
|
|
reportWsDebug("B", "s2s-ws-client.js:requestResponse", "queued response.create because slot is busy", {
|
|
openResponses: this._openResponses,
|
|
createInFlight: this._createInFlight,
|
|
pendingCreates: this._createQueue.length + 1,
|
|
});
|
|
// #endregion
|
|
this._createQueue.push(opts);
|
|
if (this._debug) console.debug(`[ws] response.create queued (a response is active); pending=${this._createQueue.length}`);
|
|
return;
|
|
}
|
|
// #region debug-point B:response-create-now
|
|
reportWsDebug("B", "s2s-ws-client.js:requestResponse", "sending response.create immediately", {
|
|
openResponses: this._openResponses,
|
|
createInFlight: this._createInFlight,
|
|
pendingCreates: this._createQueue.length,
|
|
});
|
|
// #endregion
|
|
this._createResponseNow(opts);
|
|
}
|
|
|
|
/** True while a response occupies the single backend slot. */
|
|
_responseActive() {
|
|
return this._openResponses > 0 || this._createInFlight;
|
|
}
|
|
|
|
/** Send a response.create immediately and arm the in-flight guard. Any image
|
|
* on the payload is added as user content right before the create.
|
|
* @param {{ image?: string }} [opts] */
|
|
_createResponseNow(opts = {}) {
|
|
if (!this._ws || this._ws.readyState !== WebSocket.OPEN) return;
|
|
if (opts.image) this.sendUserImage(opts.image);
|
|
this._createInFlight = true;
|
|
if (this._debug) {
|
|
console.debug(`[Vocero WS] response.create image=${!!opts.image} openResponses=${this._openResponses}`);
|
|
}
|
|
this._send({ type: "response.create" });
|
|
}
|
|
|
|
/** Replay one queued response.create if the slot is now free. Called on every
|
|
* response.done, so queued creates drain one-per-completion. */
|
|
_flushQueuedCreate() {
|
|
if (this._createQueue.length > 0 && !this._responseActive()) {
|
|
const opts = this._createQueue.shift();
|
|
if (this._debug) console.debug(`[ws] replaying queued response.create; remaining=${this._createQueue.length}`);
|
|
this._createResponseNow(opts);
|
|
}
|
|
}
|
|
|
|
/** Flush one deferred user text after the current assistant response ends. */
|
|
_flushDeferredUserText() {
|
|
if (this._deferredUserTextQueue.length === 0 || this._responseActive()) return;
|
|
const content = this._deferredUserTextQueue.shift();
|
|
if (!content) return;
|
|
// #region debug-point B:flush-deferred-user-text
|
|
reportWsDebug("B", "s2s-ws-client.js:_flushDeferredUserText", "flushing deferred MCP prompt after response.done", {
|
|
textLength: content.length,
|
|
remainingDeferred: this._deferredUserTextQueue.length,
|
|
});
|
|
// #endregion
|
|
this.sendUserText(content);
|
|
}
|
|
|
|
/** Silence the current or next speculative assistant audio for an MCP fallback turn. */
|
|
suppressAssistantAudio() {
|
|
this._suppressNextAssistantResponse = true;
|
|
if (this._activeResponseId) this._suppressedResponseId = this._activeResponseId;
|
|
this._playbackNode?.port.postMessage({ kind: "clear" });
|
|
this._aiSpeaking = false;
|
|
if (this._status === "ai-speaking") this._setStatus("processing");
|
|
// #region debug-point B:suppress-assistant-audio
|
|
reportWsDebug("B", "s2s-ws-client.js:suppressAssistantAudio", "suppressing speculative assistant audio for MCP fallback", {
|
|
activeResponseId: this._activeResponseId,
|
|
suppressedResponseId: this._suppressedResponseId,
|
|
suppressNextAssistantResponse: this._suppressNextAssistantResponse,
|
|
openResponses: this._openResponses,
|
|
createInFlight: this._createInFlight,
|
|
});
|
|
// #endregion
|
|
}
|
|
|
|
/** Interrupt any active/queued backend response immediately (server-side
|
|
* cancel + client-side queue flush). Use when an external reasoning
|
|
* (OPENAI / NODE_RED) has won the turn and you want to silence the native
|
|
* realtime model entirely — not just its audio playback. */
|
|
cancelActiveResponse() {
|
|
// 1. Silence audio playback like suppressAssistantAudio does.
|
|
this._suppressNextAssistantResponse = true;
|
|
if (this._activeResponseId) this._suppressedResponseId = this._activeResponseId;
|
|
this._playbackNode?.port.postMessage({ kind: "clear" });
|
|
this._aiSpeaking = false;
|
|
if (this._status === "ai-speaking") this._setStatus("processing");
|
|
|
|
// 2. Drop queued user-texts / response.creates injected during the native
|
|
// turn — we don't want them to fire after the reasoning answer.
|
|
const droppedTexts = this._deferredUserTextQueue.length;
|
|
const droppedCreates = this._createQueue.length;
|
|
this._deferredUserTextQueue = [];
|
|
this._createQueue = [];
|
|
|
|
// 3. If the backend has an active response, ask it to truncate/cancel.
|
|
// `conversation.item.truncate` with the active response id is the
|
|
// standard realtime signal for "stop generating this turn". If the
|
|
// backend doesn't accept truncate it will finish but we've already
|
|
// silenced playback + dropped queues so no 2nd reply is produced.
|
|
if (this._activeResponseId) {
|
|
try {
|
|
this._send({
|
|
type: "conversation.item.truncate",
|
|
item_id: this._activeResponseId,
|
|
content_index: 0,
|
|
audio_end_ms: 0,
|
|
});
|
|
} catch {}
|
|
}
|
|
|
|
// #region debug-point B:cancel-active-response
|
|
reportWsDebug("B", "s2s-ws-client.js:cancelActiveResponse",
|
|
"interrupted native realtime turn for external reasoning",
|
|
{
|
|
activeResponseId: this._activeResponseId,
|
|
suppressedResponseId: this._suppressedResponseId,
|
|
droppedDeferredTexts: droppedTexts,
|
|
droppedPendingCreates: droppedCreates,
|
|
});
|
|
// #endregion
|
|
}
|
|
|
|
/** Append an ASSISTANT message to the live conversation. Used by external
|
|
* reasoning (OPENAI / NODE_RED) to insert their computed answer directly
|
|
* as an assistant turn — WITHOUT triggering a new native model response.
|
|
* This deliberately avoids `sendUserText()` (which would make the backend
|
|
* re-answer the "marker text" as a user query, producing duplicate
|
|
* replies) and avoids `requestResponse()` (which would ask the backend
|
|
* to speak another turn on top).
|
|
* @param {string} text
|
|
*/
|
|
appendAssistantText(text) {
|
|
const content = (text || "").trim();
|
|
if (!content) return;
|
|
// #region debug-point B:append-assistant-text
|
|
reportWsDebug("B", "s2s-ws-client.js:appendAssistantText",
|
|
"inserting external reasoning answer as assistant turn",
|
|
{
|
|
textLength: content.length,
|
|
preview: content.slice(0, 180),
|
|
responseActive: this._responseActive(),
|
|
openResponses: this._openResponses,
|
|
createInFlight: this._createInFlight,
|
|
});
|
|
// #endregion
|
|
this._send({
|
|
type: "conversation.item.create",
|
|
item: {
|
|
type: "message",
|
|
role: "assistant",
|
|
content: [{ type: "text", text: content }],
|
|
},
|
|
});
|
|
}
|
|
|
|
/** @param {boolean} muted */
|
|
setMuted(muted) {
|
|
this._muted = muted;
|
|
}
|
|
|
|
/**
|
|
* Update the mic noise gate live (the user moved the Settings cursor).
|
|
* @param {NoiseGate} gate
|
|
*/
|
|
setNoiseGate(gate) {
|
|
this._noiseGate = gate;
|
|
this._captureNode?.port.postMessage({ kind: "gate", ...gate });
|
|
}
|
|
|
|
/** @param {Record<string, unknown>} event */
|
|
_send(event) {
|
|
if (!this._ws || this._ws.readyState !== WebSocket.OPEN) return;
|
|
// Throttle bulk audio event logs (~25 events per second).
|
|
// Log periodic summaries rather than every chunk.
|
|
if (event.type === "input_audio_buffer.append") {
|
|
_wsMicChunkCount += 1;
|
|
const now = performance.now();
|
|
if (_wsMicChunkCount === 1 || now - _wsMicChunkLastLogTs >= 2500 || _wsMicChunkCount % 250 === 0) {
|
|
_wsMicChunkLastLogTs = now;
|
|
_wsLog(0, "SEND:BULK", `→ input_audio_buffer.append chunk #${_wsMicChunkCount} audio_len=${String(event.audio ?? "").length} chars (base64)`);
|
|
}
|
|
let payloadLength = 0;
|
|
let stringifyMs = 0;
|
|
let payload = "";
|
|
if (event.type === "conversation.item.create") {
|
|
const startedAt = performance.now();
|
|
payload = JSON.stringify(event);
|
|
stringifyMs = performance.now() - startedAt;
|
|
payloadLength = payload.length;
|
|
// #region debug-point A:conversation-item-create
|
|
reportWsDebug("A", "s2s-ws-client.js:_send", "sending conversation.item.create payload", {
|
|
payloadLength,
|
|
stringifyMs: Math.round(stringifyMs),
|
|
});
|
|
// #endregion
|
|
}
|
|
if (this._debug && (event.type === "session.update" || event.type === "conversation.item.create")) {
|
|
const preview = (payload || JSON.stringify(event)).slice(0, 400);
|
|
console.debug(`[Vocero WS OUT] type=${event.type} payload=${preview}`);
|
|
}
|
|
this._ws.send(payload || JSON.stringify(event));
|
|
return;
|
|
}
|
|
// Always log non-bulk messages as [S2S-WS:SEND] with a payload preview.
|
|
let payloadLength = 0;
|
|
let stringifyMs = 0;
|
|
let payload = "";
|
|
const startedAt = performance.now();
|
|
payload = JSON.stringify(event);
|
|
stringifyMs = performance.now() - startedAt;
|
|
payloadLength = payload.length;
|
|
if (event.type === "conversation.item.create") {
|
|
// #region debug-point A:conversation-item-create
|
|
reportWsDebug("A", "s2s-ws-client.js:_send", "sending conversation.item.create payload", {
|
|
payloadLength,
|
|
stringifyMs: Math.round(stringifyMs),
|
|
});
|
|
// #endregion
|
|
}
|
|
const isToolish = /tool|function|response\.create/i.test(event.type || "");
|
|
const preview = payload.slice(0, isToolish ? 600 : 260);
|
|
_wsLog(
|
|
isToolish ? 1 : 0,
|
|
"SEND",
|
|
`→ ${String(event.type ?? "?")} payload=${payloadLength} bytes stringify=${Math.round(stringifyMs)} ms — ${preview}${payload.length > (isToolish ? 600 : 260) ? "…" : ""}`
|
|
);
|
|
if (this._debug && (event.type === "session.update" || event.type === "conversation.item.create")) {
|
|
console.debug(`[Vocero WS OUT] type=${event.type} payload=${payload.slice(0, 400)}`);
|
|
}
|
|
this._ws.send(payload);
|
|
}
|
|
|
|
async close() {
|
|
// Abort a queue wait in progress: flag it and wake the poll sleep so
|
|
// `_pollQueue` throws "aborted" and connect() unwinds cleanly.
|
|
this._closed = true;
|
|
if (this._queueWake) {
|
|
clearTimeout(this._queueTimer);
|
|
const wake = this._queueWake;
|
|
this._queueWake = null;
|
|
wake();
|
|
}
|
|
if (this._joinTimer) {
|
|
clearTimeout(this._joinTimer);
|
|
this._joinTimer = 0;
|
|
}
|
|
if (this._joinReject) {
|
|
const reject = this._joinReject;
|
|
this._joinResolve = null;
|
|
this._joinReject = null;
|
|
reject(_codedError("join aborted", "aborted"));
|
|
}
|
|
this._visualiser?.stop();
|
|
this._visualiser = null;
|
|
try {
|
|
if (this._ws && this._ws.readyState <= WebSocket.OPEN) {
|
|
this._ws.close(1000, "client closed");
|
|
}
|
|
} catch {
|
|
// ignored
|
|
}
|
|
this._ws = null;
|
|
|
|
try {
|
|
this._captureNode?.port.close?.();
|
|
} catch {
|
|
// ignored
|
|
}
|
|
try {
|
|
this._micSrc?.disconnect();
|
|
} catch {
|
|
// ignored
|
|
}
|
|
try {
|
|
this._captureNode?.disconnect();
|
|
} catch {
|
|
// ignored
|
|
}
|
|
try {
|
|
this._micAnalyser?.disconnect();
|
|
} catch {
|
|
// ignored
|
|
}
|
|
try {
|
|
this._outAnalyser?.disconnect();
|
|
} catch {
|
|
// ignored
|
|
}
|
|
try {
|
|
this._playbackNode?.disconnect();
|
|
} catch {
|
|
// ignored
|
|
}
|
|
try {
|
|
await this._ctx?.close();
|
|
} catch {
|
|
// ignored
|
|
}
|
|
this._ctx = null;
|
|
this._captureNode = null;
|
|
this._playbackNode = null;
|
|
this._micSrc = null;
|
|
this._micAnalyser = null;
|
|
this._outAnalyser = null;
|
|
this._setStatus("closed");
|
|
}
|
|
}
|