From 1c2b28bd5b3b61a9c0e9f9dbf7783ee38b8245cf Mon Sep 17 00:00:00 2001 From: Jim Huang Date: Thu, 10 Sep 2026 03:53:55 +0800 Subject: [PATCH 1/3] Keep one recording poll, whatever the overlap Two start attempts could each arm an interval with only one handle between them, leaving a request every fifteen seconds that nothing was left to stop. The terminal verdict also lived on the status node, where the next attempt read it as its own and never asked the one route that could establish whether that attempt had started. One function now owns the handle and clears it in the statement that arms it, and the verdict is returned to the attempt that earned it. Where two attempts overlap the newer one wins, on both sides: a start whose answer a later start has superseded writes nothing, and a poll whose request was made for a previous attempt neither relabels the interview nor stops the timer the current one is relying on. --- tests/browser/recording-consent.test.js | 14 -- tests/browser/recording-state.test.js | 248 ++++++++++++++++++++++++ web/recording-state.js | 89 +++++---- 3 files changed, 297 insertions(+), 54 deletions(-) create mode 100644 tests/browser/recording-state.test.js diff --git a/tests/browser/recording-consent.test.js b/tests/browser/recording-consent.test.js index 9cf6ed4c..51545cdd 100644 --- a/tests/browser/recording-consent.test.js +++ b/tests/browser/recording-consent.test.js @@ -165,20 +165,6 @@ test("recording-consent is recorded before a token is asked for", () => { }); -/// The poll outlives the room otherwise: a request every fifteen seconds about -/// an interview that ended, asking for a word that will not change again. -test("the recording poll is stopped, not just abandoned", () => { - const stop = withoutComments(functionBody(script, "stopRecordingPoll")); - assert.ok(stop.includes("clearInterval(recordingPoll);"), "the interval must be cleared"); - assert.ok(stop.includes("recordingPoll = null;"), "and the handle dropped so a restart cannot double it"); - for (const caller of ["pollRecordingState", "showRecordingState"]) { - assert.ok( - withoutComments(functionBody(script, caller)).includes("stopRecordingPoll()"), - `${caller} must stop the poll rather than leave it running`, - ); - } -}); - test("recording-consent sends the version it displayed", () => { const body = withoutComments(functionBody(script, "recordConsent")); assert.ok(body.includes('fetch("/api/interviews"'), "consent is posted, not assumed"); diff --git a/tests/browser/recording-state.test.js b/tests/browser/recording-state.test.js new file mode 100644 index 00000000..7814f567 --- /dev/null +++ b/tests/browser/recording-state.test.js @@ -0,0 +1,248 @@ +// Run with: node --test tests/browser/recording-state.test.js +// +// The recording label is a promise to a person: the candidate agreed to be +// recorded and is owed the answer to "is it". This runs `web/recording-state.js` +// against stub timers and a stub fetch, so what it asserts is behaviour -- +// which answers arm a poll, which stop it, and that exactly one poll is ever +// live -- rather than the text of the functions. + +import { test } from "node:test"; +import assert from "node:assert/strict"; + +import { failFetchWith } from "./source.js"; +import { + initRecording, + pollRecordingState, + setRecordingPoll, + showRecordingState, + startRecording, +} from "../../web/recording-state.js"; + +function response(status, body = {}) { + return { ok: status >= 200 && status < 300, status, json: async () => body }; +} + +/// Stub timers, fetch and the one node this module writes to. +/// +/// `recordingPoll` is closure-private, so no test can read it. `armed` and +/// `cleared` are how a test sees it instead: a handle that was armed and never +/// cleared is a poll still running, which is the failure these exist to catch +/// and the one nothing else in the tree would notice. +function harness(t, fetchStub) { + const armed = []; + const cleared = []; + const restoreFetch = failFetchWith(fetchStub); + const savedInterval = globalThis.setInterval; + const savedClear = globalThis.clearInterval; + const savedWarn = console.warn; + // The handle is the position, counting from one, so a test can name the + // interval a particular attempt armed rather than count clears. + globalThis.setInterval = (callback, delay) => armed.push(delay); + globalThis.clearInterval = (handle) => void cleared.push(handle); + console.warn = () => {}; + const nodes = { recordingState: { hidden: true, textContent: "" } }; + initRecording({ state: { interviewId: "interview/one" }, nodes, recordingEnabled: true }); + t.after(() => { + setRecordingPoll(false); + restoreFetch(); + globalThis.setInterval = savedInterval; + globalThis.clearInterval = savedClear; + console.warn = savedWarn; + }); + return { nodes, armed, cleared }; +} + +/// The verdict belongs to the attempt that earned it. +/// +/// A terminal answer used to be left on the node, where the next attempt read +/// it as its own and never polled the one route that could have established +/// whether that attempt started. A refused start is not a verdict: a start +/// that worked and could not be read back looks identical from here. +test("a start after a terminal answer polls again", async (t) => { + const { nodes, armed } = harness(t, async () => response(200, { state: "ready" })); + + await startRecording(); + assert.deepEqual(armed, [], "a recording already saved is not asked about"); + + globalThis.fetch = async () => response(503, { error: "temporarily unavailable" }); + await startRecording(); + assert.equal(nodes.recordingState.textContent, "Checking whether the recording started."); + assert.deepEqual(armed, [15_000], "the attempt that could not be read back must be polled for"); +}); + +/// One live poll, whatever the overlap. +/// +/// The request sits between the two, so two attempts can both be inside +/// `startRecording` before either has an answer. A newer terminal answer must +/// keep its verdict when the older nonterminal answer eventually arrives. +test("a stale start cannot rearm a terminal recording poll", async (t) => { + let releaseFirst; + const firstAnswer = new Promise((resolve) => { + releaseFirst = resolve; + }); + let calls = 0; + const { armed, cleared } = harness(t, async () => { + if ((calls += 1) === 1) { + await firstAnswer; + return response(200, { state: "starting" }); + } + return response(200, { state: "ready" }); + }); + + const first = startRecording(); + const second = startRecording(); + await second; + releaseFirst(); + await first; + + assert.deepEqual(armed, [], "an older nonterminal answer cannot restart polling"); + assert.ok(cleared.includes(null), "the terminal answer stops the current poll"); +}); + +/// One live poll, arming included. +/// +/// The generation guard settles which overlapping attempt owns the answer, but +/// a second attempt that starts after the first has finished is current in its +/// own right and reaches the arming with a poll already running. Clearing and +/// arming in one statement is what stops that from leaving an interval with +/// nothing holding its handle. +test("arming a poll clears the one already running", async (t) => { + const { armed, cleared } = harness(t, async () => response(503, { error: "unavailable" })); + + await startRecording(); + assert.deepEqual(armed, [15_000], "the first attempt polls"); + + await startRecording(); + + assert.deepEqual(armed, [15_000, 15_000], "so does the second"); + assert.ok(cleared.includes(1), "and the first attempt's interval must not be left running"); +}); + +/// A poll belongs to the attempt that armed it. +/// +/// The request sits across a start, so an answer fetched for the previous +/// attempt can land after a newer one has taken over the timer. Applying it +/// would label the interview from a request that is no longer about it, and a +/// terminal one would stop the poll the current attempt is relying on. +test("a poll from a previous attempt cannot stop the current one", async (t) => { + let releasePoll; + const pollAnswer = new Promise((resolve) => { + releasePoll = resolve; + }); + let calls = 0; + const { nodes, armed, cleared } = harness(t, async () => { + calls += 1; + if (calls === 2) { + await pollAnswer; + return response(200, { state: "ready" }); + } + return response(200, { state: "starting" }); + }); + + await startRecording(); + const stale = pollRecordingState(); + await startRecording(); + assert.deepEqual(armed, [15_000, 15_000], "the newer attempt armed its own poll"); + + releasePoll(); + await stale; + + assert.equal( + nodes.recordingState.textContent, + "Recording is starting.", + "a stale answer must not relabel the interview", + ); + assert.ok(!cleared.includes(2), "nor stop the poll the current attempt is relying on"); +}); + +/// The body is a second await, and the generation can move across it. +/// +/// Checking once when the headers arrive is not enough: the answer is still +/// being parsed, and a start that lands in that window takes the timer over +/// before the verdict is applied to it. +test("a poll answer parsed after a newer start is discarded", async (t) => { + let releaseBody; + let reachedBody; + const body = new Promise((resolve) => { + releaseBody = resolve; + }); + const parsing = new Promise((resolve) => { + reachedBody = resolve; + }); + let calls = 0; + const { nodes, cleared } = harness(t, async () => { + calls += 1; + if (calls === 2) { + return { + ok: true, + status: 200, + json: () => { + reachedBody(); + return body; + }, + }; + } + return response(200, { state: "starting" }); + }); + + await startRecording(); + const stale = pollRecordingState(); + await parsing; + await startRecording(); + + releaseBody({ state: "ready" }); + await stale; + + assert.equal( + nodes.recordingState.textContent, + "Recording is starting.", + "a body parsed after a newer start must not relabel the interview", + ); + assert.ok(!cleared.includes(2), "nor stop the poll that start armed"); +}); + +/// Gone, not ours, or signed out: none of them change by asking again. +test("a status route that has nothing to say stops the poll", async (t) => { + const { nodes, cleared } = harness(t, async () => response(200, { state: "starting" })); + + await startRecording(); + globalThis.fetch = async () => response(404); + await pollRecordingState(); + + assert.equal( + nodes.recordingState.textContent, + "The recording did not start. The interview is not being recorded.", + ); + // The armed handle by name, not `cleared.length`: the clear is + // unconditional, so a count is satisfied by clearing nothing. + assert.ok(cleared.includes(1), "the handle that was armed is the handle that is cleared"); +}); + +/// Terminal on the answer's own terms, not on the status code. +/// +/// `ready` arrives with a 200, so the route cannot be what stops the poll; the +/// state in the body has to, through the verdict `showRecordingState` reports. +test("a terminal state stops the poll and says so to its caller", async (t) => { + const { cleared } = harness(t, async () => response(200, { state: "starting" })); + + await startRecording(); + assert.equal(showRecordingState({ state: "ready" }), true, "a saved recording is terminal"); + globalThis.fetch = async () => response(200, { state: "ready" }); + await pollRecordingState(); + + assert.ok(cleared.includes(1), "nothing keeps asking after a terminal state"); +}); + +/// Stopped, not just abandoned. +/// +/// Clearing the interval and keeping the handle would let the next stop clear +/// a timer id the platform has already reused. +test("the poll drops the handle it cleared", async (t) => { + const { cleared } = harness(t, async () => response(200, { state: "starting" })); + + await startRecording(); + setRecordingPoll(false); + setRecordingPoll(false); + + assert.deepEqual(cleared.slice(-2), [1, null], "cleared once, then there is nothing to clear"); +}); diff --git a/web/recording-state.js b/web/recording-state.js index c29c4355..deb0c5a0 100644 --- a/web/recording-state.js +++ b/web/recording-state.js @@ -24,12 +24,26 @@ export function initRecording(deps) { } let recordingPoll = null; +let recordingAttempt = 0; /// Slower than the timer on purpose: the states a recording moves through are /// minutes apart, and a poll per second would be a request per second for a /// word that does not change. const RECORDING_POLL_MS = 15000; +/// Whether to keep asking, and the only place `recordingPoll` is read or +/// written. +/// +/// Clearing and arming in one statement is what keeps it to one timer. Every +/// caller below is reached after an await, so a stop and an arm with anything +/// between them would let two attempts each end up holding an interval with +/// only one handle between them: a request every fifteen seconds for the life +/// of the page, and nothing left that can stop it. +export function setRecordingPoll(on) { + clearInterval(recordingPoll); + recordingPoll = on ? setInterval(pollRecordingState, RECORDING_POLL_MS) : null; +} + /// Whether the recording notice has been agreed to, or does not apply. export function consentGiven() { return !recordingEnabled || nodes.recordingConsent.checked; @@ -41,47 +55,56 @@ export function consentGiven() { /// silent failure here is the worst outcome available: they believe the /// interview is being kept and it is not. export async function startRecording() { + const attempt = ++recordingAttempt; + // This attempt's own answer, not a flag read back off the node afterwards. A + // poll already in flight can land in between and answer for an attempt that + // is over, and the answer it carries is about the start before this one. + let terminal = false; try { const response = await fetch(`/api/interviews/${encodeURIComponent(state.interviewId)}/recording`, { method: "POST", }); if (!response.ok) throw new Error((await response.json())?.error || "The recording could not be started."); - showRecordingState(await response.json()); + const status = await response.json(); + if (attempt === recordingAttempt) terminal = showRecordingState(status); } catch (error) { - // Still polls. A refused start and a start that succeeded and could not be - // read back look the same from here, and the status route is the thing that - // can tell them apart: it answers 404 when there is nothing, and polling - // stops on that. - // Neutral, not a verdict. This call cannot tell a refused start from one - // that worked and could not be read back, and the status route can. - console.warn("codetrial recording_start_failed", error); - nodes.recordingState.hidden = false; - nodes.recordingState.textContent = "Checking whether the recording started."; - } - // One timer, cleared first. `connect` runs again on a reconnect, and an - // interval per attempt is a request per attempt per period forever. Armed - // only when the first answer was not already terminal, because - // `showRecordingState` clears it and an interval created afterwards would - // survive one pointless request. - clearInterval(recordingPoll); - if (!nodes.recordingState.dataset.settled) { - recordingPoll = setInterval(pollRecordingState, RECORDING_POLL_MS); + if (attempt === recordingAttempt) { + // Still polls. A refused start and a start that succeeded and could not + // be read back look the same from here, and the status route is the + // thing that can tell them apart: it answers 404 when there is nothing, + // and polling stops on that. + console.warn("codetrial recording_start_failed", error); + nodes.recordingState.hidden = false; + nodes.recordingState.textContent = "Checking whether the recording started."; + } } + if (attempt === recordingAttempt) setRecordingPoll(!terminal); } export async function pollRecordingState() { if (!state.interviewId) return; + // The attempt this answer will be about. A start that lands while the + // request is in flight takes the timer over, and this answer is then about + // the interview before it: relabelling from it says the wrong thing, and + // acting on a terminal one stops a poll the newer attempt is relying on. + const attempt = recordingAttempt; try { const response = await fetch(`/api/interviews/${encodeURIComponent(state.interviewId)}/recording`); + if (attempt !== recordingAttempt) return; if (response.ok) { - showRecordingState(await response.json()); + // Parsed first and checked again: the body is a second await, and a + // start that lands while it is being read owns the timer by the time + // the verdict would be applied to it. + const status = await response.json(); + if (attempt !== recordingAttempt) return; + if (showRecordingState(status)) setRecordingPoll(false); return; } // Gone, not ours, or signed out. None of these change by asking again, and // a session that expired would otherwise be a request every fifteen seconds // for the rest of the page's life. if ([401, 403, 404].includes(response.status)) { - stopRecordingPoll(); + setRecordingPoll(false); nodes.recordingState.hidden = false; // Every one of these is terminal, so every one of them replaces the // label. Stopping the poll and leaving "Checking whether the recording @@ -92,7 +115,6 @@ export async function pollRecordingState() { response.status === 401 ? "Sign in again to see the recording status." : "The recording did not start. The interview is not being recorded."; - nodes.recordingState.dataset.settled = "1"; } } catch (error) { // The interview is the thing that matters; a status that cannot be read is @@ -103,6 +125,10 @@ export async function pollRecordingState() { /// The state in words a candidate can act on. "failed" is not an instruction, /// so the recovery the server returns is what picks the sentence. +/// +/// Reports whether the state is terminal and does nothing about it. Both +/// callers are asking on behalf of a poll, and which of them owns that poll is +/// not this function's to know. export function showRecordingState(status) { const words = { starting: "Recording is starting.", @@ -135,16 +161,8 @@ export function showRecordingState(status) { // A `failed` recording whose recovery is another delivery attempt is not // finished: the transfer queue can still deliver it, and the candidate is the // person who wants to know when it does. - const settled = ["ready", "deleted", "cleanup_failed"].includes(status?.state) + return ["ready", "deleted", "cleanup_failed"].includes(status?.state) || (status?.state === "failed" && status?.recovery !== "retry_delivery"); - // Recorded on the element rather than in a variable, because `startRecording` - // reads it after this runs and the two are not in the same call. - if (settled) { - nodes.recordingState.dataset.settled = "1"; - stopRecordingPoll(); - } else { - delete nodes.recordingState.dataset.settled; - } } /// Takes consent back, mid-interview. @@ -180,12 +198,3 @@ export async function withdrawRecordingConsent() { setBanner("connection", "Could not stop the recording. Try again, or end the interview."); } } - -/// Stops asking about a recording once the interview is over. -/// -/// The interval outlives the room otherwise: an inert timer at best, and a -/// request every fifteen seconds for a word that will not change at worst. -export function stopRecordingPoll() { - clearInterval(recordingPoll); - recordingPoll = null; -} From 7ad462e8175258109fbf4bcff52a895518cd396b Mon Sep 17 00:00:00 2001 From: Jim Huang Date: Thu, 10 Sep 2026 03:53:55 +0800 Subject: [PATCH 2/3] Move the last test helpers out of src A production file carrying a cfg(test) helper is a test the crate ships, and two were left: a parse shim in gemini.rs and an OutputAudio fixture in livekit/media.rs. The fixture sat there on the belief that a test module cannot hand one to a sibling, which is true of siblings and not of these two: media's tests descend from livekit, so a single copy under tests/unit serves both. What production keeps is a real constructor. The rate the fixture publishes at is now the constant on both sides of the comparison that reads it, so moving that constant fails the test rather than leaving it to assert a number nothing sends. --- src/gemini.rs | 5 ---- src/livekit/media.rs | 58 +++++++++++-------------------------- tests/unit/gemini.rs | 7 +++-- tests/unit/livekit.rs | 49 +++++++++++++++++++++++++------ tests/unit/livekit/media.rs | 28 +++++++----------- 5 files changed, 72 insertions(+), 75 deletions(-) diff --git a/src/gemini.rs b/src/gemini.rs index c3135522..6397e44e 100644 --- a/src/gemini.rs +++ b/src/gemini.rs @@ -806,11 +806,6 @@ struct ServerMessage { resumption_handle: Option, } -#[cfg(test)] -fn parse_server_events(text: &str) -> Vec { - parse_server_message(text).events -} - fn parse_server_message(text: &str) -> ServerMessage { let Ok(message) = serde_json::from_str::(text) else { return ServerMessage::default(); diff --git a/src/livekit/media.rs b/src/livekit/media.rs index 6277e041..098e76ce 100644 --- a/src/livekit/media.rs +++ b/src/livekit/media.rs @@ -23,11 +23,8 @@ use jpeg_encoder::{ColorType, Encoder}; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; -use crate::gemini::GeminiLiveSession; - -#[cfg(test)] -use super::GEMINI_OUTPUT_AUDIO_SAMPLE_RATE; use super::is_interview_participant; +use crate::gemini::GeminiLiveSession; pub(super) const GEMINI_AUDIO_SAMPLE_RATE: i32 = 16_000; @@ -297,6 +294,21 @@ pub(super) struct QueuedOutputFrame { } impl OutputAudio { + pub(super) fn new( + source: NativeAudioSource, + sample_rate: u32, + frames: mpsc::Sender, + ) -> Self { + Self { + source, + sample_rate, + pending_bytes: Vec::new(), + playout_deadline: Instant::now(), + frames, + output_cancellation: CancellationToken::new(), + } + } + pub(super) fn interrupt(&mut self) { self.output_cancellation.cancel(); self.output_cancellation = CancellationToken::new(); @@ -436,14 +448,7 @@ pub(super) async fn publish_output_audio( sample_rate, queued_frames, )); - Ok(OutputAudio { - source, - sample_rate, - pending_bytes: Vec::new(), - playout_deadline: Instant::now(), - frames, - output_cancellation: CancellationToken::new(), - }) + Ok(OutputAudio::new(source, sample_rate, frames)) } pub(super) fn append_pcm16_bytes(frame: &AudioFrame<'_>, bytes: &mut Vec) { @@ -593,35 +598,6 @@ pub(super) async fn encode_video_frame_jpeg_off_thread( .map_err(|error| format!("jpeg encode task did not finish: {error}"))? } -/// An `OutputAudio` wired to a channel instead of a room. -/// -/// At module scope rather than inside `mod tests` because both this -/// file's tests and the room loop's in `livekit.rs` build one, and it -/// used to live in `livekit.rs` where the fields it sets are private to -/// here. A test module cannot export it to a sibling; this can. -#[cfg(test)] -pub(super) fn test_output_audio( - pending_bytes: Vec, -) -> (OutputAudio, mpsc::Receiver) { - let (frames, queued_frames) = mpsc::channel(4); - ( - OutputAudio { - source: NativeAudioSource::new( - AudioSourceOptions::default(), - GEMINI_OUTPUT_AUDIO_SAMPLE_RATE, - LIVEKIT_OUTPUT_CHANNELS, - LIVEKIT_OUTPUT_QUEUE_MS, - ), - sample_rate: GEMINI_OUTPUT_AUDIO_SAMPLE_RATE, - pending_bytes, - playout_deadline: Instant::now(), - frames, - output_cancellation: CancellationToken::new(), - }, - queued_frames, - ) -} - #[cfg(test)] #[path = "../../tests/unit/livekit/media.rs"] mod tests; diff --git a/tests/unit/gemini.rs b/tests/unit/gemini.rs index c947f893..14761a72 100644 --- a/tests/unit/gemini.rs +++ b/tests/unit/gemini.rs @@ -588,8 +588,8 @@ fn tool_response_message_matches_live_websocket_shape() { } #[test] -fn parse_server_events_extracts_audio_transcripts_and_tool_calls() { - let events = parse_server_events( +fn parse_server_message_extracts_audio_transcripts_and_tool_calls() { + let events = parse_server_message( r#"{ "serverContent": { "modelTurn": { @@ -609,7 +609,8 @@ fn parse_server_events_extracts_audio_transcripts_and_tool_calls() { ] } }"#, - ); + ) + .events; assert_eq!( events, diff --git a/tests/unit/livekit.rs b/tests/unit/livekit.rs index 9078cd4e..45f5c049 100644 --- a/tests/unit/livekit.rs +++ b/tests/unit/livekit.rs @@ -4,13 +4,44 @@ //! test and not an integration test: private items are in scope. use super::*; +use ::livekit::webrtc::audio_source::AudioSourceOptions; +use ::livekit::webrtc::audio_source::native::NativeAudioSource; +use tokio::sync::mpsc; + +use crate::config::load_from_pairs; // Only the tests below reach for it now: the report packet that carries this // topic is built in `report.rs`, and what is left here asserts which topics the // loop refuses from a browser. use crate::runtime::TOPIC_REPORT; -use crate::config::load_from_pairs; +/// An `OutputAudio` wired to a channel instead of a room. +/// +/// `pub(super)` and defined here rather than in each of the two test modules +/// that build one: `media`'s tests are a descendant of `livekit`, so they can +/// name it, and a second copy is a second thing to keep in step with the +/// fields. +/// +/// The rate is the constant rather than the 24000 it expands to, because +/// `accepts` compares it against the rate in the mime type a caller hands +/// `capture`. A fixture pinned to the literal on both sides would keep passing +/// while testing a rate production had moved off. +pub(super) fn test_output_audio() -> (OutputAudio, mpsc::Receiver) { + let (frames, queued_frames) = mpsc::channel(4); + ( + OutputAudio::new( + NativeAudioSource::new( + AudioSourceOptions::default(), + GEMINI_OUTPUT_AUDIO_SAMPLE_RATE, + LIVEKIT_OUTPUT_CHANNELS, + LIVEKIT_OUTPUT_QUEUE_MS, + ), + GEMINI_OUTPUT_AUDIO_SAMPLE_RATE, + frames, + ), + queued_frames, + ) +} /// The interview starts from the plan its token was minted for. /// @@ -433,7 +464,7 @@ fn a_pending_reply_is_what_the_advisory_reads_as_in_flight() { ); // Cleared when the turn is cut, and the same advisory is then due. - let (mut output_audio, _frames) = test_output_audio(Vec::new()); + let (mut output_audio, _frames) = test_output_audio(); cut_off_turn(&mut activity, &mut output_audio); assert!(!activity.reply_in_flight(), "a cut turn owes nothing"); assert!(deferred.take_if_due( @@ -838,7 +869,7 @@ fn recent_typing_holds_off_the_periodic_review() { #[test] fn wrap_up_wait_finishes_after_turn_and_playout_complete() { let now = Instant::now(); - let (mut output_audio, _) = test_output_audio(Vec::new()); + let (mut output_audio, _) = test_output_audio(); let mut activity = RuntimeActivity::new(now); let drained = now - Duration::from_secs(1); let playing = now + Duration::from_secs(1); @@ -947,7 +978,7 @@ fn agent_state_attributes_preserve_existing_values() { /// emptied queue as the turn being over, so the interview ended there. #[test] fn the_closing_message_is_not_cut_short_by_a_candidate_talking_over_it() { - let (mut output_audio, _frames) = test_output_audio(Vec::new()); + let (mut output_audio, _frames) = test_output_audio(); let closing = output_audio.output_cancellation.clone(); output_audio.playout_deadline = Instant::now() + Duration::from_secs(10); let mut activity = RuntimeActivity::new(Instant::now()); @@ -965,7 +996,7 @@ fn the_closing_message_is_not_cut_short_by_a_candidate_talking_over_it() { /// queueing this whole mechanism exists to avoid. #[test] fn only_a_chunk_that_will_be_queued_is_worth_dropping_a_turn_for() { - let (output_audio, _frames) = test_output_audio(Vec::new()); + let (output_audio, _frames) = test_output_audio(); let rate = GEMINI_OUTPUT_AUDIO_SAMPLE_RATE; assert!(output_audio.accepts(&[1, 0], &format!("audio/pcm;rate={rate}"))); @@ -1030,7 +1061,7 @@ fn the_reply_latency_stamp_is_armed_and_cleared_by_the_floor() { // A turn that gets cut takes the pending measurement with it: the candidate // is talking again, so nobody is waiting. - let (mut output_audio, _frames) = test_output_audio(Vec::new()); + let (mut output_audio, _frames) = test_output_audio(); output_audio.playout_deadline = Instant::now() + Duration::from_secs(3); cut_off_turn(&mut activity, &mut output_audio); assert_eq!( @@ -1067,7 +1098,7 @@ fn a_reply_nobody_was_measured_waiting_for_reports_no_latency() { /// length of speech nobody heard. #[test] fn a_cut_off_turn_stamps_the_moment_it_was_cut_not_when_it_would_have_ended() { - let (mut output_audio, _frames) = test_output_audio(Vec::new()); + let (mut output_audio, _frames) = test_output_audio(); let cut = output_audio.output_cancellation.clone(); output_audio.playout_deadline = Instant::now() + Duration::from_secs(15); let mut activity = RuntimeActivity::new(Instant::now()); @@ -1092,7 +1123,7 @@ fn a_cut_off_turn_stamps_the_moment_it_was_cut_not_when_it_would_have_ended() { /// draining. #[test] fn a_candidate_speaking_over_a_draining_turn_drops_what_is_left_of_it() { - let (mut output_audio, _frames) = test_output_audio(Vec::new()); + let (mut output_audio, _frames) = test_output_audio(); let stale = output_audio.output_cancellation.clone(); output_audio.playout_deadline = Instant::now() + Duration::from_secs(10); let mut activity = RuntimeActivity::new(Instant::now()); @@ -1136,7 +1167,7 @@ fn nothing_is_dropped_once_the_queue_has_drained_or_while_the_agent_speaks() { "candidate already holds the floor", ), ] { - let (mut output_audio, _frames) = test_output_audio(Vec::new()); + let (mut output_audio, _frames) = test_output_audio(); let live = output_audio.output_cancellation.clone(); output_audio.playout_deadline = deadline; let mut activity = RuntimeActivity::new(Instant::now()); diff --git a/tests/unit/livekit/media.rs b/tests/unit/livekit/media.rs index 5dfdbaa4..f40e7797 100644 --- a/tests/unit/livekit/media.rs +++ b/tests/unit/livekit/media.rs @@ -6,6 +6,8 @@ use super::*; use crate::config::load_from_pairs; +use crate::livekit::GEMINI_OUTPUT_AUDIO_SAMPLE_RATE; +use crate::livekit::tests::test_output_audio; use ::livekit::webrtc::video_frame::{I420Buffer, VideoBuffer, VideoFrame, VideoRotation}; /// A sink that records instead of dialling Gemini, and can be told to fail. @@ -53,20 +55,8 @@ fn the_agent_voice_is_published_as_a_microphone() { /// re-arms the timer for zero and spins. #[test] fn playout_ends_on_its_deadline() { - let (frames, _queued) = tokio::sync::mpsc::channel(1); - let output_audio = OutputAudio { - source: NativeAudioSource::new( - AudioSourceOptions::default(), - 24_000, - LIVEKIT_OUTPUT_CHANNELS, - LIVEKIT_OUTPUT_QUEUE_MS, - ), - sample_rate: 24_000, - pending_bytes: Vec::new(), - playout_deadline: Instant::now() + Duration::from_secs(1), - frames, - output_cancellation: CancellationToken::new(), - }; + let (mut output_audio, _frames) = test_output_audio(); + output_audio.playout_deadline = Instant::now() + Duration::from_secs(1); let deadline = output_audio.playout_deadline; assert!(output_audio.playing_at(deadline - Duration::from_nanos(1))); assert!( @@ -396,7 +386,8 @@ fn pcm16_bytes_serializes_little_endian_samples() { #[test] fn output_audio_interrupt_clears_partial_pcm_frame_and_cancels_old_queue() { - let (mut output_audio, _) = test_output_audio(vec![1, 2, 3]); + let (mut output_audio, _) = test_output_audio(); + output_audio.pending_bytes = vec![1, 2, 3]; let old_cancellation = output_audio.output_cancellation.clone(); output_audio.playout_deadline = Instant::now() + Duration::from_secs(10); @@ -410,13 +401,16 @@ fn output_audio_interrupt_clears_partial_pcm_frame_and_cancels_old_queue() { #[tokio::test] async fn output_audio_capture_queues_frames_with_current_cancellation_token() { - let (mut output_audio, mut queued_frames) = test_output_audio(Vec::new()); + let (mut output_audio, mut queued_frames) = test_output_audio(); let start_deadline = output_audio.playout_deadline; let bytes = vec![0_u8; 240 * 2]; assert!( output_audio - .capture(&bytes, "audio/pcm;rate=24000") + .capture( + &bytes, + &format!("audio/pcm;rate={GEMINI_OUTPUT_AUDIO_SAMPLE_RATE}"), + ) .await .unwrap() ); From 4fe0c6ead8a248f5a284606bb8a546253509df6c Mon Sep 17 00:00:00 2001 From: Jim Huang Date: Thu, 10 Sep 2026 03:53:56 +0800 Subject: [PATCH 3/3] Split the longest functions into named steps The room loop had grown to three hundred lines of select arm bodies, and reading any one of them meant scrolling past the other six. Each arm is a handler now, taking the event context this module already speaks rather than a second bundle of the same four borrows, and the two exits that hang up do so through the borrowing teardown gemini.rs already had instead of naming themselves for a caller to perform. The same split runs through the token handler, the setup submission and the browser preflight, where the level meter and the readiness sample move to modules a test can drive: what pinned them before was the text of their functions, and one of those assertions forbade a cleanup this makes. Splitting an unjudged function multiplies the names the mutation gate has to be told about, so five handlers join the exclusions carrying no code it could not judge a moment ago. The gate earned one deletion in exchange: the parse arm in minted_token answered a request that token_handler had already refused, and neither of its mutants could die because nothing tells a guard that is never true from one that is never false. --- .cargo/mutants.toml | 33 +- src/livekit.rs | 901 +++++++++++++++++------------ src/web/setup.rs | 295 ++++++---- src/web/token.rs | 281 +++++---- tests/browser/avatar.test.js | 15 +- tests/browser/dom-contract.test.js | 14 +- tests/browser/mic-meter.test.js | 235 ++++++++ web/audio-check.js | 55 ++ web/interview.js | 168 ++---- web/mic-meter.js | 111 ++++ 10 files changed, 1375 insertions(+), 733 deletions(-) create mode 100644 tests/browser/mic-meter.test.js create mode 100644 web/mic-meter.js diff --git a/.cargo/mutants.toml b/.cargo/mutants.toml index 350e66a9..5b81f1df 100644 --- a/.cargo/mutants.toml +++ b/.cargo/mutants.toml @@ -1,7 +1,7 @@ # Functions the mutation gate cannot judge, because `cargo test` cannot reach # them. Matched against the mutant names that `cargo mutants --list` prints. # -# EXCLUSIONS: 39 +# EXCLUSIONS: 48 # # That number is checked by `scripts/test.sh`, so adding an entry means editing # this line too. The point is not the count, it is that the list only ever grows @@ -19,6 +19,28 @@ # ignored. What covers this path is the browser interview check and # scripts/browser-check.sh, not unit tests. # +# `on_watch_tick`, `on_gemini_event` and `on_playout_settled` are three of +# `run_room`'s select arms, lifted into named handlers so the loop reads as the +# seven things it dispatches rather than three hundred lines of arm bodies. +# `on_turn_complete` is the same move one layer down, out of +# `handle_gemini_event`. Every mutant `cargo mutants` finds in them was already +# in this list a moment ago, inside the two functions they came out of, and is +# unreachable for the same reason: none of it runs without a LiveKit room and a +# Gemini socket. Four names where there were two is the honest cost of the +# split, and it buys no new unjudged code -- the same lines were unjudged when +# they were arm bodies. The sibling handlers the same change created +# (`on_hard_deadline`, `on_tool_calls`, `on_output_transcript`, +# `on_input_transcript`, `on_generated_audio`, `on_interruption`) are absent +# from the list on purpose was wrong, and the gate said so: `cargo mutants` +# replaces a whole function body as well as flipping operators, so five of them +# grew a `with Ok(())` mutant that no test can kill for the same reason their +# parent's could not. They are listed now. `on_hard_deadline` is still absent, +# and for a different reason worth writing down: its mutants are unviable +# rather than unreachable, because `ControlFlow` has no `new` or `From` for +# `cargo mutants` to call, so they never compile and never run. A return type +# that gains one would make them viable, and the gate going red is how that +# would announce itself. +# # `open_session` is the first half of `run_room`, split out of it: it joins the # room, waits for a candidate, brings up the audio track and the Gemini session, # and greets. Every line of it needs a LiveKit server for the same reason the @@ -198,6 +220,15 @@ exclude_re = [ "replace > with >= in create_interview", "replace > with == in create_interview", "run_room", + "on_watch_tick", + "on_gemini_event", + "on_playout_settled", + "on_turn_complete", + "on_tool_calls", + "on_output_transcript", + "on_input_transcript", + "on_generated_audio", + "on_interruption", "open_session", "run_gemini_check", "run_livekit", diff --git a/src/livekit.rs b/src/livekit.rs index 71fb9893..43b4fef7 100644 --- a/src/livekit.rs +++ b/src/livekit.rs @@ -718,6 +718,256 @@ async fn open_session<'a>( })) } +/// The room loop's own bookkeeping, distinct from the session it runs on. +/// +/// Four things only the loop reads, bundled so an arm that needs three of them +/// takes one parameter rather than three. +struct RoomLoop { + /// How many sockets this interview has been through. + restarts: usize, + deferred_restart: DeferredRestart, + + /// At most one idle-window review at a time, collected on the watch tick. + /// A tick of latency on a note nobody is waiting for is not worth an arm + /// in the select. + interim_review: InterimReview, + + /// Checked on the watch tick rather than given its own timer arm: the tick + /// already runs, and a resolution of one tick is plenty for a ninety-second + /// grace. + presence: CandidatePresence, +} + +/// The interview reached the deadline the server holds for it. +async fn on_hard_deadline( + room: &Room, + context: &mut GeminiEventContext<'_>, + loops: &mut RoomLoop, + interview: InterviewContext<'_>, +) -> Result, Box> { + eprintln!( + "interview reached its server-side deadline: room={} duration={}min", + interview.boot.room_name, interview.boot.duration_min + ); + + end_through_control( + room, + context, + interview, + "time_up", + &mut loops.interim_review, + ) + .await?; + // `end_through_control` has already published the report and left the room. + Ok(ControlFlow::Break(())) +} + +/// One watch tick: the candidate's absence, the interim review, and the nudge. +async fn on_watch_tick( + room: &Room, + context: &mut GeminiEventContext<'_>, + loops: &mut RoomLoop, + interview: InterviewContext<'_>, +) -> Result, Box> { + // One reading of the clock for the whole tick. Three calls gave three + // instants microseconds apart, so a cooldown stamped from one and tested + // against another was answering a question nobody asked. + let tick_at = Instant::now(); + if loops.presence.gave_up(tick_at) { + // No report: it would be graded from a session the candidate walked out + // of, and there is nobody in the room to receive it. The browser writes + // the report the candidate actually sees. + eprintln!( + "candidate did not return to room={}; ending", + interview.boot.room_name + ); + + // `shutdown` rather than `close`: it is the same teardown for a caller + // holding a borrow, which is what every handler here has. + context.gemini.shutdown().await?; + leave_room(room).await; + return Ok(ControlFlow::Break(())); + } + + if let Some(review) = loops.interim_review.finished() { + match review.await { + Ok(notes) => record_interim_notes(context.state, ¬es), + + // A panicked or cancelled review is a review that did not happen. + // The slot is already clear, so the next pause takes it. + Err(error) => { + eprintln!("interim review ended abnormally: {error}"); + } + } + } + + // Ahead of the nudge below, and cheap when it declines. The pause this + // reads is the same pause the interview is spending anyway, and what it + // buys is a final reviewer that arrives at a session somebody has already + // read. + // + // Not a shorter wait: the report call is bounded at REPORT_TIMEOUT and + // measures seconds, and the transcript still reaches it whole, so this is + // bought for what the report is written from rather than for when it lands. + // The ten minutes in issue 31 were the interview's own clock, and it is + // `end_interview` that answers those. + if !loops.interim_review.is_running() + && context + .activity + .claim_interim_review(context.state, tick_at) + { + loops + .interim_review + .start(spawn_interim_review(context.state, interview)); + } + if let Some(prompt) = context.activity.watch_prompt(context.state, tick_at) { + // Not `?`. Every write below is one the reader may be about to explain: + // a socket Gemini has closed fails the next send long before + // `next_event` drains and reports it, and audio flushes every hundred + // milliseconds, so propagating here is how a closed socket ended the + // interview from the write side without the arm that resumes it ever + // running. + if let Err(error) = context.gemini.send_text(&prompt).await { + eprintln!("Gemini nudge failed ({error}); waiting for the close to be reported"); + return Ok(ControlFlow::Continue(())); + } + context.activity.mark_speaking(); + } + + Ok(ControlFlow::Continue(())) +} + +/// One Gemini event, plus the two restarts only this arm can decide. +async fn on_gemini_event( + room: &Room, + context: &mut GeminiEventContext<'_>, + event: Option, + loops: &mut RoomLoop, + interview: InterviewContext<'_>, +) -> Result, Box> { + let Some(event) = event else { + // Gemini hung up. Usually that is the ten-minute cap on a single + // connection rather than anything wrong, so the session continues on a + // new socket instead of ending the interview. The close itself performs + // the replacement, so any advisory still held is already paid for. + loops.deferred_restart.cancel(); + if replace_gemini_session(room, context, interview, &mut loops.restarts) + .await? + .is_break() + { + return Ok(ControlFlow::Break(())); + } + return Ok(ControlFlow::Continue(())); + }; + if let GeminiEvent::GoAway { time_left } = &event { + eprintln!( + "Gemini requested a transport restart in {time_left}; room={}", + interview.boot.room_name + ); + + // A turn still generating can be carrying a tool call or an audio + // fragment that has not reached the room. Generated audio is also still + // live work: replace_gemini_session clears its queue, so wait for the + // playout arm to drain it. + if loops.deferred_restart.request( + context.activity.floor, + context.output_audio.is_playing(), + context.activity.reply_in_flight(), + context.activity.tool_response_outstanding, + ) && replace_gemini_session(room, context, interview, &mut loops.restarts) + .await? + .is_break() + { + return Ok(ControlFlow::Break(())); + } + return Ok(ControlFlow::Continue(())); + } + handle_gemini_event(room, context, event, Interruptible::Yes).await?; + + // Jim called `end_interview`. Fed through the same packet the browser and + // the server-side deadline both send, for the reason the deadline arm + // gives: the wrap-up, the report and the teardown are then the ones that + // are already tested. + // + // Read here rather than inside the tool call because this is where the + // room, the report and the way out of the loop are all in scope. + // + // Held until the tool response's own generation has landed. Gemini owes one + // for every tool response, so acting on the flag the instant it is set + // means the closing is requested while that acknowledgement is still + // coming: it is the acknowledgement's `TurnComplete` that settles the + // output, `send_wrap_up_and_wait` returns on it, and the closing turn -- + // the thanks the candidate hears and the `skipped` evidence calls the + // wrap-up asks for -- is cut off by the shutdown behind it. `TurnComplete`, + // `Interrupted` and a socket replacement all clear the flag, so this is a + // beat, not a condition that can hold the interview open. Not while paused. + // The closing would be generated into a room whose output this loop drops + // (`output_disposition`), so the candidate hears none of it and is handed a + // report out of a silence they did not know had ended. Pausing also clears + // the outstanding-response flag, so without this the next event of any kind + // ends the interview. Held instead until they come back; if they never do, + // the deadline still ends it. + if ready_to_close(context.state, context.activity) { + eprintln!( + "interviewer ended the interview: room={}", + interview.boot.room_name + ); + end_through_control( + room, + context, + interview, + "interview_complete", + &mut loops.interim_review, + ) + .await?; + return Ok(ControlFlow::Break(())); + } + + // Asked after every event, not only after a turn boundary. A candidate + // talking over the draining queue empties it through `drop_stale_playout` + // on an `InputTranscript`, and Gemini sends no `Interrupted` for a turn it + // already considers finished, so a boundary-only check waits out the whole + // `timeLeft` and lets the socket drop cold instead. + if loops.deferred_restart.take_if_due( + context.activity.floor, + context.output_audio.is_playing(), + context.activity.tool_response_outstanding, + ) && replace_gemini_session(room, context, interview, &mut loops.restarts) + .await? + .is_break() + { + return Ok(ControlFlow::Break(())); + } + + Ok(ControlFlow::Continue(())) +} + +/// The queued audio finished playing, so the floor is the candidate's again. +async fn on_playout_settled( + room: &Room, + context: &mut GeminiEventContext<'_>, + loops: &mut RoomLoop, + interview: InterviewContext<'_>, +) -> Result, Box> { + if context.output_audio.is_playing() { + return Ok(ControlFlow::Continue(())); + } + context.activity.mark_listening(); + set_agent_state(room, context.agent_state, AGENT_STATE_LISTENING).await?; + if loops.deferred_restart.take_if_due( + context.activity.floor, + context.output_audio.is_playing(), + context.activity.tool_response_outstanding, + ) && replace_gemini_session(room, context, interview, &mut loops.restarts) + .await? + .is_break() + { + return Ok(ControlFlow::Break(())); + } + + Ok(ControlFlow::Continue(())) +} + pub async fn run_room( config: &AgentConfig, room_name: &str, @@ -750,22 +1000,15 @@ pub async fn run_room( candidate: &candidate_identity, now_seconds, }; - - let mut restarts = 0usize; - let mut deferred_restart = DeferredRestart::default(); - - // At most one idle-window review at a time, collected on the watch tick - // below. A tick of latency on a note nobody is waiting for is not worth an - // arm in the select. - let mut interim_review = InterimReview::default(); + let mut loops = RoomLoop { + restarts: 0, + deferred_restart: DeferredRestart::default(), + interim_review: InterimReview::default(), + presence: CandidatePresence::default(), + }; let mut watch = tokio::time::interval(Duration::from_secs_f64(WATCH_TICK_S)); - // Checked on the watch tick rather than given its own timer arm: the tick - // already runs, and a resolution of one tick is plenty for a ninety-second - // grace. - let mut presence = CandidatePresence::default(); - // The interview's own deadline, held by the process that owns the room // rather than by the candidate's tab. The browser countdown is a display: // it is throttled when the tab is hidden, and it stops entirely on an OS @@ -781,83 +1024,16 @@ pub async fn run_room( tokio::pin!(hard_deadline); loop { - tokio::select! { + let step = tokio::select! { () = &mut hard_deadline, if !turn.state.ended => { - eprintln!( - "interview reached its server-side deadline: room={} duration={}min", - boot.room_name, boot.duration_min - ); - - let mut context = turn.context(&mut output_audio, &mut gemini, media.identity.as_deref()); - end_through_control( - &room, - &mut context, - interview, - "time_up", - &mut interim_review, - ) - .await?; - return Ok(()); + let mut context = + turn.context(&mut output_audio, &mut gemini, media.identity.as_deref()); + on_hard_deadline(&room, &mut context, &mut loops, interview).await? } _ = watch.tick(), if !turn.state.ended => { - // One reading of the clock for the whole tick. Three calls gave - // three instants microseconds apart, so a cooldown stamped from - // one and tested against another was answering a question - // nobody asked. - let tick_at = Instant::now(); - if presence.gave_up(tick_at) { - // No report: it would be graded from a session the - // candidate walked out of, and there is nobody in the room - // to receive it. The browser writes the report the - // candidate actually sees. - eprintln!("candidate did not return to room={room_name}; ending"); - gemini.close().await?; - leave_room(&room).await; - return Ok(()); - } - - if let Some(review) = interim_review.finished() { - match review.await { - Ok(notes) => record_interim_notes(&mut turn.state, ¬es), - - // A panicked or cancelled review is a review that did - // not happen. The slot is already clear, so the next - // pause takes it. - Err(error) => { - eprintln!("interim review ended abnormally: {error}"); - } - } - } - - // Ahead of the nudge below, and cheap when it declines. The - // pause this reads is the same pause the interview is spending - // anyway, and what it buys is a final reviewer that arrives at - // a session somebody has already read. - // - // Not a shorter wait: the report call is bounded at - // REPORT_TIMEOUT and measures seconds, and the transcript still - // reaches it whole, so this is bought for what the report is - // written from rather than for when it lands. The ten minutes - // in issue 31 were the interview's own clock, and it is - // `end_interview` that answers those. - if !interim_review.is_running() - && turn.activity.claim_interim_review(&turn.state, tick_at) - { - interim_review.start(spawn_interim_review(&mut turn.state, interview)); - } - if let Some(prompt) = turn.activity.watch_prompt(&turn.state, tick_at) { - // Not `?`. Every write below is one the reader may be about - // to explain: a socket Gemini has closed fails the next - // send long before `next_event` drains and reports it, and - // audio flushes every hundred milliseconds, so propagating - // here is how a closed socket ended the interview from the - // write side without the arm that resumes it ever running. - if let Err(error) = gemini.send_text(&prompt).await { - eprintln!("Gemini nudge failed ({error}); waiting for the close to be reported"); - continue; - } - turn.activity.mark_speaking(); - } + let mut context = + turn.context(&mut output_audio, &mut gemini, media.identity.as_deref()); + on_watch_tick(&room, &mut context, &mut loops, interview).await? } event = events.recv() => { let Some(event) = event else { @@ -868,6 +1044,11 @@ pub async fn run_room( gemini.close().await?; return Ok(()); }; + + // Media first, because most events are, and because attaching a + // track needs the stream and the socket apart -- which is the + // one thing a context, which borrows both together, cannot + // give. match handle_media_event( &mut media, &mut gemini, @@ -877,145 +1058,38 @@ pub async fn run_room( ) .await { - Ok(true) => continue, - Ok(false) => {} + Ok(true) => ControlFlow::Continue(()), + Ok(false) => { + let mut context = turn.context( + &mut output_audio, + &mut gemini, + media.identity.as_deref(), + ); + handle_room_event( + &room, + &mut context, + &mut loops.presence, + interview, + &ids, + event, + ) + .await? + } Err(error) => { eprintln!("Gemini media attach failed ({error}); waiting for the close to be reported"); - continue; + ControlFlow::Continue(()) } } - let mut context = turn.context(&mut output_audio, &mut gemini, media.identity.as_deref()); - if handle_room_event(&room, &mut context, &mut presence, interview, &ids, event) - .await? - .is_break() - { - return Ok(()); - } } event = gemini.next_event() => { - let Some(event) = event else { - // Gemini hung up. Usually that is the ten-minute cap on a - // single connection rather than anything wrong, so the - // session continues on a new socket instead of ending the - // interview. The close itself performs the replacement, so - // any advisory still held is already paid for. - deferred_restart.cancel(); - let mut context = turn.context(&mut output_audio, &mut gemini, media.identity.as_deref()); - if replace_gemini_session(&room, &mut context, interview, &mut restarts) - .await? - .is_break() - { - return Ok(()); - } - continue; - }; - if let GeminiEvent::GoAway { time_left } = &event { - eprintln!( - "Gemini requested a transport restart in {time_left}; room={room_name}" - ); - - // A turn still generating can be carrying a tool call or an - // audio fragment that has not reached the room. Generated - // audio is also still live work: replace_gemini_session - // clears its queue, so wait for the playout arm to drain - // it. - if deferred_restart.request( - turn.activity.floor, - output_audio.is_playing(), - turn.activity.reply_in_flight(), - turn.activity.tool_response_outstanding, - ) { - let mut context = turn.context(&mut output_audio, &mut gemini, media.identity.as_deref()); - if replace_gemini_session(&room, &mut context, interview, &mut restarts) - .await? - .is_break() - { - return Ok(()); - } - } - continue; - } - let mut context = turn.context(&mut output_audio, &mut gemini, media.identity.as_deref()); - handle_gemini_event(&room, &mut context, event, Interruptible::Yes).await?; - - // Jim called `end_interview`. Fed through the same packet the - // browser and the server-side deadline both send, for the - // reason the deadline arm gives: the wrap-up, the report and - // the teardown are then the ones that are already tested. - // - // Read here rather than inside the tool call because this is - // where the room, the report and the way out of the loop are - // all in scope. - // - // Held until the tool response's own generation has landed. - // Gemini owes one for every tool response, so acting on the - // flag the instant it is set means the closing is requested - // while that acknowledgement is still coming: it is the - // acknowledgement's `TurnComplete` that settles the output, - // `send_wrap_up_and_wait` returns on it, and the closing turn - // -- the thanks the candidate hears and the `skipped` evidence - // calls the wrap-up asks for -- is cut off by the shutdown - // behind it. `TurnComplete`, `Interrupted` and a socket - // replacement all clear the flag, so this is a beat, not a - // condition that can hold the interview open. Not while paused. - // The closing would be generated into a room whose output this - // loop drops (`output_disposition`), so the candidate hears - // none of it and is handed a report out of a silence they did - // not know had ended. Pausing also clears the - // outstanding-response flag, so without this the next event of - // any kind ends the interview. Held instead until they come - // back; if they never do, the deadline still ends it. - if ready_to_close(context.state, context.activity) { - eprintln!("interviewer ended the interview: room={room_name}"); - end_through_control( - &room, - &mut context, - interview, - "interview_complete", - &mut interim_review, - ) - .await?; - return Ok(()); - } - - // Asked after every event, not only after a turn boundary. A - // candidate talking over the draining queue empties it through - // `drop_stale_playout` on an `InputTranscript`, and Gemini - // sends no `Interrupted` for a turn it already considers - // finished, so a boundary-only check waits out the whole - // `timeLeft` and lets the socket drop cold instead. - if deferred_restart - .take_if_due( - context.activity.floor, - context.output_audio.is_playing(), - context.activity.tool_response_outstanding, - ) - && replace_gemini_session(&room, &mut context, interview, &mut restarts) - .await? - .is_break() - { - return Ok(()); - } + let mut context = + turn.context(&mut output_audio, &mut gemini, media.identity.as_deref()); + on_gemini_event(&room, &mut context, event, &mut loops, interview).await? } _ = tokio::time::sleep_until(output_audio.playout_deadline.into()), if turn.activity.floor == Floor::AwaitingPlayout && output_audio.is_playing() => { - if output_audio.is_playing() { - continue; - } - turn.activity.mark_listening(); - set_agent_state(&room, &mut turn.agent_state, AGENT_STATE_LISTENING).await?; - if deferred_restart.take_if_due( - turn.activity.floor, - output_audio.is_playing(), - turn.activity.tool_response_outstanding, - ) { - let mut context = turn.context(&mut output_audio, &mut gemini, media.identity.as_deref()); - if replace_gemini_session(&room, &mut context, interview, &mut restarts) - .await? - .is_break() - { - return Ok(()); - } - } + let mut context = + turn.context(&mut output_audio, &mut gemini, media.identity.as_deref()); + on_playout_settled(&room, &mut context, &mut loops, interview).await? } frame = next_audio_frame(&mut media.audio), if media.audio.is_some() => { // The fastest writer in the loop, and so the one that reaches a @@ -1035,6 +1109,7 @@ pub async fn run_room( // ended stream, and there is no path through here that wants to // hold on to one. release_if_ended(&mut media.audio, ended); + ControlFlow::Continue(()) } frame = next_video_frame(&mut media.video), if media.video.is_some() => { if turn.state.paused { @@ -1042,7 +1117,11 @@ pub async fn run_room( } else if let Err(error) = pump_video(&mut media, &mut gemini, frame).await { eprintln!("Gemini video write failed ({error}); waiting for the close to be reported"); } + ControlFlow::Continue(()) } + }; + if step.is_break() { + return Ok(()); } } } @@ -1558,6 +1637,10 @@ fn output_disposition(event: &GeminiEvent, discarding: bool, paused: bool) -> Ou OutputDisposition::Deliver } +/// Gemini said something. One arm each, because the arms share only the socket +/// they arrived on: what a tool call has to do and what a cut-off turn has to +/// undo have no step in common, and reading either one used to mean scrolling +/// past the other five. async fn handle_gemini_event( room: &Room, context: &mut GeminiEventContext<'_>, @@ -1574,186 +1657,234 @@ async fn handle_gemini_event( OutputDisposition::Deliver => {} } match event { - GeminiEvent::ToolCall(calls) => { - for call in calls { - let shown_before = framework_progress(context.state); - let response = execute_tool_call(context.state, &call); - context.gemini.send_tool_response(&call, response).await?; - - // Gemini now owes a generation for this, and will deliver it on - // this socket or not at all. - context.activity.tool_response_outstanding = true; - if checklist_changed(&shown_before, context.state) { - publish_framework_progress(room, context.state).await?; - } - } - } - GeminiEvent::OutputTranscript(text) => { - // `text` is passed whole, not the trimmed form: fragments have to - // be concatenated exactly as received. The trimmed view only - // decides whether this event carried anything at all. - if transcript_text(&text).is_some() { - let turn = &mut context.turns.interviewer; - let whole = turn - .record(&mut context.state.transcript, "Interviewer", &text) - .to_string(); - publish_transcript(room, &whole, turn.segment_id("interviewer"), false, None) - .await?; - } - } + GeminiEvent::ToolCall(calls) => on_tool_calls(room, context, calls).await, + GeminiEvent::OutputTranscript(text) => on_output_transcript(room, context, &text).await, GeminiEvent::InputTranscript(text) => { - if let (Some(_), Some(identity)) = (transcript_text(&text), context.candidate_identity) - { - // The candidate is talking over audio Gemini finished producing - // a while ago. Gemini will not call this an interruption, - // because as far as it is concerned that turn ended when it - // stopped generating; only this side knows the queue is still - // draining. Cut it here or the reply lands behind the rest of - // the old turn. - drop_stale_playout(room, context, interruptible).await?; - context.activity.note_candidate_finished(Instant::now()); - let turn = &mut context.turns.candidate; - let whole = turn - .record(&mut context.state.transcript, CANDIDATE_SPEAKER, &text) - .to_string(); - publish_transcript( - room, - &whole, - turn.segment_id("candidate"), - false, - Some(identity), - ) - .await?; - } + on_input_transcript(room, context, &text, interruptible).await } GeminiEvent::Audio { bytes, mime_type } => { - // Read before the interrupt below, because that is the point the - // candidate stops waiting. Stamped on the last input transcript - // fragment, so it covers endpointing plus model latency plus - // anything still queued ahead of the reply: the silence the - // candidate actually sits through. - // - // `None` means Gemini is answering something it never transcribed, - // and there is no moment the candidate finished to measure from. - // Printing anything then is worse than printing nothing. - let waited = context.activity.awaiting_reply_since; - - // A new turn's first chunk while the previous one is still - // draining. `InputTranscript` normally clears the queue before this - // point, so reaching here means Gemini answered something it never - // transcribed. Backstop rather than the main path, and it must run - // before `capture` or the new audio queues behind the old. - // - // Only for a chunk that will actually be queued. Dropping ahead of - // a chunk `capture` rejects leaves the candidate with a sentence - // cut in half and no reply behind it. - if context.output_audio.accepts(&bytes, &mime_type) { - drop_stale_playout(room, context, interruptible).await?; - } - if context.output_audio.capture(&bytes, &mime_type).await? { - // Cleared here rather than where it is read: a chunk `capture` - // rejects is not the reply starting, and consuming the stamp on - // one would lose the measurement for the chunk that is. - if let Some(since) = waited { - context.activity.awaiting_reply_since = None; - eprintln!( - "timing: {:.2}s from the candidate finishing to the reply starting", - since.elapsed().as_secs_f64() - ); - } - context.activity.mark_speaking(); - - // Speech is queued, not played: the floor stays busy until the - // buffered audio actually finishes. - context.activity.last_agent_speech = context.output_audio.playout_deadline; - set_agent_state(room, context.agent_state, AGENT_STATE_SPEAKING).await?; - } + on_generated_audio(room, context, &bytes, &mime_type, interruptible).await } - GeminiEvent::TurnComplete => { - // Whatever the tool response was owed has now arrived. - context.activity.tool_response_outstanding = false; - - // The gap between Gemini finishing and the queue emptying. Gemini - // synthesises far faster than speech plays, so this is how long the - // agent will still be talking after it has stopped thinking, and - // therefore how long a candidate answering now would have waited - // before `drop_stale_playout` existed. - let backlog = context - .output_audio - .playout_deadline - .saturating_duration_since(Instant::now()); - - // Only a backlog a candidate would notice. `!is_zero()` fired on a - // millisecond and printed "0.0s", so every one of these lines in a - // real session said nothing at all. - if backlog >= NOTABLE_PLAYOUT_BACKLOG { - eprintln!( - "timing: turn generated, {:.1}s of it still to play", - backlog.as_secs_f64() - ); - } + GeminiEvent::TurnComplete => on_turn_complete(room, context).await, + GeminiEvent::Interrupted => on_interruption(room, context).await, - // Gemini finishing its turn also means the candidate utterance it - // answered is over, so both sides close here. - close_turns(room, context).await?; - context.activity.floor = Floor::AwaitingPlayout; - if !context.output_audio.is_playing() { - context.activity.mark_listening(); - set_agent_state(room, context.agent_state, AGENT_STATE_LISTENING).await?; - } + // Named rather than left to the catch-all: the room loop intercepts + // this before dispatching, so the only way one arrives here is through + // `send_wrap_up_and_wait`, where the interview ends within + // `WRAP_UP_WAIT` and there is no socket left to replace. + GeminiEvent::GoAway { .. } => Ok(()), + _ => Ok(()), + } +} + +/// Answers every call in the batch, and republishes the checklist when one of +/// them moved it. +async fn on_tool_calls( + room: &Room, + context: &mut GeminiEventContext<'_>, + calls: Vec, +) -> Result<(), Box> { + for call in calls { + let shown_before = framework_progress(context.state); + let response = execute_tool_call(context.state, &call); + context.gemini.send_tool_response(&call, response).await?; + + // Gemini now owes a generation for this, and will deliver it on this + // socket or not at all. + context.activity.tool_response_outstanding = true; + if checklist_changed(&shown_before, context.state) { + publish_framework_progress(room, context.state).await?; } - GeminiEvent::Interrupted => { - // A barge-in cancels a pending close. `cut_off_turn` below clears - // `tool_response_outstanding`, which is the only thing holding the - // end back, so without this the candidate who says "wait, actually" - // over Jim's acknowledgement clears the hold with the same event - // that carries their objection, and is cut off and handed a report. - // Ending is the one decision here nobody can take back, and someone - // who has just started a sentence is not finished. Jim can call the - // tool again once they are. - if std::mem::take(&mut context.state.end_requested) { - eprintln!("interviewer's ending cancelled: the candidate spoke over the close"); - } + } + Ok(()) +} + +/// What the interviewer said, as it is said. +async fn on_output_transcript( + room: &Room, + context: &mut GeminiEventContext<'_>, + text: &str, +) -> Result<(), Box> { + // `text` is passed whole, not the trimmed form: fragments have to be + // concatenated exactly as received. The trimmed view only decides whether + // this event carried anything at all. + if transcript_text(text).is_some() { + let turn = &mut context.turns.interviewer; + let whole = turn + .record(&mut context.state.transcript, "Interviewer", text) + .to_string(); + publish_transcript(room, &whole, turn.segment_id("interviewer"), false, None).await?; + } + Ok(()) +} + +/// What the candidate said, and the playout it cuts short. +async fn on_input_transcript( + room: &Room, + context: &mut GeminiEventContext<'_>, + text: &str, + interruptible: Interruptible, +) -> Result<(), Box> { + if let (Some(_), Some(identity)) = (transcript_text(text), context.candidate_identity) { + // The candidate is talking over audio Gemini finished producing a while + // ago. Gemini will not call this an interruption, because as far as it + // is concerned that turn ended when it stopped generating; only this + // side knows the queue is still draining. Cut it here or the reply + // lands behind the rest of the old turn. + drop_stale_playout(room, context, interruptible).await?; + context.activity.note_candidate_finished(Instant::now()); + let turn = &mut context.turns.candidate; + let whole = turn + .record(&mut context.state.transcript, CANDIDATE_SPEAKER, text) + .to_string(); + publish_transcript( + room, + &whole, + turn.segment_id("candidate"), + false, + Some(identity), + ) + .await?; + } + Ok(()) +} - // The only other path that empties the queue, and it used to do so - // silently. If a turn is cut this way the candidate hears a - // fragment or nothing, and without this line the log shows only the - // consequence: a turn that completed with nothing left to play. - let unplayed = cut_off_turn(context.activity, context.output_audio); - - // What Gemini heard is the whole diagnosis. It interrupts on its - // own voice activity detection, so a cut with the candidate - // mid-sentence is barge-in working, and a cut with nothing - // transcribed is the microphone hearing the interviewer through the - // candidate's speakers. The line reported the size of the loss and - // left the cause to guesswork across a whole session of them. - let heard = context.turns.candidate.tail(80); +/// A chunk of the interviewer's voice, queued for the room. +async fn on_generated_audio( + room: &Room, + context: &mut GeminiEventContext<'_>, + bytes: &[u8], + mime_type: &str, + interruptible: Interruptible, +) -> Result<(), Box> { + // Read before the interrupt below, because that is the point the candidate + // stops waiting. Stamped on the last input transcript fragment, so it + // covers endpointing plus model latency plus anything still queued ahead of + // the reply: the silence the candidate actually sits through. + // + // `None` means Gemini is answering something it never transcribed, and + // there is no moment the candidate finished to measure from. Printing + // anything then is worse than printing nothing. + let waited = context.activity.awaiting_reply_since; + + // A new turn's first chunk while the previous one is still draining. + // `InputTranscript` normally clears the queue before this point, so + // reaching here means Gemini answered something it never transcribed. + // Backstop rather than the main path, and it must run before `capture` or + // the new audio queues behind the old. + // + // Only for a chunk that will actually be queued. Dropping ahead of a chunk + // `capture` rejects leaves the candidate with a sentence cut in half and no + // reply behind it. + if context.output_audio.accepts(bytes, mime_type) { + drop_stale_playout(room, context, interruptible).await?; + } + if context.output_audio.capture(bytes, mime_type).await? { + // Cleared here rather than where it is read: a chunk `capture` rejects + // is not the reply starting, and consuming the stamp on one would lose + // the measurement for the chunk that is. + if let Some(since) = waited { + context.activity.awaiting_reply_since = None; eprintln!( - "timing: Gemini cut its own turn, {:.1}s of it unplayed; candidate audio so far: {}", - unplayed.as_secs_f64(), - if heard.is_empty() { - "(nothing transcribed)" - } else { - heard - } + "timing: {:.2}s from the candidate finishing to the reply starting", + since.elapsed().as_secs_f64() ); + } + context.activity.mark_speaking(); - // A cut-off turn is still over. Without this the next thing either - // party says appends to the abandoned turn under its segment id, so - // the panel would glue two separate utterances into one row and the - // report prompt would read them as one line. - close_turns(room, context).await?; + // Speech is queued, not played: the floor stays busy until the buffered + // audio actually finishes. + context.activity.last_agent_speech = context.output_audio.playout_deadline; + set_agent_state(room, context.agent_state, AGENT_STATE_SPEAKING).await?; + } + Ok(()) +} - set_agent_state(room, context.agent_state, AGENT_STATE_LISTENING).await?; - } +/// Gemini finished the turn. The room has not: the queue is still draining. +async fn on_turn_complete( + room: &Room, + context: &mut GeminiEventContext<'_>, +) -> Result<(), Box> { + // Whatever the tool response was owed has now arrived. + context.activity.tool_response_outstanding = false; + + // The gap between Gemini finishing and the queue emptying. Gemini + // synthesises far faster than speech plays, so this is how long the agent + // will still be talking after it has stopped thinking, and therefore how + // long a candidate answering now would have waited before + // `drop_stale_playout` existed. + let backlog = context + .output_audio + .playout_deadline + .saturating_duration_since(Instant::now()); - // Named rather than left to the catch-all: the room loop intercepts - // this before dispatching, so the only way one arrives here is through - // `send_wrap_up_and_wait`, where the interview ends within - // `WRAP_UP_WAIT` and there is no socket left to replace. - GeminiEvent::GoAway { .. } => {} - _ => {} + // Only a backlog a candidate would notice. `!is_zero()` fired on a + // millisecond and printed "0.0s", so every one of these lines in a real + // session said nothing at all. + if backlog >= NOTABLE_PLAYOUT_BACKLOG { + eprintln!( + "timing: turn generated, {:.1}s of it still to play", + backlog.as_secs_f64() + ); + } + + // Gemini finishing its turn also means the candidate utterance it answered + // is over, so both sides close here. + close_turns(room, context).await?; + context.activity.floor = Floor::AwaitingPlayout; + if !context.output_audio.is_playing() { + context.activity.mark_listening(); + set_agent_state(room, context.agent_state, AGENT_STATE_LISTENING).await?; + } + Ok(()) +} + +/// Gemini cut its own turn, because it heard the candidate start one. +async fn on_interruption( + room: &Room, + context: &mut GeminiEventContext<'_>, +) -> Result<(), Box> { + // A barge-in cancels a pending close. `cut_off_turn` below clears + // `tool_response_outstanding`, which is the only thing holding the end + // back, so without this the candidate who says "wait, actually" over Jim's + // acknowledgement clears the hold with the same event that carries their + // objection, and is cut off and handed a report. Ending is the one decision + // here nobody can take back, and someone who has just started a sentence is + // not finished. Jim can call the tool again once they are. + if std::mem::take(&mut context.state.end_requested) { + eprintln!("interviewer's ending cancelled: the candidate spoke over the close"); } + + // The only other path that empties the queue, and it used to do so + // silently. If a turn is cut this way the candidate hears a fragment or + // nothing, and without this line the log shows only the consequence: a turn + // that completed with nothing left to play. + let unplayed = cut_off_turn(context.activity, context.output_audio); + + // What Gemini heard is the whole diagnosis. It interrupts on its own voice + // activity detection, so a cut with the candidate mid-sentence is barge-in + // working, and a cut with nothing transcribed is the microphone hearing the + // interviewer through the candidate's speakers. The line reported the size + // of the loss and left the cause to guesswork across a whole session of + // them. + let heard = context.turns.candidate.tail(80); + eprintln!( + "timing: Gemini cut its own turn, {:.1}s of it unplayed; candidate audio so far: {}", + unplayed.as_secs_f64(), + if heard.is_empty() { + "(nothing transcribed)" + } else { + heard + } + ); + + // A cut-off turn is still over. Without this the next thing either party + // says appends to the abandoned turn under its segment id, so the panel + // would glue two separate utterances into one row and the report prompt + // would read them as one line. + close_turns(room, context).await?; + + set_agent_state(room, context.agent_state, AGENT_STATE_LISTENING).await?; Ok(()) } diff --git a/src/web/setup.rs b/src/web/setup.rs index fd3d2011..39b5574d 100644 --- a/src/web/setup.rs +++ b/src/web/setup.rs @@ -1,7 +1,7 @@ //! The solo self-serve cold start's Setup page: served instead of the full //! app when `run_web` finds no config file at all. -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use axum::Json; @@ -212,29 +212,61 @@ const SETUP_PAGE: &str = r#" "#; -/// Writes the config to `config_path`, which `primary_config_path` searches -/// for: the file has to be found again on the next launch, from whatever -/// directory that launch happens to start in. +/// The four values a submission carries, trimmed and checked. +struct SetupFields { + livekit_url: String, + livekit_api_key: String, + livekit_api_secret: String, + google_api_key: String, +} + +impl SetupFields { + /// The keys this submission becomes, in the order the file writes them. + /// + /// One list, hanging off the values it is derived from: what the launch is + /// probed with has to be what gets written, or the answer was about a + /// different config. + /// + /// A blank `googleApiKey` leaves the key out rather than writing it empty. + /// `read_config_file` keeps an empty value, and `load_values` lays file + /// pairs over the environment, so the line would erase a `GOOGLE_API_KEY` + /// the operator had exported and drop the process to web-only -- announced + /// on a console a double-clicked binary does not have. A hand-written + /// config omits the key to defer to the environment, and this writes the + /// same file. + fn pairs(&self) -> Vec<(&str, &str)> { + let mut pairs = vec![ + ("LIVEKIT_URL", self.livekit_url.as_str()), + ("LIVEKIT_API_KEY", self.livekit_api_key.as_str()), + ("LIVEKIT_API_SECRET", self.livekit_api_secret.as_str()), + ]; + if !self.google_api_key.is_empty() { + pairs.push(("GOOGLE_API_KEY", self.google_api_key.as_str())); + } + pairs + } +} + +/// Everything a submission can be refused for before anything is sent or +/// written. /// -/// Parsed as `Value`, not a derived struct: this crate has no `serde` -/// derive dependency, and a missing field reads as empty rather than a -/// parse error. -async fn submit_setup( - Json(submission): Json, - ready: Arc, - production: bool, - path: PathBuf, -) -> Response { - // Trimmed here, once, because everything downstream assumes it was. A - // pasted URL keeps its leading space through `validate_livekit_url`, which - // trims to decide and hands the original on, and `livekit_scheme` then - // matches at offset 0 and rewrites nothing: working credentials come back - // as "did not work". A key with a trailing space is signed into the JWT - // verbatim and refused, though the same value hand-written into the config - // file works, because `read_config_file` trims what it reads. And a - // `googleApiKey` of spaces is not empty to `is_empty`, so it takes the - // branch that probes Gemini and fails there for a field the form calls - // optional. +/// Read out of a `Value`, not a derived struct: this crate has no `serde` +/// derive dependency, and a missing field reads as empty rather than a parse +/// error. +/// +/// One function because the three rules answer one question -- can this file +/// hold this value and be read back as the same value -- and they are only +/// correct together. Trimmed here, once, because everything downstream assumes +/// it was. A pasted URL keeps its leading space through `validate_livekit_url`, +/// which trims to decide and hands the original on, and `livekit_scheme` then +/// matches at offset 0 and rewrites nothing: working credentials come back as +/// "did not work". A key with a trailing space is signed into the JWT verbatim +/// and refused, though the same value hand-written into the config file works, +/// because `read_config_file` trims what it reads. And a `googleApiKey` of +/// spaces is not empty to `is_empty`, so it takes the branch that probes Gemini +/// and fails there for a field the form calls optional. +#[allow(clippy::result_large_err)] // Responses are immediately returned by HTTP handlers. +fn validated_fields(submission: &Value) -> Result { let field = |key: &str| { submission .get(key) @@ -253,10 +285,10 @@ async fn submit_setup( // calls it missing, so accepting one here would write a file the launch // on the other side of this page refuses. if field(key).is_empty() { - return super::json_response( + return Err(super::json_response( StatusCode::BAD_REQUEST, json!({ "error": format!("{key} is required") }), - ); + )); } } @@ -273,10 +305,10 @@ async fn submit_setup( "googleApiKey", ] { if field(key).chars().any(char::is_control) { - return super::json_response( + return Err(super::json_response( StatusCode::BAD_REQUEST, json!({ "error": format!("{key} must not contain control characters") }), - ); + )); } // `read_config_file` does `trim_matches('"')` then @@ -289,7 +321,7 @@ async fn submit_setup( .into_iter() .find(|quote| field(key).starts_with(*quote) || field(key).ends_with(*quote)) { - return super::json_response( + return Err(super::json_response( StatusCode::BAD_REQUEST, json!({ "error": format!( @@ -298,34 +330,25 @@ async fn submit_setup( one checked here" ) }), - ); + )); } } - let livekit_url = field("livekitUrl"); - let livekit_api_key = field("livekitApiKey"); - let livekit_api_secret = field("livekitApiSecret"); - let google_api_key = field("googleApiKey"); - - // The keys this submission becomes, in the order the file below writes - // them. One list: what the launch is asked about has to be what gets - // written, or the answer was about a different config. - // - // A blank `googleApiKey` leaves the key out rather than writing it empty. - // `read_config_file` keeps an empty value, and `load_values` lays file - // pairs over the environment, so the line would erase a `GOOGLE_API_KEY` - // the operator had exported and drop the process to web-only -- announced - // on a console a double-clicked binary does not have. A hand-written config - // omits the key to defer to the environment, and this writes the same file. - let mut pairs = vec![ - ("LIVEKIT_URL", livekit_url.as_str()), - ("LIVEKIT_API_KEY", livekit_api_key.as_str()), - ("LIVEKIT_API_SECRET", livekit_api_secret.as_str()), - ]; - if !google_api_key.is_empty() { - pairs.push(("GOOGLE_API_KEY", google_api_key.as_str())); - } + Ok(SetupFields { + livekit_url: field("livekitUrl"), + livekit_api_key: field("livekitApiKey"), + livekit_api_secret: field("livekitApiSecret"), + google_api_key: field("googleApiKey"), + }) +} +/// Proves the credentials before a file is written from them. +/// +/// `Some` is the refusal. The order is the point: the URL rule is local and +/// instant, the LiveKit probe is one round trip, and the Gemini probe opens a +/// live session, so a submission wrong in the cheapest way is not made to wait +/// for the most expensive check to say so. +async fn probe_credentials(fields: &SetupFields, production: bool) -> Option { // The rule the launch on the other side of this page applies to the URL, // applied while there is still a form to report it in. Without it a URL // this accepts and `web_provider_pool` refuses is written, answered with @@ -340,80 +363,91 @@ async fn submit_setup( // Reported as it stands, naming `LIVEKIT_URL` rather than the form's // `livekitUrl`: the two spellings are the same value, and the one in the // message is what the reader will find in the file afterwards. - if let Err(error) = crate::config::validate_livekit_url(&livekit_url, production) { - return super::json_response(StatusCode::BAD_REQUEST, json!({ "error": error })); + if let Err(error) = crate::config::validate_livekit_url(&fields.livekit_url, production) { + return Some(super::json_response( + StatusCode::BAD_REQUEST, + json!({ "error": error }), + )); } if let Err(error) = crate::livekit::validate_livekit_credentials( - &livekit_url, - &livekit_api_key, - &livekit_api_secret, + &fields.livekit_url, + &fields.livekit_api_key, + &fields.livekit_api_secret, crate::current_epoch_seconds(), ) .await { - return super::json_response( + return Some(super::json_response( StatusCode::BAD_REQUEST, json!({ "error": format!( "livekitUrl, livekitApiKey or livekitApiSecret did not work: {error}" ) }), - ); + )); } - if !google_api_key.is_empty() { - let config = match load_from_pairs(pairs.iter().copied()) { - Ok(config) => config, - Err(error) => { - return super::json_response( - StatusCode::BAD_REQUEST, - json!({ "error": error.to_string() }), - ); - } - }; - - // Same proof as `check-gemini`: open a real session. - // `CODETRIAL_GEMINI_LIVE_URL` lets tests redirect this away from the - // real endpoint. - let room_name = format!("{}-smoke", config.room_prefix); - let boot = bootstrap(&config, &room_name, None, config.default_duration_min); - let url = std::env::var("CODETRIAL_GEMINI_LIVE_URL") - .unwrap_or_else(|_| gemini_live_websocket_url(&config.google_api_key)); - match open_live_session_at(&url, &boot, None).await { - Ok(session) => { - let _ = session.close().await; - } - Err(error) => { - // Redacted like every other caller of this chain. The close - // frame's reason is folded in at the bottom of it, so the text - // here is partly Gemini's, and the key was sent in the URL. - let reason = crate::gemini::redact_api_key( - &format!("googleApiKey did not work: {error}"), - &google_api_key, - ); - return super::json_response(StatusCode::BAD_REQUEST, json!({ "error": reason })); - } - } + if fields.google_api_key.is_empty() { + return None; } + let config = match load_from_pairs(fields.pairs().iter().copied()) { + Ok(config) => config, + Err(error) => { + return Some(super::json_response( + StatusCode::BAD_REQUEST, + json!({ "error": error.to_string() }), + )); + } + }; - // From `pairs`, so the file holds exactly the config that was checked and - // probed above. Every value is known to carry no newline by now, which is - // what makes one line per key a faithful encoding of it. - let contents = pairs - .iter() - .map(|(key, value)| format!("{key}={value}\n")) - .collect::(); + // Same proof as `check-gemini`: open a real session. + // `CODETRIAL_GEMINI_LIVE_URL` lets tests redirect this away from the real + // endpoint. + let room_name = format!("{}-smoke", config.room_prefix); + let boot = bootstrap(&config, &room_name, None, config.default_duration_min); + let url = std::env::var("CODETRIAL_GEMINI_LIVE_URL") + .unwrap_or_else(|_| gemini_live_websocket_url(&config.google_api_key)); + match open_live_session_at(&url, &boot, None).await { + Ok(session) => { + let _ = session.close().await; + None + } + Err(error) => { + // Redacted like every other caller of this chain. The close frame's + // reason is folded in at the bottom of it, so the text here is + // partly Gemini's, and the key was sent in the URL. + let reason = crate::gemini::redact_api_key( + &format!("googleApiKey did not work: {error}"), + &fields.google_api_key, + ); + Some(super::json_response( + StatusCode::BAD_REQUEST, + json!({ "error": reason }), + )) + } + } +} +/// Writes the config file, and refuses rather than overwriting one. +/// +/// The path is `config_path`, which `primary_config_path` searches for: the +/// file has to be found again on the next launch, from whatever directory that +/// launch happens to start in. +/// +/// `Some` is the refusal. The mode and the `create_new` are the whole reason +/// this is not two lines: the file holds `LIVEKIT_API_SECRET` and +/// `GOOGLE_API_KEY`, and both defaults are wrong for it. +fn write_config_file(path: &Path, contents: &str) -> Option { // The directory is the caller's answer, and a released binary's copy of it // does not exist until the first submission. if let Some(parent) = path.parent() && let Err(error) = std::fs::create_dir_all(parent) { - return super::json_response( + return Some(super::json_response( StatusCode::INTERNAL_SERVER_ERROR, json!({ "error": format!("could not create {}: {error}", parent.display()) }), - ); + )); } // `create_new`, so an existing name is reported rather than followed and @@ -432,21 +466,52 @@ async fn submit_setup( options.mode(0o600); } let written = options - .open(&path) + .open(path) .and_then(|mut file| std::io::Write::write_all(&mut file, contents.as_bytes())); - if let Err(error) = written { - let reason = if error.kind() == std::io::ErrorKind::AlreadyExists { - format!( - "{} already exists; move it aside and reload", - path.display() - ) - } else { - format!("could not write {}: {error}", path.display()) - }; - return super::json_response( - StatusCode::INTERNAL_SERVER_ERROR, - json!({ "error": reason }), - ); + let Err(error) = written else { + return None; + }; + let reason = if error.kind() == std::io::ErrorKind::AlreadyExists { + format!( + "{} already exists; move it aside and reload", + path.display() + ) + } else { + format!("could not write {}: {error}", path.display()) + }; + Some(super::json_response( + StatusCode::INTERNAL_SERVER_ERROR, + json!({ "error": reason }), + )) +} + +/// One submission: checked, probed, written, and the app let out behind it. +async fn submit_setup( + Json(submission): Json, + ready: Arc, + production: bool, + path: PathBuf, +) -> Response { + let fields = match validated_fields(&submission) { + Ok(fields) => fields, + Err(response) => return response, + }; + + if let Some(refusal) = probe_credentials(&fields, production).await { + return refusal; + } + + // From `pairs`, so the file holds exactly the config that was checked and + // probed above. Every value is known to carry no newline by now, which is + // what makes one line per key a faithful encoding of it. + let contents = fields + .pairs() + .iter() + .map(|(key, value)| format!("{key}={value}\n")) + .collect::(); + + if let Some(refusal) = write_config_file(&path, &contents) { + return refusal; } // Only reached once the file is on disk; tells the caller to stop serving diff --git a/src/web/token.rs b/src/web/token.rs index e7f75bf5..09b7db08 100644 --- a/src/web/token.rs +++ b/src/web/token.rs @@ -14,12 +14,12 @@ use axum::response::{IntoResponse, Response}; use ring::rand::SecureRandom; use serde_json::{Value, json}; -use crate::accounts::{blocking, release_interview_room}; +use crate::accounts::{Accounts, SignedInUser, blocking, release_interview_room}; use crate::config::{DEFAULT_DURATION_MIN, MAX_DURATION_MIN, MIN_DURATION_MIN}; use crate::current_epoch_seconds; use crate::token::{LivekitTokenInput, TOKEN_TTL_SECONDS, livekit_observer_token, livekit_token}; -use super::auth::current_user; +use super::auth::{Owner, current_user}; use super::consent::{checked_consent, claim_consent}; use super::pool::{ProviderChoice, room_and_available_provider}; use super::{ @@ -308,34 +308,77 @@ pub(crate) fn token_duration_min(value: Option<&Value>, recording_max_min: Optio } } -pub(crate) async fn token_handler( - State(state): State, - ConnectInfo(peer): ConnectInfo, - request: Request, -) -> Response { - // Ahead of the rate limit on purpose: the bucket is keyed by address, so - // letting anonymous callers spend it would let them lock out the signed-in - // candidate sharing their NAT. A refused request mints nothing. - // +/// A room on a provider that still has minutes, or the refusal to send back. +/// +/// The two failures are one function because they are one question with two +/// answers a candidate cannot tell apart: no provider is a server the operator +/// never finished setting up, and every provider spent is a server that worked +/// until this morning. +#[allow(clippy::result_large_err)] // Responses are immediately returned by HTTP handlers. +async fn reserved_room(state: &AppState) -> Result<(String, &crate::config::Provider), Response> { + match room_and_available_provider(state).await { + ProviderChoice::Ready(room_name, provider) => Ok((room_name, provider)), + ProviderChoice::NoneConfigured => Err(json_response( + StatusCode::INTERNAL_SERVER_ERROR, + json!({ + "error": "Server is missing LiveKit credentials. Create config/codetrial.env.local or set LIVEKIT_URL, LIVEKIT_API_KEY and LIVEKIT_API_SECRET." + }), + )), + + // Naming the cause, because the browser cannot. A refused upgrade + // reaches the page as a bare socket error with no status attached, so a + // candidate told only "could not connect" would go looking at their own + // network for a quota this server already knows is spent. + ProviderChoice::AllExhausted => Err(json_response( + StatusCode::SERVICE_UNAVAILABLE, + json!({ + "code": "livekit_quota_exhausted", + "error": "Every configured LiveKit project is out of connection minutes. Ask the operator to top one up." + }), + )), + } +} + +/// Every gate that can refuse a request before any external work happens. +/// +/// One function rather than four checks spread down the handler, because the +/// order between them is the property worth protecting and it is invisible +/// when they are apart. The account gate runs ahead of the rate limit because +/// the bucket is keyed by address, so an anonymous caller allowed to spend it +/// could lock out the signed-in candidate sharing their NAT. The +/// verified-email gate runs ahead of the rate limit too, so a caller who can +/// never receive a recording is told why rather than being spent against a +/// budget they share. +/// +/// It does not cover the room reservation: `reserved_room` runs after this +/// returns and advances the provider rotation, so an unverified caller refused +/// here has cost nothing only because the refusal comes first at the call +/// site. Nothing but that ordering enforces it. +#[allow(clippy::result_large_err)] // Responses are immediately returned by HTTP handlers. +async fn admit_caller( + state: &AppState, + headers: &axum::http::HeaderMap, + peer: SocketAddr, +) -> Result { // Where accounts exist, an interview belongs to one. Hiding the start // button would not stop anyone opening /interview directly, so the gate // lives on the credential. let Some(accounts) = state.accounts.clone() else { - return state.accounts_error(); + return Err(state.accounts_error()); }; - let user = match current_user(state.accounts.as_ref(), request.headers()).await { + let user = match current_user(state.accounts.as_ref(), headers).await { Ok(Some(user)) => user, Ok(None) => { - return json_response( + return Err(json_response( StatusCode::UNAUTHORIZED, json!({ "error": "Enter your GitHub username to start an interview." }), - ); + )); } Err(_) => { - return json_response( + return Err(json_response( StatusCode::INTERNAL_SERVER_ERROR, json!({ "error": "Could not read account session." }), - ); + )); } }; @@ -345,44 +388,122 @@ pub(crate) async fn token_handler( // who cannot start and a candidate whose finished interview cannot be sent // anywhere. if state.config.recording.is_some() && user.verified_email.is_none() { - return json_response( + return Err(json_response( StatusCode::FORBIDDEN, json!({ "code": "recording_requires_verified_identity", "error": "This interview is recorded, so it needs a GitHub sign-in. A typed username cannot receive the recording." }), - ); + )); } - let client = client_ip(request.headers(), peer, state.config.trusted_proxy_hops); + let client = client_ip(headers, peer, state.config.trusted_proxy_hops); if !state.token_limit.allow(client, Instant::now()) { - return rate_limited_response("Too many session requests. Wait a minute and try again."); + return Err(rate_limited_response( + "Too many session requests. Wait a minute and try again.", + )); } - // One record, so the three values cannot come from different providers. - let (room_name, provider) = match room_and_available_provider(&state).await { - ProviderChoice::Ready(room_name, provider) => (room_name, provider), - ProviderChoice::NoneConfigured => { - return json_response( - StatusCode::INTERNAL_SERVER_ERROR, - json!({ - "error": "Server is missing LiveKit credentials. Create config/codetrial.env.local or set LIVEKIT_URL, LIVEKIT_API_KEY and LIVEKIT_API_SECRET." - }), - ); - } + Ok(Owner { accounts, user }) +} - // Naming the cause, because the browser cannot. A refused upgrade - // reaches the page as a bare socket error with no status attached, so a - // candidate told only "could not connect" would go looking at their own - // network for a quota this server already knows is spent. - ProviderChoice::AllExhausted => { - return json_response( - StatusCode::SERVICE_UNAVAILABLE, - json!({ - "code": "livekit_quota_exhausted", - "error": "Every configured LiveKit project is out of connection minutes. Ask the operator to top one up." - }), - ); +/// Starts the interviewer, and gives the consent back when there is no room. +/// +/// `Some` is the refusal to send. The release is the reason this is not two +/// lines at the call site: a candidate told to retry, whose first attempt +/// spent their consent on a dispatch that never happened, is refused a second +/// time for a reason they cannot see or fix. +async fn dispatch_or_release_consent( + state: &AppState, + accounts: &Arc, + interview: Option<&String>, + owner: &SignedInUser, + room_name: &str, + provider: &crate::config::Provider, +) -> Option { + let Some(dispatcher) = &state.dispatcher else { + return None; + }; + if dispatcher.ensure_agent(room_name, provider) { + return None; + } + if let Some(interview) = interview { + let accounts = accounts.clone(); + let interview = interview.clone(); + let room_name = room_name.to_string(); + let owner = owner.id; + if let Err(error) = + blocking(move || release_interview_room(&accounts, &interview, owner, &room_name)).await + { + eprintln!("could not release interview consent after a refused dispatch: {error}"); } + } + + // Refusing is the honest failure. Handing out the token anyway would put + // the candidate in an empty room reading "Waiting" with nothing, on screen + // or in any log they can see, saying why. + Some(json_response( + StatusCode::SERVICE_UNAVAILABLE, + json!({ + "error": "The server is running as many interviews as it can right now. Try again in a few minutes." + }), + )) +} + +/// The signed token, or the refusal to send instead. +/// +/// Split out for the reason the gates above it were: what remains at the call +/// site is the sequence, and the two ways signing can fail are a pair worth +/// reading together rather than thirty lines wedged between the consent check +/// and the claim. +#[allow(clippy::result_large_err)] // Responses are immediately returned by HTTP handlers. +fn minted_token( + state: &AppState, + provider: &crate::config::Provider, + body: &[u8], + room_name: &str, + identity: &str, +) -> Result { + match token_response( + &TokenConfig { + api_key: &provider.api_key, + api_secret: &provider.api_secret, + server_url: &provider.url, + recording_max_min: state.config.recording.as_ref().map(|it| it.max_minutes), + }, + body, + room_name, + identity, + current_epoch_seconds(), + ) { + Ok(response) => Ok(response), + + // No arm for a parse failure. `token_response` parses only a body that + // is not empty, and `token_handler` has already refused a non-empty + // body that will not parse, with the same call and the same 400. An arm + // here could only answer a request that cannot reach it, which is why + // both of its mutants survived the gate: nothing distinguishes a guard + // that is never true from one that is never false. + Err(error) => Err(json_response( + StatusCode::INTERNAL_SERVER_ERROR, + json!({ "error": error.to_string() }), + )), + } +} + +pub(crate) async fn token_handler( + State(state): State, + ConnectInfo(peer): ConnectInfo, + request: Request, +) -> Response { + let Owner { accounts, user } = match admit_caller(&state, request.headers(), peer).await { + Ok(admitted) => admitted, + Err(response) => return response, + }; + + // One record, so the three values cannot come from different providers. + let (room_name, provider) = match reserved_room(&state).await { + Ok(reserved) => reserved, + Err(response) => return response, }; let Ok(body) = to_bytes(request.into_body(), MAX_BODY_BYTES).await else { @@ -407,35 +528,9 @@ pub(crate) async fn token_handler( Err(response) => return response, }; let identity = generated_candidate_identity(); - let response = match token_response( - &TokenConfig { - api_key: &provider.api_key, - api_secret: &provider.api_secret, - server_url: &provider.url, - recording_max_min: state.config.recording.as_ref().map(|it| it.max_minutes), - }, - body.as_ref(), - &room_name, - &identity, - current_epoch_seconds(), - ) { + let response = match minted_token(&state, provider, body.as_ref(), &room_name, &identity) { Ok(response) => response, - - // A body that would not parse is the caller's fault, not ours. The - // status comes from the one parse inside `token_response` rather than - // from a second copy of the same check out here. - Err(error) if error.is::() => { - return json_response( - StatusCode::BAD_REQUEST, - json!({ "error": "Session request must be JSON." }), - ); - } - Err(error) => { - return json_response( - StatusCode::INTERNAL_SERVER_ERROR, - json!({ "error": error.to_string() }), - ); - } + Err(refusal) => return refusal, }; // Before the dispatcher, so two requests racing on one interview cannot @@ -449,37 +544,19 @@ pub(crate) async fn token_handler( return response; } - // Validate the request before starting anything external. Otherwise a - // malformed body can leave an interviewer running for a request that got a - // 400 response. - if let Some(dispatcher) = &state.dispatcher - && !dispatcher.ensure_agent(&room_name, provider) + // After the token and the claim, so a request that was going to be refused + // anyway never leaves an interviewer running behind it. + if let Some(refusal) = dispatch_or_release_consent( + &state, + &accounts, + interview.as_ref(), + &user, + &room_name, + provider, + ) + .await { - // The claim above is given back first. A busy server tells the - // candidate to retry, and a retry refused because the first attempt - // spent their consent is worse than the refusal it followed. - if let Some(interview) = &interview { - let accounts = accounts.clone(); - let interview = interview.clone(); - let room_name = room_name.clone(); - let owner = user.id; - if let Err(error) = - blocking(move || release_interview_room(&accounts, &interview, owner, &room_name)) - .await - { - eprintln!("could not release interview consent after a refused dispatch: {error}"); - } - } - - // Refusing is the honest failure. Handing out the token anyway would - // put the candidate in an empty room reading "Waiting" with nothing, on - // screen or in any log they can see, saying why. - return json_response( - StatusCode::SERVICE_UNAVAILABLE, - json!({ - "error": "The server is running as many interviews as it can right now. Try again in a few minutes." - }), - ); + return refusal; } state diff --git a/tests/browser/avatar.test.js b/tests/browser/avatar.test.js index 499a7481..96c35d21 100644 --- a/tests/browser/avatar.test.js +++ b/tests/browser/avatar.test.js @@ -576,15 +576,24 @@ test("avatar analyser reads Jim and never the candidate", () => { // And the teardown is gated on the analyser's own track, or a second // participant leaving killed lip sync for the rest of the session. assert.match(script, /if \(!isAvatarAnalyserTrack\(track\)\) return;/); - // Never the microphone, checked across the whole file rather than inside one + // Never the microphone, checked across the whole path rather than inside one // function slice: wiring the candidate's stream in from anywhere else would // have passed a slice-scoped assertion while the avatar watched the candidate. // An allowlist, not a count. The preflight legitimately builds a source from // the candidate's own microphone to drive the level meter, so the property // worth pinning is that the set does not grow: exactly two call sites, and // the avatar's is the remote track. - const sources = captures(script, /createMediaStreamSource\(([^)]*)\)/g); - assert.deepEqual(sources, ["stream", "new MediaStream([track.mediaStreamTrack]"], + // + // `mic-meter.js` is read in beside the path rather than added to + // `INTERVIEW_SOURCES`, which is the modules `interview.js` hands its bindings + // to and is pinned to exactly those. The meter is where the microphone's own + // source lives, so leaving it out would let this allowlist grow unwatched in + // the one file most likely to grow it. + const sources = captures( + `${script}\n${read("web/mic-meter.js")}`, + /createMediaStreamSource\(([^)]*)\)/g, + ); + assert.deepEqual(sources, ["new MediaStream([track.mediaStreamTrack]", "stream"], "a new createMediaStreamSource call site appeared; the avatar must only ever read Jim"); assert.doesNotMatch(script, /createMediaStreamSource\([^)]*localUserStream/); assert.ok(ANALYSER_WINDOW > 1, "a single frame of amplitude flickers the jaw"); diff --git a/tests/browser/dom-contract.test.js b/tests/browser/dom-contract.test.js index 54e69132..a9feb0c4 100644 --- a/tests/browser/dom-contract.test.js +++ b/tests/browser/dom-contract.test.js @@ -299,15 +299,15 @@ test("a replacement preflight camera gets a fresh face check", () => { // which has no bypass, stayed shut for the rest of the preflight. test("a preflight microphone that goes away is asked for again", () => { const script = interviewSource(); - // One loop covers both kinds, so the microphone is no longer the case that - // can be forgotten: naming a kind here is what asks the pool to look again. - assert.match(script, - /for \(const kind of \["audio", "video"\]\)[\s\S]*?pool\.dropTrack\(track\);\s*pool\.retry\(\);/, - "a dead device of either kind must reopen the request the pool makes"); + // The drop itself moved into `preflightReadiness` in `web/audio-check.js`, + // where `tests/browser/mic-meter.test.js` drives it against a pool holding a + // dead track. What is left here is the wiring only this file can show. // Anchored to the audio hook and lazily matched, so a rename of the video - // hook cannot silently widen the slice this is read out of. + // hook cannot silently widen the slice this is read out of. What `forget` + // has to do is not asserted here: it lives in `web/mic-meter.js` now, and + // `tests/browser/mic-meter.test.js` drives it rather than reading it. assert.match(script, - /pool\.configure\("audio",[\s\S]*?onLost: \(\) => \{\s*meterGeneration \+= 1;\s*micPeak = 0;\s*recentPeaks\.length = 0;/, + /pool\.configure\("audio",[\s\S]*?onLost: \(\) => \{\s*meter\.forget\(\);/, "dropping a microphone must invalidate its meter and readiness state"); }); diff --git a/tests/browser/mic-meter.test.js b/tests/browser/mic-meter.test.js new file mode 100644 index 00000000..3b5ecf86 --- /dev/null +++ b/tests/browser/mic-meter.test.js @@ -0,0 +1,235 @@ +// Run with: node --test tests/browser/mic-meter.test.js +// +// The media gate has no bypass, so what it reads has to be about the devices +// that are still there. These drive `createMicMeter` and `preflightReadiness` +// against a fake meter and a fake pool, so what is asserted is behaviour -- +// which peaks are latched, which are forgotten, and whether the gate opens -- +// rather than the text of the functions. + +import { test } from "node:test"; +import assert from "node:assert/strict"; + +import { MIC_CONFIRM_FRAMES, preflightReadiness } from "../../web/audio-check.js"; +import { createMicMeter } from "../../web/mic-meter.js"; + +/// A pool with one audio track, whose readiness a test can change. +function fakePool(audioState = "live") { + const errors = {}; + return { + stream: {}, + trackOf: (kind) => (kind === "audio" ? { readyState: audioState } : null), + setError(kind, error) { + errors[kind] = error; + }, + errorOf: (kind) => errors[kind] ?? null, + dropped: [], + dropTrack(track) { + this.dropped.push(track); + }, + retry() { + this.retried = true; + }, + }; +} + +/// A meter that hands its callbacks back instead of opening an AudioContext. +function fakeMeter() { + const started = []; + const start = (stream, onPeak, onError, shouldStop) => { + started.push({ onPeak, onError, shouldStop }); + }; + return { start, started, latest: () => started.at(-1) }; +} + +const loudEnough = (meter, level = 0.5) => { + for (let i = 0; i < MIC_CONFIRM_FRAMES; i += 1) meter.latest().onPeak(level); +}; + +test("a proven peak is latched, so a candidate need not keep talking", () => { + const started = fakeMeter(); + const meter = createMicMeter({ + pool: fakePool(), + isFinished: () => false, + onLevel: () => {}, + onFailure: () => {}, + startMeter: started.start, + }); + + meter.start(); + loudEnough(started, 0.5); + assert.ok(meter.peak() > 0, "six loud frames prove a microphone"); + + for (let i = 0; i < MIC_CONFIRM_FRAMES; i += 1) started.latest().onPeak(0); + assert.ok(meter.peak() > 0, "going quiet must not un-prove it"); +}); + +test("forgetting a microphone drops what it proved and silences its meter", () => { + const started = fakeMeter(); + const meter = createMicMeter({ + pool: fakePool(), + isFinished: () => false, + onLevel: () => {}, + onFailure: () => {}, + startMeter: started.start, + }); + + meter.start(); + loudEnough(started); + const abandoned = started.latest(); + assert.ok(meter.peak() > 0); + + meter.forget(); + + assert.equal(meter.peak(), 0, "nothing a departed microphone proved carries over"); + // The generation bump, observed rather than counted: the meter still running + // over the old track must not report another level. + assert.equal(abandoned.shouldStop(), true, "the meter over the old track must stop"); +}); + +test("a replacement microphone is proven on its own frames", () => { + const started = fakeMeter(); + const meter = createMicMeter({ + pool: fakePool(), + isFinished: () => false, + onLevel: () => {}, + onFailure: () => {}, + startMeter: started.start, + }); + + meter.start(); + loudEnough(started); + meter.forget(); + meter.start(); + + assert.equal(meter.peak(), 0, "the new device has proven nothing yet"); + loudEnough(started); + assert.ok(meter.peak() > 0, "and proves itself on its own frames"); + assert.equal(started.latest().shouldStop(), false, "the current meter keeps running"); +}); + +/// Timers a test can run by hand. +/// +/// The retry is the whole point of these two, and a real `setTimeout` would +/// make them wait to find out whether one was even scheduled. +function fakeTimers(t) { + const pending = []; + const savedSet = globalThis.setTimeout; + const savedClear = globalThis.clearTimeout; + globalThis.setTimeout = (fn) => pending.push(fn); + globalThis.clearTimeout = (handle) => { + if (handle) pending[handle - 1] = null; + }; + t.after(() => { + globalThis.setTimeout = savedSet; + globalThis.clearTimeout = savedClear; + }); + return { + scheduled: () => pending.filter(Boolean).length, + run: () => { + for (const fn of pending) if (fn) fn(); + }, + }; +} + +test("a meter that fails forgets the peak and asks again", (t) => { + const timers = fakeTimers(t); + const started = fakeMeter(); + const failures = []; + const meter = createMicMeter({ + pool: fakePool(), + isFinished: () => false, + onLevel: () => {}, + onFailure: (error) => failures.push(error), + startMeter: started.start, + }); + + meter.start(); + loudEnough(started); + started.latest().onError("device lost"); + + assert.equal(meter.peak(), 0, "a device that went away has proven nothing about its replacement"); + assert.deepEqual(failures, ["device lost"]); + assert.equal(timers.scheduled(), 1, "the gate has no bypass, so the meter must ask again"); + timers.run(); + assert.equal(started.started.length, 2, "and asking again starts a meter"); +}); + +/// The preflight's own `onFailure` repaints, which drops the ended track, +/// which forgets this meter -- all before the error callback returns. A retry +/// armed after that runs over a stream whose track the pool has already let go. +test("a meter forgotten while it was failing does not restart itself", (t) => { + const timers = fakeTimers(t); + const started = fakeMeter(); + let meter; + meter = createMicMeter({ + pool: fakePool(), + isFinished: () => false, + onLevel: () => {}, + onFailure: () => meter.forget(), + startMeter: started.start, + }); + + meter.start(); + started.latest().onError("device lost"); + timers.run(); + + assert.equal(started.started.length, 1, "a forgotten meter must not restart itself"); +}); + +/// The bug this exists for: a candidate proves their microphone, unplugs it, +/// and clicks Start. The drop is what invalidates the peak, so a sample that +/// read the peak first would open the gate on a device that is gone. +test("an ended microphone closes the gate even when a peak was proven", () => { + const pool = fakePool("ended"); + const started = fakeMeter(); + const meter = createMicMeter({ + pool, + isFinished: () => false, + onLevel: () => {}, + onFailure: () => {}, + startMeter: started.start, + }); + meter.start(); + loudEnough(started); + assert.ok(meter.peak() > 0, "the microphone was proven before it went away"); + + // `dropTrack` is what fires `onLost` in the real pool, which is what calls + // `forget`. Wiring that here is the whole point: the sample has to drop the + // dead track before it reads the peak, or it reads a peak the drop clears. + pool.dropTrack = (track) => { + pool.dropped.push(track); + meter.forget(); + }; + + const state = preflightReadiness({ + pool, + browserSupported: true, + outputConfirmed: true, + micPeak: meter.peak, + faceCheck: { ready: true, error: null }, + }); + + assert.equal(state.steps.mic, false, "an unplugged microphone is not a proven one"); + assert.equal(state.ready, false, "and the gate does not open on it"); + assert.ok(pool.retried, "the pool is asked for a replacement"); +}); + +/// Both kinds, one rule. An ended track never revives, and while it sits in +/// the stream the pool's retry sees a device of that kind and asks for +/// nothing, so the gate -- which has no bypass -- would stay shut for the rest +/// of the preflight. +test("a dead device of either kind reopens the request the pool makes", () => { + for (const dead of ["audio", "video"]) { + const pool = fakePool(); + pool.trackOf = (kind) => ({ readyState: kind === dead ? "ended" : "live" }); + preflightReadiness({ + pool, + browserSupported: true, + outputConfirmed: true, + micPeak: () => 1, + faceCheck: { ready: true, error: null }, + }); + assert.equal(pool.dropped.length, 1, `an ended ${dead} track must be dropped`); + assert.ok(pool.retried, `and the pool asked for a replacement ${dead} device`); + } +}); diff --git a/web/audio-check.js b/web/audio-check.js index 9bd3f9d5..1df21d88 100644 --- a/web/audio-check.js +++ b/web/audio-check.js @@ -135,3 +135,58 @@ export function videoTrackReady(track) { export function outputUsable(audioContextState) { return audioContextState === "running"; } + +/// Every signal the media gate reads, in one call. +/// +/// A function rather than eight reads at each call site, so a new signal is +/// added here and both callers get it. The camera's error is the one thing +/// written on the way through: it is the only signal that can only be judged +/// against a track the candidate already granted. +/// +/// `micPeak` arrives as a function rather than a value because the drop below +/// changes it: dropping an ended microphone fires the meter's `onLost`, which +/// forgets the peak it had proven. A peak read at the call site is read before +/// that, so an unplugged microphone would answer this sample with the level +/// the device that went away once reached -- and the sample that matters most +/// is the one `finish` takes when the candidate clicks Start. +export function preflightReadiness({ + pool, + browserSupported, + outputConfirmed, + micPeak, + faceCheck, +}) { + // Keyed on the track, not on the pool's stream. The stream exists from the + // moment the preflight asks for a device, so testing it here would overwrite + // the reason the request actually failed -- the permission the candidate + // denied -- with "no active video track" on every frame, and paint the panel + // red while the prompt is still on screen. + const camera = pool.trackOf("video"); + if (camera) pool.setError("video", videoTrackReady(camera) ? null : "no active video track"); + // An ended track never revives, and while it sits in the stream the retry + // sees a device of that kind and asks for nothing. The gate has no bypass, + // so an unplugged device would strand the candidate. A muted track can come + // back on its own, so only the ended one is dropped. + // + // Both kinds, one rule. The camera is dropped here rather than left to the + // retry tick because its liveness is judged per frame just above; the + // microphone has no such judgement, because a meter over a device that went + // away reports silence rather than an error. `dropTrack` fires `onLost`, so + // whatever was running over the track is torn down with it. + for (const kind of ["audio", "video"]) { + const track = pool.trackOf(kind); + if (track?.readyState !== "ended") continue; + pool.dropTrack(track); + pool.retry(); + } + return mediaReadiness({ + browserSupported, + outputConfirmed, + micPeak: micPeak(), + micError: pool.errorOf("audio"), + cameraReady: videoTrackReady(pool.trackOf("video")), + cameraError: pool.errorOf("video"), + faceReady: faceCheck.ready, + faceError: faceCheck.error, + }); +} diff --git a/web/interview.js b/web/interview.js index 3143903f..31edf1ab 100644 --- a/web/interview.js +++ b/web/interview.js @@ -1,16 +1,14 @@ import { loadJudge, loadProblem } from "./problem-data.js"; import { - MIC_CONFIRM_FRAMES, - mediaReadiness, outputUsable, - peakLevel, + preflightReadiness, videoTrackReady, - sustainedPeak, } from "./audio-check.js"; import { highlight } from "./highlight.js"; import { indentSelection } from "./editor.js"; import { createDevicePool } from "./devices.js"; import { createFaceCheck } from "./face-check.js"; +import { createMicMeter, startMediaMeter } from "./mic-meter.js"; import { createTranscriptView, finalRunnerStatus, @@ -497,6 +495,24 @@ async function signInRequired() { return true; } +/// The preflight's readiness, on the page. +/// +/// Out of `refresh` because none of it is a decision: every line here is one +/// signal of `mediaReadiness` written to the one node that shows it, and the +/// decisions that surround it -- which hint outranks the status, when focus +/// moves on -- were hard to find among nine assignments that never branch. +function paintPreflight(state, hint) { + nodes.audioStatus.textContent = hint || state.message; + nodes.audioOutputState.textContent = state.steps.output ? "Confirmed" : "play a short tone."; + nodes.cameraState.textContent = state.steps.camera ? "Ready" : "grant access and keep video on."; + nodes.audioStatus.classList.toggle("critical", ["mic-error", "camera-error", "face-error", "browser"].includes(state.blocker)); + nodes.audioStepOutput.classList.toggle("done", state.steps.output); + nodes.audioStepMic.classList.toggle("done", state.steps.mic); + nodes.audioStepCamera.classList.toggle("done", state.steps.camera); + nodes.audioHeard.disabled = state.steps.output; + nodes.audioHeard.textContent = state.steps.output ? "Confirmed" : "I heard it"; +} + /// Resolves once the candidate has proven output, microphone, and camera. /// The tone doubles as the user gesture browsers require before /// any audio plays, so confirming it also unblocks the interviewer's voice. @@ -505,7 +521,6 @@ function runAudioCheck() { const mediaDevices = navigator.mediaDevices; const browserSupported = Boolean(mediaDevices?.getUserMedia); let outputConfirmed = false; - let meterRetry = null; let advanced = false; // Built before anything reads it: `refresh` runs on the first frame and @@ -515,26 +530,40 @@ function runAudioCheck() { isFinished: () => finished, onChange: () => refresh(), }); - const trackOf = (kind) => pool.trackOf(kind); // Derived, not stored. This was a `let` written inside `sampleReadiness` // and read from the face check, so whether the camera worked depended on // who had run most recently. - const cameraReady = () => videoTrackReady(trackOf("video")); - let micPeak = 0; + const cameraReady = () => videoTrackReady(pool.trackOf("video")); const faceCheck = createFaceCheck({ video: nodes.cameraIntegrityVideo, - trackOf: () => trackOf("video"), + trackOf: () => pool.trackOf("video"), isReady: cameraReady, isFinished: () => finished, createDetector: createFacePresenceDetector, verdictOf: facePresenceVerdict, onVerdict: () => refresh(), }); + const meter = createMicMeter({ + pool, + isFinished: () => finished, + startMeter: startMediaMeter, + // The bar is the page's, not the meter's: the module is kept free of + // nodes so it can run against a fake meter in a test. + onLevel: (level) => { + nodes.audioMeterFill.style.width = `${Math.min(100, Math.round(level * 300))}%`; + refresh(); + }, + onFailure: () => { + // The hint outranks the status line on every frame, so a stale one + // would leave the bar red and still reading "play the test tone". + hint = null; + refresh(); + }, + }); let context = null; let hint = null; let hintUntil = 0; let finished = false; - const recentPeaks = []; // Set once, because refresh() runs on every animation frame: the // candidate is the only reliable output sensor, so confirming is // accepted whenever they click rather than gated on tone timing. @@ -549,58 +578,19 @@ function runAudioCheck() { refresh(); }; - // Reads all eight signals in one call, so a new signal is added here and - // both callers get it. The camera's error is the one thing still written on - // the way through: it is the only signal that can only be judged against a - // track the candidate already granted. - const sampleReadiness = () => { - // Keyed on the track, not on the pool's stream. The stream exists from - // the moment the preflight asks for a device, so testing it here would - // overwrite the reason the request actually failed -- the permission the - // candidate denied -- with "no active video track" on every frame, and - // paint the panel red while the prompt is still on screen. - const camera = trackOf("video"); - if (camera) pool.setError("video", cameraReady() ? null : "no active video track"); - // An ended track never revives, and while it sits in the stream the retry - // sees a device of that kind and asks for nothing. The gate has no bypass, - // so an unplugged device would strand the candidate. A muted track can - // come back on its own, so only the ended one is dropped. - // - // Both kinds, one rule. The camera is dropped here rather than left to the - // retry tick because its liveness is judged per frame just above; the - // microphone has no such judgement, because a meter over a device that - // went away reports silence rather than an error. `dropTrack` fires - // `onLost`, so whatever was running over the track is torn down with it. - for (const kind of ["audio", "video"]) { - const track = trackOf(kind); - if (track?.readyState !== "ended") continue; - pool.dropTrack(track); - pool.retry(); - } - return mediaReadiness({ + const sampleReadiness = () => + preflightReadiness({ + pool, browserSupported, outputConfirmed, - micPeak, - micError: pool.errorOf("audio"), - cameraReady: cameraReady(), - cameraError: pool.errorOf("video"), - faceReady: faceCheck.ready, - faceError: faceCheck.error, + micPeak: meter.peak, + faceCheck, }); - }; const refresh = () => { const state = sampleReadiness(); if (hint && Date.now() >= hintUntil) hint = null; - nodes.audioStatus.textContent = hint || state.message; - nodes.audioOutputState.textContent = state.steps.output ? "Confirmed" : "play a short tone."; - nodes.cameraState.textContent = state.steps.camera ? "Ready" : "grant access and keep video on."; - nodes.audioStatus.classList.toggle("critical", ["mic-error", "camera-error", "face-error", "browser"].includes(state.blocker)); - nodes.audioStepOutput.classList.toggle("done", state.steps.output); - nodes.audioStepMic.classList.toggle("done", state.steps.mic); - nodes.audioStepCamera.classList.toggle("done", state.steps.camera); - nodes.audioHeard.disabled = state.steps.output; - nodes.audioHeard.textContent = state.steps.output ? "Confirmed" : "I heard it"; + paintPreflight(state, hint); // Consent is a separate gate from media readiness on purpose. It is not // a device that can be proven, it is an answer, and folding it into // `mediaReadiness` would put a legal question inside the function that @@ -627,7 +617,7 @@ function runAudioCheck() { if (!sampleReadiness().ready) return; finished = true; nodes.audioCheck.hidden = true; - clearTimeout(meterRetry); + meter.stop(); pool.cancelRetry(); faceCheck.close(); void context?.close().catch(() => {}); @@ -689,54 +679,13 @@ function runAudioCheck() { nodes.recordingConsentStep.hidden = !recordingEnabled; nodes.recordingConsent.addEventListener("change", refresh); - // One meter at a time. The pool can hand back a replacement microphone - // now, and both ways back into here -- a fresh track and the meter's own - // retry -- would otherwise leave the previous AudioContext reading the - // track that went away and repainting the bar from it every frame. - let meterGeneration = 0; - const watchMic = () => { - const generation = ++meterGeneration; - return startMediaMeter( - pool.stream, - (peak) => { - pool.setError("audio", null); - recentPeaks.push(peak); - if (recentPeaks.length > MIC_CONFIRM_FRAMES) recentPeaks.shift(); - // Latched once proven: the candidate should not have to keep talking - // to hold the Start button open while they read the screen. - micPeak = Math.max(micPeak, sustainedPeak(recentPeaks)); - nodes.audioMeterFill.style.width = `${Math.min(100, Math.round(peak * 300))}%`; - refresh(); - }, - (error) => { - pool.setError("audio", error); - // A device that went away has not proven anything about the one that - // replaces it. - micPeak = 0; - recentPeaks.length = 0; - // The hint outranks the status line on every frame, so a stale one - // would leave the bar red and still reading "play the test tone". - hint = null; - refresh(); - // The gate has no bypass, so it must recover on its own once the - // candidate grants access or plugs a device back in. Distinct from - // the pool's retry: this one restarts the level meter over a track we - // already hold, that one asks the browser for a device again. - meterRetry = setTimeout(watchMic, 2000); - }, - () => finished || generation !== meterGeneration, - ); - }; // A live track is enough for the microphone: the meter is what proves one // actually carries sound, and it runs for the rest of the preflight. pool.configure("audio", { accept: (track) => Boolean(track), - onTrack: () => watchMic(), + onTrack: () => meter.start(), onLost: () => { - meterGeneration += 1; - micPeak = 0; - recentPeaks.length = 0; - clearTimeout(meterRetry); + meter.forget(); nodes.audioMeterFill.style.width = "0%"; refresh(); }, @@ -757,27 +706,6 @@ function runAudioCheck() { }); } -async function startMediaMeter(stream, onPeak, onError, shouldStop) { - try { - const context = new (window.AudioContext || window.webkitAudioContext)(); - const analyser = context.createAnalyser(); - analyser.fftSize = 1024; - context.createMediaStreamSource(stream).connect(analyser); - const samples = new Uint8Array(analyser.frequencyBinCount); - const tick = () => { - if (shouldStop()) { - void context.close().catch(() => {}); - return; - } - analyser.getByteTimeDomainData(samples); - onPeak(peakLevel(samples)); - requestAnimationFrame(tick); - }; - tick(); - } catch (error) { - onError(String(error?.message || error)); - } -} async function connect(preflight, presenting = false) { setAgentStateLabel(providerUiState("connecting").label); diff --git a/web/mic-meter.js b/web/mic-meter.js new file mode 100644 index 00000000..33129928 --- /dev/null +++ b/web/mic-meter.js @@ -0,0 +1,111 @@ +/// The microphone level meter, and the loudest sustained peak it has proven. +/// +/// Its own module for the reason `devices.js` and `face-check.js` are theirs: +/// the generation counter, the retry timer and the rolling window of peaks are +/// one piece of state in three parts, and the invariant that binds them -- a +/// departed microphone proves nothing about the one replacing it -- is worth +/// testing directly rather than reading off the preflight that uses it. +/// +/// Nothing here touches the page. The level goes out through `onLevel` and the +/// caller paints the bar, so this can run against a fake meter in a test. + +import { MIC_CONFIRM_FRAMES, peakLevel, sustainedPeak } from "./audio-check.js"; + +/// The default `startMeter`: a real `AudioContext` over a real stream. +/// +/// Injected rather than called directly, the way `createFaceCheck` takes its +/// detector, because a test cannot build one of these and the thing worth +/// testing is not the analyser. +export async function startMediaMeter(stream, onPeak, onError, shouldStop) { + try { + const context = new (window.AudioContext || window.webkitAudioContext)(); + const analyser = context.createAnalyser(); + analyser.fftSize = 1024; + context.createMediaStreamSource(stream).connect(analyser); + const samples = new Uint8Array(analyser.frequencyBinCount); + const tick = () => { + if (shouldStop()) { + void context.close().catch(() => {}); + return; + } + analyser.getByteTimeDomainData(samples); + onPeak(peakLevel(samples)); + requestAnimationFrame(tick); + }; + tick(); + } catch (error) { + onError(String(error?.message || error)); + } +} + +/// One meter at a time. The pool can hand back a replacement microphone, and +/// both ways back in -- a fresh track and the meter's own retry -- would +/// otherwise leave the previous AudioContext reading the track that went away +/// and reporting levels from it every frame. +export function createMicMeter({ + pool, + isFinished, + onLevel, + onFailure, + startMeter = startMediaMeter, + retryMs = 2000, +}) { + let peak = 0; + let generation = 0; + let retry = null; + const recent = []; + + /// Nothing a departed microphone proved carries over to its replacement. + const forgetPeak = () => { + peak = 0; + recent.length = 0; + }; + + const watch = () => { + const mine = ++generation; + return startMeter( + pool.stream, + (level) => { + pool.setError("audio", null); + recent.push(level); + if (recent.length > MIC_CONFIRM_FRAMES) recent.shift(); + // Latched once proven: the candidate should not have to keep talking + // to hold the Start button open while they read the screen. + peak = Math.max(peak, sustainedPeak(recent)); + onLevel(level); + }, + (error) => { + pool.setError("audio", error); + forgetPeak(); + onFailure(error); + // Not if this meter was forgotten on the way through. `onFailure` + // repaints, the repaint drops an ended track, and dropping it forgets + // this meter -- all before the line below runs, so a retry armed here + // is one `forget` already cleared and cannot clear again. It would + // start a meter over a stream the pool has let go of. + if (mine !== generation || isFinished()) return; + // The gate has no bypass, so it must recover on its own once the + // candidate grants access or plugs a device back in. Distinct from the + // pool's retry: this one restarts the level meter over a track we + // already hold, that one asks the browser for a device again. + retry = setTimeout(watch, retryMs); + }, + () => isFinished() || mine !== generation, + ); + }; + + return { + peak: () => peak, + start: watch, + /// The track is gone. The generation bump is what stops the meter still + /// running over it from reporting another level. + forget() { + generation += 1; + forgetPeak(); + clearTimeout(retry); + }, + stop() { + clearTimeout(retry); + }, + }; +}