271 lines
12 KiB
JavaScript
271 lines
12 KiB
JavaScript
import assert from "node:assert/strict";
|
|
import { afterEach, test } from "node:test";
|
|
import { VoiceSessionController } from "../../src/voice_assistant/audio_gateway/web/voice-session.js";
|
|
|
|
const controllers = [];
|
|
class Node {
|
|
constructor() { this.disconnected = false; }
|
|
connect() {}
|
|
disconnect() { this.disconnected = true; }
|
|
}
|
|
class Context {
|
|
constructor() {
|
|
this.destination = {};
|
|
this.currentTime = 0;
|
|
this.sources = [];
|
|
this.audioWorklet = { addModule: async () => {} };
|
|
this.nextResume = null;
|
|
}
|
|
createAnalyser() { return new Node(); }
|
|
createGain() { return Object.assign(new Node(), { gain: { value: 1 } }); }
|
|
createMediaStreamSource() { this.input = new Node(); return this.input; }
|
|
async resume() { const pending = this.nextResume; this.nextResume = null; if (pending) await pending; }
|
|
createBuffer(channels, length, sampleRate) {
|
|
return { duration: length / sampleRate, copyToChannel(samples) { this.samples = samples; } };
|
|
}
|
|
createBufferSource() {
|
|
const source = Object.assign(new Node(), {
|
|
stopped: false, onended: null,
|
|
start: () => this.sources.push(source),
|
|
stop() { this.stopped = true; },
|
|
end() { this.onended?.(); },
|
|
});
|
|
return source;
|
|
}
|
|
}
|
|
class Worklet extends Node {
|
|
constructor() {
|
|
super();
|
|
this.port = { messages: [], postMessage(data) { this.messages.push(data); }, onmessage: null };
|
|
}
|
|
}
|
|
class Socket {
|
|
static OPEN = 1;
|
|
static CLOSED = 3;
|
|
constructor(url) { this.url = url; this.readyState = 0; this.sent = []; this.bufferedAmount = 0; }
|
|
send(data) { if (this.sendFailure) throw new Error("Transport failed"); this.sent.push(data); }
|
|
open() { this.readyState = 1; this.onopen?.(); }
|
|
close() { this.readyState = 3; this.onclose?.(); }
|
|
receive(event) { this.onmessage?.({ data: JSON.stringify(event) }); }
|
|
}
|
|
globalThis.WebSocket = Socket;
|
|
globalThis.AudioWorkletNode = Worklet;
|
|
globalThis.location = { protocol: "http:", host: "localhost:8001" };
|
|
afterEach(() => { for (const controller of controllers.splice(0)) controller.close(); });
|
|
const tick = () => new Promise((resolve) => setImmediate(resolve));
|
|
const ready = (overrides = {}) => ({
|
|
type: "session.ready", turn_detection: "server",
|
|
audio: { encoding: "pcm_s16le", sample_rate: 16000, channels: 1 },
|
|
output_audio: { encoding: "pcm_s16le", sample_rate: 48000, channels: 1 },
|
|
max_audio_chunk_bytes: 65536, ...overrides,
|
|
});
|
|
function messages(socket) { return socket.sent.filter((data) => typeof data === "string").map(JSON.parse); }
|
|
async function setup(options = {}, confirm = true) {
|
|
const errors = [], states = [], transcripts = [], replies = [];
|
|
const controller = new VoiceSessionController({
|
|
onError: (error) => errors.push(error), onStateChange: (state) => states.push(state),
|
|
onTranscript: (...args) => transcripts.push(args), onAgentText: (...args) => replies.push(args),
|
|
...options,
|
|
});
|
|
controllers.push(controller);
|
|
const context = new Context();
|
|
await controller.start({ stream: {}, audioContext: context });
|
|
const socket = controller.ws;
|
|
socket.open();
|
|
if (confirm) socket.receive(ready());
|
|
return { controller, context, socket, errors, states, transcripts, replies };
|
|
}
|
|
|
|
test("requests server VAD and starts capture only after session.ready", async () => {
|
|
const { controller, socket } = await setup({}, false);
|
|
assert.equal(messages(socket)[0].turn_detection, "server");
|
|
assert.deepEqual(controller.worklet.port.messages, []);
|
|
controller.handlePCM(new ArrayBuffer(4));
|
|
assert.equal(socket.sent.length, 1);
|
|
socket.receive(ready());
|
|
assert.deepEqual(controller.worklet.port.messages, [{ type: "start", chunkSamples: 1600 }]);
|
|
assert.equal(controller.state, "listening");
|
|
});
|
|
|
|
for (const state of ["listening", "recording", "processing", "speaking", "waiting-playback"]) {
|
|
test(`sends continuous PCM while ${state}, without turn metadata`, async () => {
|
|
const { controller, socket } = await setup();
|
|
controller.setState(state);
|
|
const audio = new ArrayBuffer(4);
|
|
controller.handlePCM(audio);
|
|
assert.deepEqual(messages(socket).at(-1), { type: "audio.input", sequence_number: 0 });
|
|
assert.equal(socket.sent.at(-1), audio);
|
|
});
|
|
}
|
|
|
|
test("global input sequence never resets when backend opens another turn", async () => {
|
|
const { controller, socket } = await setup();
|
|
controller.handlePCM(new ArrayBuffer(4));
|
|
socket.receive({ type: "turn.started", turn_id: 1 });
|
|
controller.handlePCM(new ArrayBuffer(4));
|
|
socket.receive({ type: "interrupted", turn_id: 1 });
|
|
socket.receive({ type: "turn.started", turn_id: 2 });
|
|
controller.handlePCM(new ArrayBuffer(4));
|
|
assert.deepEqual(messages(socket).filter((m) => m.type === "audio.input").map((m) => m.sequence_number), [0, 1, 2]);
|
|
assert.equal(messages(socket).some((m) => ["audio.start", "audio.stop", "interrupt"].includes(m.type)), false);
|
|
assert.equal(controller.worklet.port.messages.length, 1);
|
|
});
|
|
|
|
test("uses the negotiated chunk size for microphone capture", async () => {
|
|
const { controller, socket } = await setup({}, false);
|
|
socket.receive(ready({ max_audio_chunk_bytes: 1000 }));
|
|
assert.deepEqual(controller.worklet.port.messages, [{ type: "start", chunkSamples: 500 }]);
|
|
});
|
|
|
|
test("turn.ended waits for speakers without pausing microphone input", async () => {
|
|
const { controller, context, socket } = await setup();
|
|
socket.receive({ type: "turn.started", turn_id: 1 });
|
|
controller.handleAudioChunk(new ArrayBuffer(8));
|
|
await controller.playbackPending;
|
|
assert.equal(controller.state, "speaking");
|
|
socket.receive({ type: "turn.ended", turn_id: 1 });
|
|
await tick();
|
|
assert.equal(controller.state, "waiting-playback");
|
|
controller.handlePCM(new ArrayBuffer(4));
|
|
assert.equal(messages(socket).at(-1).type, "audio.input");
|
|
context.sources[0].end();
|
|
await tick();
|
|
assert.equal(controller.state, "listening");
|
|
assert.equal(controller.worklet.port.messages.length, 1);
|
|
});
|
|
|
|
test("a new backend turn stops old playback and ignores its pending completion", async () => {
|
|
const { controller, context, socket } = await setup();
|
|
socket.receive({ type: "turn.started", turn_id: 1 });
|
|
controller.handleAudioChunk(new ArrayBuffer(8));
|
|
await controller.playbackPending;
|
|
socket.receive({ type: "turn.ended", turn_id: 1 });
|
|
await tick();
|
|
socket.receive({ type: "turn.started", turn_id: 2 });
|
|
await tick();
|
|
assert.equal(context.sources[0].stopped, true);
|
|
assert.equal(controller.turnId, 2);
|
|
assert.equal(controller.state, "recording");
|
|
});
|
|
|
|
test("old queued PCM is discarded after playback generation changes", async () => {
|
|
const { controller, context, socket } = await setup();
|
|
let release;
|
|
context.nextResume = new Promise((resolve) => { release = resolve; });
|
|
socket.receive({ type: "turn.started", turn_id: 1 });
|
|
controller.handleAudioChunk(new ArrayBuffer(8));
|
|
const obsolete = controller.playbackPending;
|
|
await tick();
|
|
socket.receive({ type: "turn.started", turn_id: 2 });
|
|
release();
|
|
await obsolete;
|
|
assert.equal(context.sources.length, 0);
|
|
controller.handleAudioChunk(new ArrayBuffer(4));
|
|
await controller.playbackPending;
|
|
assert.equal(context.sources.length, 1);
|
|
});
|
|
|
|
test("a failure in obsolete playback does not close the new turn", async () => {
|
|
const { controller, context, socket, errors } = await setup();
|
|
let reject;
|
|
context.nextResume = new Promise((resolve, fail) => { reject = fail; });
|
|
socket.receive({ type: "turn.started", turn_id: 1 });
|
|
controller.handleAudioChunk(new ArrayBuffer(8));
|
|
const obsolete = controller.playbackPending;
|
|
await tick();
|
|
socket.receive({ type: "turn.started", turn_id: 2 });
|
|
reject(new Error("Old playback failed"));
|
|
await obsolete;
|
|
assert.equal(errors.length, 0);
|
|
assert.equal(controller.turnId, 2);
|
|
assert.equal(socket.readyState, Socket.OPEN);
|
|
});
|
|
|
|
test("stale turn events cannot change the current response", async () => {
|
|
const { controller, socket, transcripts, replies } = await setup();
|
|
socket.receive({ type: "turn.started", turn_id: 2 });
|
|
socket.receive({ type: "stt.final", turn_id: 1, text: "old" });
|
|
socket.receive({ type: "agent.response", turn_id: 1, text: "old" });
|
|
socket.receive({ type: "interrupted", turn_id: 1 });
|
|
socket.receive({ type: "turn.ended", turn_id: 1 });
|
|
socket.receive({ type: "turn.started", turn_id: 1 });
|
|
assert.equal(controller.turnId, 2);
|
|
assert.deepEqual(transcripts, []);
|
|
assert.deepEqual(replies, []);
|
|
socket.receive({ type: "stt.final", turn_id: 2, text: "new" });
|
|
socket.receive({ type: "agent.response", turn_id: 2, text: "reply" });
|
|
assert.deepEqual(transcripts, [["new", true]]);
|
|
assert.deepEqual(replies, [["reply", true]]);
|
|
});
|
|
|
|
test("interrupted returns to listening without stopping capture or starting a client turn", async () => {
|
|
const { controller, socket } = await setup();
|
|
socket.receive({ type: "turn.started", turn_id: 1 });
|
|
socket.receive({ type: "interrupted", turn_id: 1 });
|
|
assert.equal(controller.state, "listening");
|
|
assert.equal(controller.turnId, null);
|
|
assert.deepEqual(messages(socket).map((m) => m.type), ["session.start"]);
|
|
assert.equal(controller.worklet.port.messages.length, 1);
|
|
});
|
|
|
|
for (const metadata of [{ turn_detection: "client" }, { turn_detection: undefined }, { max_audio_chunk_bytes: 1 }]) {
|
|
test(`rejects incompatible session metadata: ${JSON.stringify(metadata)}`, async () => {
|
|
const { controller, socket, errors } = await setup({}, false);
|
|
socket.receive(ready(metadata));
|
|
assert.equal(controller.state, "closed");
|
|
assert.equal(errors.length, 1);
|
|
});
|
|
}
|
|
|
|
test("backpressure closes the session before queuing more PCM", async () => {
|
|
const { controller, socket, errors } = await setup({ maxBufferedBytes: 8 });
|
|
socket.bufferedAmount = 8;
|
|
controller.handlePCM(new ArrayBuffer(4));
|
|
assert.equal(controller.state, "closed");
|
|
assert.equal(errors.length, 1);
|
|
assert.equal(messages(socket).some((m) => m.type === "audio.input"), false);
|
|
});
|
|
|
|
test("invalid PCM closes the session without emitting metadata", async () => {
|
|
const { controller, socket, errors } = await setup();
|
|
controller.handlePCM(new ArrayBuffer(3));
|
|
assert.equal(errors.length, 1);
|
|
assert.equal(messages(socket).some((m) => m.type === "audio.input"), false);
|
|
});
|
|
|
|
test("close stops capture and playback even if session.end cannot be sent", async () => {
|
|
const { controller, socket, context } = await setup();
|
|
const worklet = controller.worklet;
|
|
socket.receive({ type: "turn.started", turn_id: 1 });
|
|
controller.handleAudioChunk(new ArrayBuffer(8));
|
|
await controller.playbackPending;
|
|
socket.sendFailure = true;
|
|
controller.close();
|
|
assert.equal(controller.state, "closed");
|
|
assert.equal(worklet.disconnected, true);
|
|
assert.deepEqual(worklet.port.messages.at(-1), { type: "stop" });
|
|
assert.equal(context.sources[0].stopped, true);
|
|
assert.equal(socket.readyState, Socket.CLOSED);
|
|
});
|
|
|
|
test("closing during worklet loading prevents orphan microphone graphs and sockets", async () => {
|
|
let release;
|
|
const context = new Context();
|
|
context.audioWorklet.addModule = () => new Promise((resolve) => { release = resolve; });
|
|
const controller = new VoiceSessionController();
|
|
controllers.push(controller);
|
|
const starting = controller.start({ stream: {}, audioContext: context });
|
|
controller.close();
|
|
release();
|
|
await starting;
|
|
assert.equal(controller.ws, null);
|
|
assert.equal(context.input, undefined);
|
|
});
|
|
|
|
test("missing session.ready times out and closes the connection", async () => {
|
|
const { controller, errors } = await setup({ handshakeTimeoutMs: 5 }, false);
|
|
await new Promise((resolve) => setTimeout(resolve, 20));
|
|
assert.equal(controller.state, "closed");
|
|
assert.equal(errors.length, 1);
|
|
});
|