From d89b4a2eb923ef93301df8b0fcfc0eb6cf38cc1f Mon Sep 17 00:00:00 2001 From: Hell HBBD Date: Thu, 22 Oct 2026 17:31:19 +0800 Subject: [PATCH] Wait for Live output before starting interviews Live setup can succeed without the model answering. Starting the clock at setup spends candidate time against a silent provider. Require model output before timing an interview, and retain untimed practice after a failed opening without retrying that generation. Readiness must preserve existing setup failover, keepalive, candidate media and reconnect grace. Retain provider failure reasons and observed usage so failed starts remain diagnosable without duplicate generation or accounting. Protect editable work before the clock starts and after an untimed failure. Detect early interviewer departures, keep recording delays from blocking cleanup, and prevent premature report requests. Keep Strict failure guidance consistent with its disabled controls. Isolate startup tests from unrelated model loading and rendering so they measure startup ordering reliably. Closes #159 --- docs/provider-cost-and-degradation.md | 115 ++- scripts/gen-wire-fixtures.mjs | 13 + src/gemini.rs | 120 ++- src/livekit.rs | 606 +++++++++++--- tests/browser/dom-contract.test.js | 20 - tests/browser/interview-lifecycle.test.js | 357 ++++++++- tests/browser/interview-startup.test.js | 928 ++++++++++++++++++++++ tests/browser/lib.test.js | 26 + tests/browser/report-recovery.test.js | 1 + tests/browser/restore-starter.test.js | 13 +- tests/browser/strict-selection.test.js | 17 +- tests/fixtures/startup-receipt.json | 12 + tests/unit/gemini.rs | 354 +++++++++ tests/unit/livekit.rs | 741 ++++++++++++++++- tests/unit/livekit/media.rs | 3 + tests/unit/livekit/startup.rs | 401 ++++++++++ web/before-unload.js | 14 +- web/interview.js | 123 ++- web/lib.js | 78 +- 19 files changed, 3697 insertions(+), 245 deletions(-) create mode 100644 tests/browser/interview-startup.test.js create mode 100644 tests/fixtures/startup-receipt.json create mode 100644 tests/unit/livekit/startup.rs diff --git a/docs/provider-cost-and-degradation.md b/docs/provider-cost-and-degradation.md index 3dd8dbcd..2166771e 100644 --- a/docs/provider-cost-and-degradation.md +++ b/docs/provider-cost-and-degradation.md @@ -45,33 +45,102 @@ at once or was held for an unpause or the end of a thinking hold: every replacement asks for one, so a run of sockets that each answer their briefing and die still ends the interview. -A candidate turn, required prompt (including the opening greeting), or tool -continuation that gets nothing back for 10 seconds, not even the end of a turn -that chose silence, replaces the socket even without a `GoAway`. This bounds -sockets that stay open without answering. -`CODETRIAL_GEMINI_FIRST_OUTPUT_S` sets that deadline within 5 to 120 seconds, -never above the reply timeout below, for a model whose replies are slow to -begin. The deadline waits while Gemini reports the candidate still mid-turn. A -prompt sent after such a reply waits its own full deadline, and when Gemini -cuts a stalled generation to take the prompt, the cut leaves the reply owed -rather than settled, since the candidate did not cause it. A generation that -has started and then stops producing output is given longer, 45 seconds, since -its next chunk may simply be slow; `CODETRIAL_GEMINI_REPLY_TIMEOUT_S` sets that -interval within 20 to 120 seconds. A recovery prompt, a replacement +Before the interview timer starts, one bounded Live generation verifies that +the configured model can answer, not merely accept setup. The opening greeting +is that request on the interview's own socket; there is no separate paid probe +turn. The initial socket open retains its bounded cold-open retry and key +failover policy; those attempts send setup, not a readiness generation. Once +setup succeeds, the greeting is sent once and a failed or silent generation +is not retried. Connect and setup retain their existing limits, and +`CODETRIAL_GEMINI_FIRST_OUTPUT_S` bounds the greeting send and wait for the +first model output, floored at `DEFAULT_GEMINI_FIRST_OUTPUT_S` (10 seconds) +and capped by the reply timeout. A shorter setting still makes ordinary +post-startup recovery more eager, but cannot end startup before 10 seconds. A +silent startup fails without replacing the socket or repeating the greeting; +the recovery policy below applies only after readiness. Keepalive pings and +candidate media continue during that wait, without another explicit generation +request, so a slow first output cannot overflow the microphone's setup queue. +Candidate departure during startup does not cancel the opening: startup keeps +its original output deadline and queues the departure and any return for the +interview loop's normal presence grace. Room disconnection still cancels startup, +closes the Live session and releases the task's capacity slot without waiting for +the output deadline. Other room events remain queued for the interview loop. +Nonempty audio, model text, output transcription, or a nonempty tool-call batch counts; input +transcription, voice activity, usage, empty output and turn completion alone +do not. The private clock travels as non-generating context before the opening +turn; it does not add another generation request. + +Events observed by the probe retain their original receive timestamps and are +preserved, in order, for the normal room loop, with the event queue's bound +also applied to the startup buffer. Only observed usage is logged on failure; +no observation does not mean the request was free. +The server clock and hard deadline begin at successful output, and the browser +starts its countdown from receipt of readiness. It retains an early readiness +notice while publishing tracks or starting recording. Recording is still +requested after joining the room, not delayed until the greeting is heard. +Its response does not block readiness or startup-failure cleanup. Recording +status remains independent of the readiness outcome, since the server may have +started recording before the opening failed. + +Startup outcomes use the existing `interviewer_state` control topic with +`startup: "ready"` or `startup: "failed"`, and `reconnecting: false`. A failure +can carry a classified `reason`: `quota`, `billing`, `credentials`, or `refused`. +Only a provider response identifies those causes; timeout, 1011, 503, and an +unexplained close use `unavailable`. The browser uses fixed wording through the +existing degraded-start banner, not raw provider text or a second notification +channel. A missing notice or a disconnected room also ends its finite startup +wait without starting the timer. Replacement notices contain no startup field +and cannot restart the clock. + +Only the pinned interviewer can settle browser startup. Before leaving after a +failure, the agent waits for the candidate's `startup_received` control receipt, +whose `deliveryId` hashes the exact failure packet bytes. Publication and receipt +share the existing one-second notice budget; the notice is not retransmitted. +The browser retains the classified failure immediately and gives its receipt a +bounded chance to leave before cleanup. This keeps the agent identifiable while +the packet is delivered: LiveKit removes a departed sender from its roster, so a +packet arriving afterwards has no participant. Such unattributed notices cannot +settle startup. An unexpected departure without an identified notice ends with +the generic unavailable guidance instead of guessing a quota or billing cause. +Before an interviewer identity is pinned, an agent departure also ends the wait; +after pinning, departures from other agents cannot fail it. + +A deployment without a live session still offers timed offline practice as +before. That local clock is not granted after a joined session fails readiness. +Editable work is protected by the browser's leave warning independently of the +clock, including while awaiting readiness and during untimed offline practice. +End and Pause remain disabled during startup. Practice enables them again when +startup fails; Strict keeps editing and these controls disabled and offers a +return to the lobby or another media preflight instead of promising offline +practice. A readiness result cannot start the clock after the page has moved +out of its live phase. + +After startup, a candidate turn, required prompt, or tool continuation that +gets nothing back for 10 seconds, not even the end of a turn that chose silence, +replaces the socket even without a `GoAway`. This bounds sockets that stay open +without answering. `CODETRIAL_GEMINI_FIRST_OUTPUT_S` sets that deadline within +5 to 120 seconds, never above the reply timeout below, for a model whose replies +are slow to begin. The deadline waits while Gemini reports the candidate still +mid-turn. A prompt sent after such a reply waits its own full deadline, and when +Gemini cuts a stalled generation to take the prompt, the cut leaves the reply +owed rather than settled, since the candidate did not cause it. A generation +that has started and then stops producing output is given longer, 45 seconds, +since its next chunk may simply be slow; `CODETRIAL_GEMINI_REPLY_TIMEOUT_S` +sets that interval within 20 to 120 seconds. A recovery prompt, a replacement socket's briefing or a same-socket reply check, is timed from its own request and never given less than the default 10 seconds for its first output, so a shorter `CODETRIAL_GEMINI_FIRST_OUTPUT_S` cannot cut one recovery straight into the next. A superseding candidate turn or ordinary prompt uses the configured -deadline again. The `codetrial turn_timing` summaries -record observed reply waits, and [reply timing](turn-timing.md) defines their -measurement boundaries and unavailable observations; the -[latency records](latency-observability.md) break the server reply path into -segments, including tool batches and output queueing. A reply the watchdog -replaces is paid for twice, so an endpoint whose slow replies do arrive -warrants a longer timeout. For recovery, a candidate turn is timed from the -last transcript fragment of it, and a fragment arriving within two seconds of -a turn that said something is taken as the lagging tail of speech that turn -already answered, so it owes nothing. The replacement, resumed or cold, carries +deadline again. The `codetrial turn_timing` summaries record observed reply +waits, and [reply timing](turn-timing.md) defines their measurement boundaries +and unavailable observations; the [latency records](latency-observability.md) +break the server reply path into segments, including tool batches and output +queueing. A reply the watchdog replaces is paid for twice, so an endpoint whose +slow replies do arrive warrants a longer timeout. For recovery, a candidate +turn is timed from the last transcript fragment of it, and a fragment arriving +within two seconds of a turn that said something is taken as the lagging tail +of speech that turn already answered, so it owes nothing. The replacement, +resumed or cold, carries the unanswered reply across and publishes the same reconnecting state. A reply cut off mid-generation is owed too, and the replacement is told not to repeat what was already said. Optional editor reviews accept silence; queued audio and diff --git a/scripts/gen-wire-fixtures.mjs b/scripts/gen-wire-fixtures.mjs index 082f2f45..a7b79731 100755 --- a/scripts/gen-wire-fixtures.mjs +++ b/scripts/gen-wire-fixtures.mjs @@ -416,6 +416,19 @@ const files = { }, ], }, + "startup-receipt.json": { + topic: lib.topics.control, + cases: [ + { + name: "received startup failure", + payload: await lib.startupReceipt( + new TextEncoder().encode( + '{"type":"interviewer_state","reconnecting":false,"startup":"failed","reason":"billing"}', + ), + ), + }, + ], + }, "test-results.json": { topic: lib.topics.tests, cases: testResultsCases() }, "integrity-chain.json": await integrityChain(), "board-stream.json": { topic: lib.topics.board, cases: boardCases() }, diff --git a/src/gemini.rs b/src/gemini.rs index d77da579..2bf158aa 100644 --- a/src/gemini.rs +++ b/src/gemini.rs @@ -55,7 +55,7 @@ pub(crate) const CLOSE_TIMEOUT: Duration = Duration::from_secs(2); /// How often the room loop pings. A socket nobody is writing to cannot fail a /// write, and a paused interview writes nothing, so without a ping a peer that /// went away there would not be noticed until something was finally said. -const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15); +pub(crate) const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15); /// Silence the reader accepts before it treats the peer as gone. Gemini answers /// a ping within milliseconds, so three missed intervals is not a slow network; /// it is a socket that will never deliver another event, and ending the reader @@ -129,6 +129,7 @@ pub struct GeminiLiveSession { reader: JoinHandle<()>, events: Receiver<(std::time::Instant, GeminiEvent)>, event_received_at: Option, + pending_events: std::collections::VecDeque<(std::time::Instant, GeminiEvent)>, /// When this socket was opened, which is how the caller tells Gemini's /// ordinary connection cap from an endpoint that will not stay up. It lives /// here rather than in the loop that reconnects because it is a fact about @@ -187,17 +188,41 @@ impl GeminiLiveSession { pub(crate) fn recovery_handle(&self, keys: &GeminiKeys) -> Option<(String, String)> { let key = self.credential.as_ref()?; + self.record_failure(keys); + if keys.select().ok().as_ref() != Some(key) { + return None; + } + Some((key.clone(), self.resumption_handle()?)) + } + + pub(crate) fn record_failure(&self, keys: &GeminiKeys) { if let Some(failure) = *self .failure .lock() .unwrap_or_else(|error| error.into_inner()) + && let Some(key) = &self.credential { keys.failed(key, failure, ApiSurface::Live); } - if keys.select().ok().as_ref() != Some(key) { - return None; + } + + pub(crate) fn startup_error( + &self, + error: Box, + ) -> Box { + match *self + .failure + .lock() + .unwrap_or_else(|error| error.into_inner()) + { + Some(failure) => ApiFailure { + status: 0, + credential: Some(failure), + detail: Some("Gemini refused startup".to_string()), + } + .into(), + None => error, } - Some((key.clone(), self.resumption_handle()?)) } pub async fn send_text( @@ -267,7 +292,10 @@ impl GeminiLiveSession { } pub async fn next_event(&mut self) -> Option { - let (received_at, event) = self.events.recv().await?; + let (received_at, event) = match self.pending_events.pop_front() { + Some(event) => event, + None => self.events.recv().await?, + }; self.event_received_at = Some(received_at); Some(event) } @@ -276,6 +304,52 @@ impl GeminiLiveSession { self.event_received_at } + /// The startup gate observes output without consuming the greeting. The + /// room loop must still deliver it and account for earlier usage frames. + /// Until that loop starts, this wait must also ping: the reader's idle + /// watchdog can be shorter than the configured reply timeout. + pub(crate) async fn wait_for_model_output( + &mut self, + keepalive: &mut tokio::time::Interval, + ) -> Result<(), Box> { + loop { + let event = tokio::select! { + event = self.events.recv() => event, + _ = keepalive.tick() => { + self.keep_alive().await?; + continue; + } + }; + let Some((received_at, event)) = event else { + break; + }; + let ready = match &event { + GeminiEvent::Audio { bytes, .. } => !bytes.is_empty(), + GeminiEvent::Text(text) | GeminiEvent::OutputTranscript(text) => { + !text.trim().is_empty() + } + GeminiEvent::ToolCall(calls) => !calls.is_empty(), + _ => false, + }; + if self.pending_events.len() == GEMINI_EVENT_QUEUE { + return Err(io::Error::other("Gemini startup event limit reached").into()); + } + self.pending_events.push_back((received_at, event)); + if ready { + return Ok(()); + } + } + Err(ApiFailure { + status: 0, + credential: *self + .failure + .lock() + .unwrap_or_else(|error| error.into_inner()), + detail: Some("Gemini ended before producing startup output".to_string()), + } + .into()) + } + /// How long this socket has been up. pub fn age(&self) -> Duration { self.opened_at.elapsed() @@ -428,6 +502,13 @@ impl GeminiLiveSession { } } +impl Drop for GeminiLiveSession { + fn drop(&mut self) { + // Cancelling startup must not detach a reader holding a metered socket. + self.reader.abort(); + } +} + /// Modality counts reported by the provider; absent details are not zero usage. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct ModalityUsage { @@ -701,6 +782,24 @@ pub(crate) fn is_billing_failure(error: &(dyn std::error::Error + 'static)) -> b || credentials::exhausted_by_billing(error) } +/// Machine-owned reasons only; silence and an internal provider error do not +/// identify quota or billing. The browser uses these on interviewer_state. +pub(crate) fn startup_failure_reason(error: &(dyn std::error::Error + 'static)) -> &'static str { + if credentials::exhausted_by_billing(error) { + return "billing"; + } + if credentials::exhausted_until(error).is_some() { + return "quota"; + } + match credential_failure(error) { + Some(CredentialFailure::Quota) => "quota", + Some(CredentialFailure::Billing) => "billing", + Some(CredentialFailure::Invalid) => "credentials", + Some(CredentialFailure::Refused) => "refused", + None => "unavailable", + } +} + /// Bounds `open` by `limit`, the way `FIRST_OPEN_LIMIT` bounds a first open. pub(crate) async fn first_open_within( limit: Duration, @@ -715,14 +814,12 @@ pub(crate) async fn first_open_within( }) } -/// Opens one Live session the way an interview's first open would choose its -/// key, moving past keys Gemini rejects while backups remain. +/// Checks configured Live credentials, moving past keys Gemini rejects while +/// backups remain. /// /// For `check-gemini`, which should pass whenever an interview could start. -/// It makes at most one attempt per configured key, and no more than the -/// interview's first open could: that open gets one attempt plus the restart -/// budget, all inside `FIRST_OPEN_LIMIT`, so a key past either passes a check -/// but never runs an interview. +/// It makes at most one attempt per configured key under its diagnostic budget, +/// all inside `FIRST_OPEN_LIMIT`. /// Marking a rejected key unavailable is not a bound on its own: a quota /// cooldown lasts a minute, and slow rejections from enough keys outlast it, /// putting the first key back in rotation. Other failures return at once: a @@ -1833,6 +1930,7 @@ async fn open_live_session_redacted_at( writer, reader, events, + pending_events: std::collections::VecDeque::new(), resumption, usage, checkpoint_at, diff --git a/src/livekit.rs b/src/livekit.rs index f07cde1a..3be8efb0 100644 --- a/src/livekit.rs +++ b/src/livekit.rs @@ -489,6 +489,29 @@ fn runtime_activity(config: &AgentConfig, started_at: Instant, session_id: u64) .with_first_output_deadline(Duration::from_secs(u64::from(config.gemini_first_output_s))) } +// Latency includes the opening wait, but recovery starts its clock at readiness +// so replay cannot expire an already answered greeting before its first queued +// event reaches the room loop. +fn ready_runtime_activity( + config: &AgentConfig, + room_name: &str, + started_at: Instant, + opening_requested_at: Instant, + session_id: u64, + greeting: &str, +) -> RuntimeActivity { + let mut activity = runtime_activity(config, started_at, session_id); + activity.latency.bind( + room_name, + session_id, + activity.live_socket, + config.gemini_silence_ms, + ); + activity.mark_prompted(opening_requested_at, Some(greeting), false); + activity.resume_reply_wait(started_at); + activity +} + /// Opens a session that remembers nothing, retrying while the budget allows. /// /// `Err` means credentials are unavailable or retries stopped, ending the @@ -500,9 +523,13 @@ async fn open_cold_session( restarts: &mut usize, room: Option<&RoomLink>, ) -> Result> { + let deadline = room.is_none().then(|| { + tokio::time::Instant::from_std(interview.started_at) + crate::gemini::FIRST_OPEN_LIMIT + }); retry_cold_open( interview.keys, restarts, + deadline, || live_session_with_keys(interview.keys, interview.boot, None), async |notice| { let Some(room) = room else { return }; @@ -526,6 +553,7 @@ async fn open_cold_session( async fn retry_cold_open( keys: &GeminiKeys, restarts: &mut usize, + deadline: Option, mut open: F, mut notify: N, ) -> Result> @@ -549,6 +577,18 @@ where let (reason, until) = match opened { Ok(session) => return Ok(session), + // A known cooldown that cannot fit the first-open budget cannot + // recover here. Preserve its cause rather than letting the outer + // timeout replace it with an unexplained transport failure. + Err(error) + if crate::gemini::exhausted_until(error.as_ref()).is_some_and(|until| { + deadline + .is_some_and(|deadline| tokio::time::Instant::from_std(until) >= deadline) + }) => + { + return Err(error); + } + // Every key is out on quota. A sole key would ride out the same // rate limit under this budget, so a rotation waits too. Err(error) @@ -602,6 +642,154 @@ fn interviewer_state(reconnecting: bool) -> serde_json::Value { serde_json::json!({ "type": "interviewer_state", "reconnecting": reconnecting }) } +fn startup_state(error: Option<&(dyn std::error::Error + 'static)>) -> serde_json::Value { + let mut notice = interviewer_state(false); + notice["startup"] = if error.is_some() { "failed" } else { "ready" }.into(); + if let Some(error) = error { + notice["reason"] = crate::gemini::startup_failure_reason(error).into(); + } + notice +} + +/// The normal opening generation is the probe, not a second paid turn. Setup +/// alone proves nothing, and no watchdog or replacement runs before it answers. +async fn probe_live_output( + gemini: &mut GeminiLiveSession, + state: &mut RuntimeState, + greeting: &str, + limit: Duration, + startup: &mut StartupContext<'_>, + input: &mut impl StartupInputs, +) -> Result> { + let result = tokio::time::timeout(limit, async { + send_model_text( + gemini, + state, + ModelInputKind::Turn, + TurnCause::Turn, + greeting, + ) + .await?; + + // These survive cancellation of the output wait when media wins the + // select. Recreating either would restart the deadline or ping on every + // microphone frame. The generation above is never part of that select. + let mut keepalive = tokio::time::interval(crate::gemini::KEEPALIVE_INTERVAL); + keepalive.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + ready = gemini.wait_for_model_output(&mut keepalive) => { + ready?; + return Ok(Instant::now()); + } + input = input.next(startup.media) => { + if matches!(&input, StartupInput::Room(None | Some(RoomEvent::Disconnected { .. }))) { + *startup.room_closed = true; + return Err(std::io::Error::new( + std::io::ErrorKind::Interrupted, + "room closed before startup output", + ).into()); + } + match input { + StartupInput::Room(event) => { + if let Some(event) = event + && !handle_media_event( + startup.media, + gemini, + startup.candidate_identity, + startup.candidate_video, + &event, + ).await? + { + startup.deferred_events.push(event); + } + } + StartupInput::Audio(frame) => { + let ended = frame.is_none(); + if frame.as_ref().is_some_and(media::frame_has_voice) { + startup.media.voiced_at = Some(Instant::now()); + } + pump_audio(startup.media, gemini, frame).await?; + release_if_ended(&mut startup.media.audio, ended); + } + StartupInput::Video(frame) => { + let speaking = startup.media.voiced_recently(Instant::now()); + pump_video(startup.media, gemini, frame, speaking).await?; + } + + // The room loop applies presence grace after readiness. + // Preserve the departure and any later return in order. + StartupInput::ParticipantLeft(event) => startup.deferred_events.push(event), + } + } + } + } + }) + .await + .unwrap_or_else(|_| { + Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "Gemini did not produce startup output before the deadline", + ) + .into()) + }); + result.map_err(|error| { + if *startup.room_closed { + error + } else { + gemini.startup_error(error) + } + }) +} + +struct StartupContext<'a> { + media: &'a mut CandidateMedia, + candidate_identity: &'a str, + candidate_video: bool, + deferred_events: &'a mut Vec, + room_closed: &'a mut bool, +} + +enum StartupInput { + Room(Option), + ParticipantLeft(RoomEvent), + Audio(Option<::livekit::webrtc::audio_frame::AudioFrame<'static>>), + Video(Option<::livekit::webrtc::video_frame::BoxVideoFrame>), +} + +trait StartupInputs: Send { + fn next(&mut self, media: &mut CandidateMedia) -> impl Future + Send; +} + +impl StartupInputs for &tokio::sync::Mutex> { + async fn next(&mut self, media: &mut CandidateMedia) -> StartupInput { + next_startup_input(media, self).await + } +} + +async fn next_startup_input( + media: &mut CandidateMedia, + events: &tokio::sync::Mutex>, +) -> StartupInput { + tokio::select! { + event = async { events.lock().await.recv().await } => match event { + Some(event @ RoomEvent::ParticipantDisconnected(_)) => StartupInput::ParticipantLeft(event), + event => StartupInput::Room(event), + }, + frame = next_audio_frame(&mut media.audio), if media.audio.is_some() => StartupInput::Audio(frame), + frame = next_video_frame(&mut media.video), if media.video.is_some() => StartupInput::Video(frame), + } +} + +fn startup_output_limit(config: &AgentConfig) -> Duration { + Duration::from_secs(u64::from( + config + .gemini_first_output_s + .max(crate::config::DEFAULT_GEMINI_FIRST_OUTPUT_S) + .min(config.gemini_reply_timeout_s), + )) +} + /// The notice a replacement shows while it waits on `reason`. `None` is the /// attempt itself, whose length no one knows. fn reconnect_wait(reason: &str, wait: Option) -> serde_json::Value { @@ -1529,6 +1717,87 @@ struct OpenSession<'a> { turn: TurnState, started_at: Instant, restarts: usize, + deferred_events: std::collections::VecDeque, +} + +// A failed ready notice can follow a paid generation. Stop the reader before +// draining so the common startup summary receives each observation once. +async fn publish_startup_readiness( + gemini: &mut GeminiLiveSession, + usage: &mut crate::gemini::TokenUsage, + publication: impl std::future::Future>>, +) -> Result<(), Box> { + let result = tokio::time::timeout(RECONNECT_NOTICE_LIMIT, publication) + .await + .unwrap_or_else(|_| { + Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "publishing startup readiness timed out", + ) + .into()) + }); + if result.is_err() { + let _ = gemini.shutdown().await; + for observed in gemini.drain_usage() { + usage.add(observed); + } + } + result +} + +// Publication only queues the notice. Keep its sender in the roster until the +// candidate acknowledges it, within the same budget and without retransmission. +async fn publish_startup_failure( + publication: impl std::future::Future>>, + receipt: impl std::future::Future, +) { + let _ = tokio::time::timeout(RECONNECT_NOTICE_LIMIT, async { + if publication.await.is_ok() { + receipt.await; + } + }) + .await; +} + +fn is_startup_receipt( + topic: Option<&str>, + sender: Option<&str>, + candidate: &str, + payload: &[u8], + id: &str, +) -> bool { + topic == Some(TOPIC_CONTROL) + && is_interview_participant(sender, candidate) + && serde_json::from_slice::(payload) + .is_ok_and(|value| value["type"] == "startup_received" && value["deliveryId"] == id) +} + +async fn account_startup( + config: &AgentConfig, + room_name: &str, + session_id: u64, + log_startup_failure: impl FnOnce(String), + opening: impl AsyncFnOnce( + &mut crate::gemini::TokenUsage, + ) -> Result>, +) -> Result> { + let mut startup_usage = crate::gemini::TokenUsage::default(); + let opened = opening(&mut startup_usage).await; + + // Room setup can fail before a Gemini socket exists. Both that path and a + // paid opening failure must reach exactly one summary with observed usage. + if let Err(error) = &opened { + log_startup_failure(live_usage_line( + room_name, + session_id, + &config.gemini_live_model, + 0, + LiveOutcome::from_error(error.as_ref(), LiveOutcome::Error), + None, + &startup_usage, + )); + } + opened } enum OpenSessionOutcome<'a> { @@ -1552,33 +1821,47 @@ async fn open_session<'a>( room_name: &'a str, now_seconds: u64, session_id: u64, + log_startup_failure: impl FnOnce(String), ) -> Result, Box> { let agent_identity = agent_identity(room_name); - let (room, mut events) = match join_room(config, room_name, &agent_identity, now_seconds).await - { + let joined = async { + let (room, mut events) = join_room(config, room_name, &agent_identity, now_seconds).await?; + let room = RoomLink::live(room); + let mut agent_state = session::AgentState::default(); + session::publish_opening_attributes(&room, &mut agent_state, config.gemini_silence_ms) + .await?; + + // The candidate's metadata picks the problem, so nothing else can start + // until someone joins. + let mut deferred_events = Vec::new(); + let candidate = + candidate_metadata(room.connected(), &mut events, &mut deferred_events).await?; + Ok::<_, Box>(( + room, + events, + agent_state, + candidate, + deferred_events, + )) + } + .await; + let (room, events, mut agent_state, candidate, mut deferred_events) = match joined { Ok(joined) => joined, - Err(error) => return Ok(OpenSessionOutcome::FailedBeforeCandidate(error)), + Err(error) => { + log_startup_failure(startup_failure_line( + &config.gemini_live_model, + room_name, + session_id, + error.as_ref(), + )); + return Ok(OpenSessionOutcome::FailedBeforeCandidate(error)); + } }; - let room = RoomLink::live(room); - let mut agent_state = session::AgentState::default(); - if let Err(error) = - session::publish_opening_attributes(&room, &mut agent_state, config.gemini_silence_ms).await - { - return Ok(OpenSessionOutcome::FailedBeforeCandidate(error)); - } - - // The candidate's metadata picks the problem, so nothing else can start - // until someone joins. - let mut deferred_events = Vec::new(); - let candidate = - match candidate_metadata(room.connected(), &mut events, &mut deferred_events).await { - Ok(candidate) => candidate, - Err(error) => return Ok(OpenSessionOutcome::FailedBeforeCandidate(error)), - }; let Some((candidate_identity, candidate_metadata)) = candidate else { eprintln!("no candidate joined room={room_name}; leaving it"); return Ok(OpenSessionOutcome::NoCandidate); }; + account_startup(config, room_name, session_id, log_startup_failure, async |startup_usage| { let events = tokio::sync::Mutex::new(events); let boot = candidate_bootstrap(config, room_name, Some(&candidate_metadata)); eprintln!( @@ -1622,99 +1905,166 @@ async fn open_session<'a>( // dependency is on room state, not on data. let setup_began = Instant::now(); let mut restarts = 0; - let mut room_ready = Duration::ZERO; - let mut gemini_ready = Duration::ZERO; - let (mut output_audio, mut gemini) = tokio::try_join!( - async { - isolate_local_agent(config, room_name, &agent_identity, now_seconds).await?; - let published = - publish_output_audio(room.connected(), GEMINI_OUTPUT_AUDIO_SAMPLE_RATE).await; - room_ready = setup_began.elapsed(); - published - }, - async { - let interview = InterviewContext { - config, - keys, - boot: &boot, - started_at: setup_began, - candidate_identity: &candidate_identity, - events: &events, - slot: None, - }; - let opened = crate::gemini::first_open_within( - crate::gemini::FIRST_OPEN_LIMIT, - open_cold_session(interview, &mut restarts, None), - ) - .await; - gemini_ready = setup_began.elapsed(); - opened - }, - )?; + let mut room_closed = false; + let prepared = async { + let mut room_ready = Duration::ZERO; + let mut gemini_ready = Duration::ZERO; + let (mut output_audio, mut gemini) = tokio::try_join!( + async { + isolate_local_agent(config, room_name, &agent_identity, now_seconds).await?; + let published = publish_output_audio(room.connected(), GEMINI_OUTPUT_AUDIO_SAMPLE_RATE).await; + room_ready = setup_began.elapsed(); + published + }, + async { + let interview = InterviewContext { + config, + keys, + boot: &boot, + started_at: setup_began, + candidate_identity: &candidate_identity, + events: &events, + slot: None, + }; + let opened = crate::gemini::first_open_within( + crate::gemini::FIRST_OPEN_LIMIT, + open_cold_session(interview, &mut restarts, None), + ) + .await; + gemini_ready = setup_began.elapsed(); + opened + }, + )?; + output_audio.tool_ack = tool_ack; + output_audio.tool_ack_task = tool_ack_task; + eprintln!( + "timing: room and Gemini session ready {:.2}s after the candidate joined (room {:.2}s, gemini {:.2}s)", + setup_began.elapsed().as_secs_f64(), + room_ready.as_secs_f64(), + gemini_ready.as_secs_f64() + ); - output_audio.tool_ack = tool_ack; - output_audio.tool_ack_task = tool_ack_task; + // Only prompt construction uses this provisional clock. The opening + // starts with its full budget, and no watch or round transition runs + // before ready. + let started_at = Instant::now(); + + let mut turn = TurnState { + state: initial_runtime_state(&boot, started_at), + agent_state: std::mem::take(&mut agent_state), + activity: runtime_activity(config, started_at, session_id), + turns: SpeakerTurns::default(), + transcripts: Default::default(), + }; + turn.state + .evidence_ledger + .record_model_input(ModelInputKind::LiveSetup, &boot.instructions); + let mut media = CandidateMedia::new(); + + // Keep subscriptions consumed while waiting for candidate metadata. The + // startup wait pumps them too; dropping a subscription here would leave + // both waits deaf for the whole interview. + for event in std::mem::take(&mut deferred_events) { + if !handle_media_event( + &mut media, + &mut gemini, + &candidate_identity, + boot.candidate_video, + &event, + ) + .await? { + deferred_events.push(event); + } + } - // Which branch held the greeting back: an earlier Gemini connect can only - // help when Gemini is the later of the two. - eprintln!( - "timing: room and Gemini session ready {:.2}s after the candidate joined (room {:.2}s, gemini {:.2}s)", - setup_began.elapsed().as_secs_f64(), - room_ready.as_secs_f64(), - gemini_ready.as_secs_f64() - ); + let greeting = with_timer(&turn.state, boot.greeting.clone()); + let opening_requested_at = Instant::now(); + let started_at = match probe_live_output( + &mut gemini, + &mut turn.state, + &greeting, + startup_output_limit(config), + &mut StartupContext { + media: &mut media, + candidate_identity: &candidate_identity, + candidate_video: boot.candidate_video, + deferred_events: &mut deferred_events, + room_closed: &mut room_closed, + }, + &mut &events, + ) + .await + { + Ok(started_at) => started_at, + Err(error) => { + if !room_closed { + gemini.record_failure(keys); + } + let _ = gemini.shutdown().await; - // The candidate's clock, not a third one. The browser starts counting when - // its own connect resolves, which is this instant seen from the other side, - // and stamping again here put the agent a second or two behind the tab for - // the rest of the interview. Every consumer below reads this as "when the - // interview started", and that is when it started. - let started_at = setup_began; - - let mut turn = TurnState { - state: initial_runtime_state(&boot, started_at), - agent_state: std::mem::take(&mut agent_state), - activity: runtime_activity(config, started_at, session_id), - turns: SpeakerTurns::default(), - transcripts: Default::default(), - }; - turn.activity.latency.bind( - room.name(), - session_id, - turn.activity.live_socket, - config.gemini_silence_ms, - ); - turn.state - .evidence_ledger - .record_model_input(ModelInputKind::LiveSetup, &boot.instructions); - let mut media = CandidateMedia::new(); - - // Before the greeting, not after: these arrived while we were waiting for - // the candidate, and one of them is what attaches the microphone. Greeting - // first would open the interview deaf for as long as Gemini takes to - // answer. - for event in deferred_events.drain(..) { - handle_media_event( - &mut media, + // A failed generation can still be billed. Only observed + // counters are reported; absence of observations is not a claim + // of zero cost. + for usage in gemini.drain_usage() { + startup_usage.add(usage); + } + return Err(error); + } + }; + turn.state.started_at = started_at; + turn.activity = ready_runtime_activity( + config, + room_name, + started_at, + opening_requested_at, + session_id, + &greeting, + ); + publish_startup_readiness( &mut gemini, - &candidate_identity, - boot.candidate_video, - &event, + startup_usage, + publish_interviewer_state(&room, &startup_state(None)), ) .await?; + Ok::<_, Box>(( + output_audio, + gemini, + media, + turn, + started_at, + )) } + .await; - let greeting = with_timer(&turn.state, boot.greeting.clone()); - send_model_text( - &mut gemini, - &mut turn.state, - ModelInputKind::Turn, - TurnCause::Turn, - &greeting, - ) - .await?; - turn.activity - .mark_prompted(Instant::now(), Some(&greeting), false); + let (output_audio, gemini, media, turn, started_at) = match prepared { + Ok(prepared) => prepared, + Err(error) => { + if !room_closed && let Ok(packet) = browser_packet(TOPIC_CONTROL, &startup_state(Some(error.as_ref()))) { + let id = crate::sha256_hex(&[&packet.payload]); + publish_startup_failure( + room.publish_data(packet), + async { + let mut events = events.lock().await; + while let Some(event) = events.recv().await { + match event { + RoomEvent::DataReceived { payload, topic, participant, .. } => { + let sender = participant.as_ref().map(|participant| participant.identity().0); + if is_startup_receipt(topic.as_deref(), sender.as_deref(), &candidate_identity, &payload, &id) { + return; + } + } + RoomEvent::Disconnected { .. } => return, + RoomEvent::ParticipantDisconnected(participant) if participant.identity().0 == candidate_identity => return, + _ => {} + } + } + }, + ).await; + } + leave_room(&room).await; + return Err(error); + } + }; Ok(OpenSessionOutcome::Started(Box::new(OpenSession { room, @@ -1728,7 +2078,9 @@ async fn open_session<'a>( turn, started_at, restarts, + deferred_events: deferred_events.into(), }))) + }).await } /// The room loop's own bookkeeping, distinct from the session it runs on. @@ -2650,32 +3002,21 @@ pub(crate) async fn run_room_with_slot( u64::try_from(since.as_millis()).unwrap_or(u64::MAX) }); let keys = Arc::new(GeminiKeys::from_config(config)); - let session = open_session(config, &keys, room_name, now_seconds, session_id) - .await - .inspect_err(|error| { - eprintln!( - "{}", - startup_failure_line( - &config.gemini_live_model, - room_name, - session_id, - error.as_ref(), - ) - ); - })?; + let session = open_session( + config, + &keys, + room_name, + now_seconds, + session_id, + |summary| { + eprintln!("{summary}"); + }, + ) + .await?; let session = match session { OpenSessionOutcome::Started(session) => session, OpenSessionOutcome::NoCandidate => return Ok(RoomRunOutcome::NoCandidate), OpenSessionOutcome::FailedBeforeCandidate(error) => { - eprintln!( - "{}", - startup_failure_line( - &config.gemini_live_model, - room_name, - session_id, - error.as_ref(), - ) - ); return Ok(RoomRunOutcome::FailedBeforeCandidate(error)); } }; @@ -2691,6 +3032,7 @@ pub(crate) async fn run_room_with_slot( mut turn, started_at, restarts, + mut deferred_events, } = *session; let interview = InterviewContext { @@ -2732,9 +3074,8 @@ pub(crate) async fn run_room_with_slot( // // The grace is deliberate: the browser normally ends the interview itself, // and this only has to catch the case where it never does. - let hard_deadline = tokio::time::sleep_until( - (Instant::now() + interview_hard_deadline(boot.duration_min)).into(), - ); + let hard_deadline = + tokio::time::sleep_until((started_at + interview_hard_deadline(boot.duration_min)).into()); tokio::pin!(hard_deadline); let result: Result<(), Box> = async { @@ -2758,7 +3099,12 @@ pub(crate) async fn run_room_with_slot( ); on_watch_tick(&room, &mut context, &mut loops, interview).await? } - event = async { events.lock().await.recv().await } => { + event = async { + match deferred_events.pop_front() { + Some(event) => Some(event), + None => events.lock().await.recv().await, + } + } => { let Some(event) = event else { eprintln!( "LiveKit event stream ended for room={room_name}; ending with no report, \ diff --git a/tests/browser/dom-contract.test.js b/tests/browser/dom-contract.test.js index 3d8ef3cf..48038c5e 100644 --- a/tests/browser/dom-contract.test.js +++ b/tests/browser/dom-contract.test.js @@ -869,26 +869,6 @@ test("an interviewer reconnect banner is worded from the agent's notice", () => ); }); -// A degraded start looks identical whether the server has no LiveKit -// credentials, is at capacity, or the microphone failed to publish. The server -// writes candidate-facing text for exactly this; dropping it on the floor is -// what makes a refusal indistinguishable from the product working. -test("a refused interview says why instead of silently degrading", () => { - const script = interviewSource(); - const connect = functionBody(script, "connect"); - - assert.match( - connect, - /const reason =\s*strict &&[\s\S]*?:\s*failure\?\.message;/, - "other refusals retain the server's reason", - ); - assert.match(connect, /providerUiState\("degraded", reason\)/); - // The reason has to reach the candidate, not only the console. - const bannerAt = connect.indexOf('setBanner("connection", degraded.message)'); - const offlineAt = connect.indexOf("Offline mode is ready"); - assert.ok(bannerAt !== -1 && offlineAt !== -1 && bannerAt < offlineAt); -}); - // One banner element with several owners, ranked, so a new owner is a row here // rather than another boolean lock and another branch. test("the presence banner ranks its owners instead of racing them", () => { diff --git a/tests/browser/interview-lifecycle.test.js b/tests/browser/interview-lifecycle.test.js index 9b20b2f4..e4b67ac0 100644 --- a/tests/browser/interview-lifecycle.test.js +++ b/tests/browser/interview-lifecycle.test.js @@ -1,11 +1,18 @@ import { test } from "node:test"; import assert from "node:assert/strict"; +import { runInNewContext } from "node:vm"; +import { + createInterviewerReadiness, + interviewerReconnectMessage, + isAgent, + providerUiState, + withUnclearLines, +} from "../../web/lib.js"; import { createBeforeUnloadGuard, shouldWarnBeforeUnload, } from "../../web/before-unload.js"; import { reportIsPersisted } from "../../web/history.js"; -import { withUnclearLines } from "../../web/lib.js"; import { functionBody, read } from "./source.js"; function beforeUnloadEvent(target) { @@ -84,12 +91,23 @@ test("the unload listener can be disposed without leaving a stale handler", () = assert.equal(beforeUnloadEvent(target).defaultPrevented, false); }); +test("untimed work is protected until its report is persisted", () => { + const target = new CountingTarget(); + const guard = createBeforeUnloadGuard(target); + const lifecycle = { endsAt: 0, workStarted: true, reportPersisted: false }; + guard.setEnabled(shouldWarnBeforeUnload(lifecycle)); + assert.equal(beforeUnloadEvent(target).defaultPrevented, true); + lifecycle.reportPersisted = true; + guard.setEnabled(shouldWarnBeforeUnload(lifecycle)); + assert.equal(beforeUnloadEvent(target).defaultPrevented, false); +}); + test("interview wiring starts after preflight and only persistence releases it", () => { const page = read("web/interview.js"); const init = functionBody(page, "init"); const preflight = init.indexOf("const preflight = await runAudioCheck()"); const connect = init.indexOf("await connect(preflight, presenting)"); - const deadline = init.indexOf("state.endsAt = Date.now()"); + const deadline = init.indexOf("state.endsAt = startup.startedAt"); const guard = init.indexOf("updateBeforeUnloadGuard()", deadline); assert.ok(preflight !== -1 && preflight < connect); @@ -116,6 +134,341 @@ test("interview wiring starts after preflight and only persistence releases it", ); }); +test("startup notices settle once and replacement notices cannot start a timer", async () => { + let timeout; + let cancelled = 0; + let now = 10; + const gate = createInterviewerReadiness({ + now: () => now, + schedule: (fn) => { + timeout = fn; + return 1; + }, + cancel: () => { + cancelled += 1; + }, + }); + let settled = false; + gate.wait.then(() => { + settled = true; + }); + gate.receive({ type: "interviewer_state", reconnecting: false }); + gate.receive({ type: "other", startup: "ready" }); + await Promise.resolve(); + assert.equal(settled, false); + now = 30; + gate.receive({ type: "interviewer_state", startup: "ready" }); + now = 50; + gate.receive({ type: "interviewer_state", startup: "ready" }); + timeout(); + assert.deepEqual(await gate.wait, { startup: "ready", startedAt: 30 }); + assert.equal(cancelled, 1); + + const missing = createInterviewerReadiness({ + schedule: (fn) => { + timeout = fn; + return 2; + }, + cancel: () => {}, + }); + timeout(); + assert.equal((await missing.wait).startup, "failed"); +}); + +test("the actual connection preserves refusal reasons and gates the init deadline on readiness", async () => { + const page = read("web/interview.js"); + const init = functionBody(page, "init"); + const tail = init.slice(init.indexOf("const startup = await connect")); + for (const outcome of [ + "ready", + "failed", + "offline", + "refused", + "strict-refused", + "strict-timeout", + "joined-error", + "recording-failed", + "ended-ready", + "strict-failed", + ]) { + const strict = outcome.startsWith("strict-"); + const refused = outcome.includes("refused") || outcome === "strict-timeout"; + const serverReason = "The interview service is at capacity."; + const expectedReason = + outcome === "strict-timeout" + ? "Strict Interview connection timed out. Return to the lobby or retry media preflight." + : strict + ? serverReason + : "The live interview service is busy or rate limited."; + const delivered = []; + let gate; + let ticks = 0; + let intervals = 0; + let banner = ""; + let mediaStopped = 0; + let boardRefreshes = 0; + const warnings = []; + const context = { + protected: false, + state: { endsAt: 0, phase: "live" }, + board: { ready: false }, + exampleBoard: { + refresh: () => { + boardRefreshes += 1; + }, + }, + preflight: {}, + presenting: false, + createInterviewerReadiness: () => { + gate = createInterviewerReadiness({ + now: () => 1000, + schedule: () => 1, + cancel: () => {}, + }); + return gate; + }, + providerUiState, + interviewerReconnectMessage, + setAgentStateLabel() {}, + consentGiven: () => true, + recordConsent: async () => + outcome === "recording-failed" ? "interview-one" : null, + AbortSignal, + TOKEN_REQUEST_TIMEOUT_MS: 30_000, + fetch: async () => { + if (outcome === "offline") + throw new Error("Live session not configured"); + if (refused) + return { + ok: false, + status: outcome === "strict-timeout" ? 504 : 503, + json: async () => ({ error: serverReason, code: "capacity" }), + }; + return { + ok: true, + json: async () => ({ durationMin: 45, problemPage: "two-sum" }), + }; + }, + problem: { page: "two-sum" }, + strict, + previousAssignment: () => null, + rememberAssignment() {}, + loadProblem: async () => ({ page: "two-sum", starterCode: {} }), + loadJudge: async () => null, + renderProblem() {}, + initializeLanguages() {}, + initializeCandidateCases() {}, + updateRunAvailability() {}, + tabStorage: {}, + params: new URLSearchParams(), + durationMin: 45, + codingMinutes: 45, + behavioralMinutes: 0, + interviewLoop: "coding_only", + mode: "coding", + interviewProfile: {}, + interviewGrounding: null, + editorOptions: { execution: true }, + nodes: { + hideExamples: { checked: false }, + interviewerAccent: { value: "" }, + interviewerVoice: { value: "" }, + editor: { disabled: false }, + run: { disabled: false }, + audioStatus: { textContent: "" }, + audioCheck: { hidden: true, querySelectorAll: () => [] }, + audioLeave: {}, + audioJoin: { addEventListener() {} }, + end: { disabled: true }, + pause: { disabled: true }, + withdrawConsent: {}, + endingWithdrawConsent: {}, + }, + applyGrantedDuration() {}, + connectLiveKit: async () => { + context.state.joinedRoom = true; + if (outcome === "joined-error") + throw new Error("Integrity publication failed"); + }, + recordingEnabled: outcome === "recording-failed", + startRecording: () => new Promise(() => {}), + recordReplay() {}, + recordStage() {}, + currentCode: () => "", + exampleDrawingEnabled: true, + whiteboard: false, + stopPreflight() {}, + stopLocalMedia: () => { + mediaStopped += 1; + }, + setBanner: (_source, text) => { + banner = text; + delivered.push("banner"); + }, + addTranscript() { + delivered.push("offline"); + }, + publishCode() {}, + updateBeforeUnloadGuard() { + context.protected = shouldWarnBeforeUnload(context.state); + }, + tickTimer: () => { + ticks += 1; + }, + setInterval: () => { + intervals += 1; + }, + console: { warn: (_message, error) => warnings.push(error?.message) }, + }; + const running = runInNewContext( + `${functionBody(page, "connect")}\n}\n(async () => {${tail}\n})()`, + context, + ); + await new Promise((resolve) => setImmediate(resolve)); + if (!refused && outcome !== "offline" && outcome !== "joined-error") { + assert.equal(context.state.endsAt, 0); + assert.equal(ticks, 0); + assert.equal( + context.board.ready, + false, + `${outcome}: ${warnings.join(", ")}`, + ); + assert.equal(boardRefreshes, 0); + assert.equal( + context.protected, + true, + `${outcome}: work is protected before readiness`, + ); + if (outcome === "ended-ready") context.state.phase = "ending"; + gate.receive({ + type: "interviewer_state", + startup: outcome.endsWith("ready") ? "ready" : "failed", + reason: "billing", + }); + } + await running; + if (outcome === "ended-ready") { + assert.equal(context.state.endsAt, 0); + assert.equal(intervals, 0); + assert.equal(context.board.ready, false); + assert.equal(banner, ""); + continue; + } + if (outcome === "strict-failed") { + assert.match(banner, /billing/); + assert.doesNotMatch(banner, /work the problem offline/); + assert.equal(context.nodes.editor.disabled, true); + assert.equal(context.nodes.run.disabled, true); + assert.equal(context.state.endsAt, 0); + continue; + } + assert.equal( + context.protected, + !strict, + `${outcome}: editable work stays protected`, + ); + if (refused) { + assert.ok(banner.includes(expectedReason), `${outcome}: ${banner}`); + assert.equal(mediaStopped, 1); + assert.deepEqual(delivered, strict ? ["banner"] : ["banner", "offline"]); + if (strict) { + assert.equal(context.nodes.editor.disabled, true); + assert.equal(context.nodes.run.disabled, true); + assert.equal(context.nodes.audioStatus.textContent, expectedReason); + assert.equal(context.state.endsAt, 0); + assert.equal(ticks, 0); + assert.equal(intervals, 0); + assert.equal(context.board.ready, false); + continue; + } + } + assert.equal(context.board.ready, true); + assert.equal(boardRefreshes, 1); + if (outcome === "ready") { + assert.equal(context.state.endsAt, 1000 + 45 * 60 * 1000); + assert.equal(ticks, 1); + assert.equal(intervals, 1); + } else if (outcome === "offline" || outcome === "refused") { + assert.ok(context.state.endsAt > 0); + assert.equal(ticks, 1); + assert.equal(intervals, 1); + } else { + assert.equal(context.state.endsAt, 0); + assert.equal(ticks, 0); + assert.equal(intervals, 0); + assert.match( + banner, + outcome === "joined-error" + ? /provider could not be reached/ + : /billing/, + ); + assert.equal(mediaStopped, 1); + } + } +}); + +test("only the pinned interviewer's departure fails startup readiness", async () => { + const source = functionBody(read("web/interview.js"), "connectLiveKit"); + const handlerSource = + /room\.on\(livekit\.RoomEvent\.ParticipantDisconnected,\s*(\(participant\) => \{[\s\S]*?\n \})\);/.exec( + source, + )?.[1]; + assert.ok(handlerSource, "the participant departure handler exists"); + const gate = createInterviewerReadiness({ + schedule: () => 1, + cancel: () => {}, + }); + const state = { + agentIdentity: "interviewer-primary", + interviewerReadiness: gate, + }; + let updates = 0; + const handler = runInNewContext(`(${handlerSource})`, { + state, + isAgent, + updateAgentState: () => { + updates += 1; + }, + }); + let settled = false; + gate.wait.then(() => { + settled = true; + }); + handler({ identity: "interviewer-other", kind: "AGENT" }); + handler({ identity: "candidate" }); + await Promise.resolve(); + assert.equal(settled, false); + handler({ identity: state.agentIdentity, kind: "AGENT" }); + assert.equal((await gate.wait).startup, "failed"); + assert.equal(updates, 3); +}); + +test("an agent departing before identity binding fails readiness, but a candidate does not", async () => { + const source = functionBody(read("web/interview.js"), "connectLiveKit"); + const handlerSource = + /room\.on\(livekit\.RoomEvent\.ParticipantDisconnected,\s*(\(participant\) => \{[\s\S]*?\n \})\);/.exec( + source, + )?.[1]; + assert.ok(handlerSource); + const gate = createInterviewerReadiness({ schedule: () => 1, cancel() {} }); + const handler = runInNewContext(`(${handlerSource})`, { + state: { agentIdentity: null, interviewerReadiness: gate }, + isAgent, + updateAgentState() {}, + }); + let settled = false; + gate.wait.then(() => { + settled = true; + }); + handler({ identity: "candidate", kind: "STANDARD" }); + await Promise.resolve(); + assert.equal(settled, false); + handler({ identity: "agent", kind: "AGENT" }); + await Promise.resolve(); + assert.equal(settled, true); + assert.equal((await gate.wait).startup, "failed"); +}); + test("LiveKit stays connected while the native leave warning is pending", () => { const connect = functionBody(read("web/interview.js"), "connectLiveKit"); const roomOptions = /new livekit\.Room\(\{([\s\S]*?)\}\)/.exec(connect)?.[1]; diff --git a/tests/browser/interview-startup.test.js b/tests/browser/interview-startup.test.js new file mode 100644 index 00000000..ab484e39 --- /dev/null +++ b/tests/browser/interview-startup.test.js @@ -0,0 +1,928 @@ +import { after, before, test } from "node:test"; +import assert from "node:assert/strict"; +import { fileURLToPath } from "node:url"; +import { topics } from "../../web/lib.js"; +import { DEFAULT_RUNTIME_CONFIG, startStaticServer } from "./source.js"; + +let browser; +let server; +let base; +before(async () => { + try { + const { chromium } = await import("playwright"); + browser = await chromium.launch({ + args: [ + "--use-fake-device-for-media-stream", + "--use-fake-ui-for-media-stream", + `--use-file-for-fake-audio-capture=${fileURLToPath(new URL("../../web/audio/test-voice.wav", import.meta.url))}`, + ], + }); + } catch (error) { + if (process.env.CI) throw error; + return; + } + ({ server, base } = await startStaticServer({ + runtimeConfig: DEFAULT_RUNTIME_CONFIG, + })); +}); +after(async () => { + await browser?.close(); + await new Promise((closed) => (server ? server.close(closed) : closed())); +}); + +// Delivery and teardown are released by the test, not by a timer. The page +// still registers its real callbacks and crosses the real media preflight. +function sdkFixture() { + const agent = { identity: "agent-startup", kind: "AGENT", attributes: {} }; + const other = { identity: "agent-other", kind: "AGENT", attributes: {} }; + class Room { + constructor() { + this.handlers = new Map(); + this.remoteParticipants = new Map([ + [agent.identity, agent], + [other.identity, other], + ]); + this.disconnectCalls = 0; + this.published = []; + this.receipts = []; + this.localParticipant = { + identity: "candidate-startup", + publishTrack: async () => {}, + publishData: async (data, options) => { + this.published.push(data); + const message = JSON.parse(new TextDecoder().decode(data)); + if (message.type === "startup_received") { + this.receipts.push({ message, options }); + if (this.holdReceipt) + await new Promise((resolve) => { + this.releaseReceipt = resolve; + }); + } + }, + }; + window.testRoom = this; + } + on(event, callback) { + const callbacks = this.handlers.get(event) || []; + callbacks.push(callback); + this.handlers.set(event, callbacks); + return this; + } + off(event, callback) { + this.handlers.set( + event, + (this.handlers.get(event) || []).filter( + (handler) => handler !== callback, + ), + ); + return this; + } + async connect() { + if (window.testDepartDuringConnect) await this.depart(); + } + async disconnect() { + this.disconnectCalls += 1; + } + emit(event, ...args) { + return Promise.all( + (this.handlers.get(event) || []).map((callback) => callback(...args)), + ); + } + startup(message, topic, identity = agent.identity) { + return this.emit( + "DataReceived", + new TextEncoder().encode(JSON.stringify(message)), + this.remoteParticipants.get(identity), + undefined, + topic, + ); + } + depart(identity = agent.identity) { + const participant = this.remoteParticipants.get(identity); + this.remoteParticipants.delete(identity); + return this.emit("ParticipantDisconnected", participant); + } + } + window.LivekitClient = { + Room, + RoomEvent: new Proxy({}, { get: (_target, name) => name }), + Track: { Source: { Microphone: "microphone", Camera: "camera" } }, + }; +} + +async function joinedPage( + t, + { + clock = false, + departDuringConnect = false, + recording = false, + strict = false, + setup = async () => {}, + } = {}, +) { + let page; + let phase = "opening page"; + const errors = []; + const events = []; + t.after(async () => { + try { + if (t.passed === false || t.signal?.aborted || errors.length) { + let deadline; + try { + const snapshot = page + ? page + .evaluate(() => ({ + controls: Object.fromEntries( + [ + "audio-check", + "audio-check-status", + "audio-check-join", + "audio-step-output", + "audio-step-mic", + "audio-step-camera", + "editor", + "end", + "pause", + "timer", + "ending-overlay", + "presence-banner", + ].map((id) => { + const node = document.getElementById(id); + return [ + id, + node && { + hidden: node.hidden, + disabled: node.disabled, + classes: node.className, + text: node.matches("textarea") + ? undefined + : node.textContent.slice(0, 250), + }, + ]; + }), + ), + captures: window.testCaptures, + tracks: window.testMediaTracks?.map((track) => ({ + kind: track.kind, + state: track.readyState, + enabled: track.enabled, + muted: track.muted, + })), + room: window.testRoom && { + disconnects: window.testRoom.disconnectCalls, + published: window.testRoom.published.length, + receipts: window.testRoom.receipts.length, + }, + })) + .catch((error) => ({ unavailable: error.message })) + : Promise.resolve(null); + const dom = await Promise.race([ + snapshot, + new Promise((resolve) => { + deadline = setTimeout( + () => resolve({ unavailable: "snapshot exceeded 2 seconds" }), + 2000, + ); + }), + ]); + t.diagnostic(JSON.stringify({ phase, errors, events, dom })); + } finally { + clearTimeout(deadline); + } + } + assert.deepEqual(errors, []); + } finally { + await page?.close(); + } + }); + page = await browser.newPage(); + page.on("pageerror", (error) => errors.push(error.message)); + const note = (event) => { + events.push(event); + if (events.length > 10) events.shift(); + }; + page.on("console", (message) => { + if (["error", "warning"].includes(message.type())) + note({ console: message.text().slice(0, 250) }); + }); + page.on("requestfailed", (request) => + note({ + path: new URL(request.url()).pathname, + failure: request.failure()?.errorText, + }), + ); + const clockStart = new Date("2030-01-01T00:00:00Z"); + if (clock) await page.clock.install({ time: clockStart }); + await page.addInitScript( + ({ departDuringConnect }) => { + window.testDepartDuringConnect = departDuringConnect; + // Keep native capture and preflight, without depending on downloaded + // vision models while exercising startup event ordering. + window.FaceDetector = class { + async detect() { + return [{}]; + } + }; + window.testMediaTracks = []; + window.testCaptures = []; + const capture = navigator.mediaDevices.getUserMedia.bind( + navigator.mediaDevices, + ); + navigator.mediaDevices.getUserMedia = async (...args) => { + const entry = { + audio: Boolean(args[0]?.audio), + video: Boolean(args[0]?.video), + state: "pending", + }; + window.testCaptures.push(entry); + try { + const stream = await capture(...args); + entry.state = "resolved"; + window.testMediaTracks.push(...stream.getTracks()); + return stream; + } catch (error) { + entry.state = error.name; + throw error; + } + }; + }, + { departDuringConnect }, + ); + if (recording) { + await page.route("**/runtime-config.js", (route) => + route.fulfill({ + contentType: "text/javascript", + body: `${DEFAULT_RUNTIME_CONFIG}\nglobalThis.CODETRIAL_RECORDING_ENABLED = true;\nglobalThis.CODETRIAL_CONSENT_VERSION = "fixture";`, + }), + ); + await page.route("**/api/interviews", (route) => + route.fulfill({ + json: { interviewId: "startup-recording" }, + }), + ); + await page.route("**/api/interviews/startup-recording/replay", (route) => + route.fulfill({ json: {} }), + ); + } + await page.route("**/vendor/livekit-client.js", (route) => + route.fulfill({ + contentType: "text/javascript", + body: `(${sdkFixture.toString()})();`, + }), + ); + // Readiness must also work in voice-only mode. Loading a VRM and software + // WebGL here made Join clicks compete with rendering unrelated to this suite. + await page.route("**/avatar/model.js", (route) => + route.fulfill({ + contentType: "text/javascript", + body: `export async function loadAvatarModel() { + throw new Error("Startup fixture uses voice-only mode"); + }`, + }), + ); + await page.route("**/api/token", (route) => + route.fulfill({ + contentType: "application/json", + body: JSON.stringify({ + serverUrl: "wss://example.invalid", + token: "fixture-token", + durationMin: 45, + ...(strict + ? { + problemPage: "chargeback-pair-match", + previousAssignment: "fixture-assignment", + } + : {}), + }), + }), + ); + await setup(page); + phase = "loading interview"; + await page.goto( + `${base}/interview.html?${strict ? "selection=strict" : "problem=chargeback-pair-match"}&duration=45`, + ); + phase = "media preflight"; + if (!recording) await page.locator("#camera-skip").click(); + await page.locator("#audio-heard").click(); + if (recording) await page.locator("#recording-consent").check(); + phase = "waiting for Join readiness"; + await page.waitForFunction( + () => !document.querySelector("#audio-check-join").disabled, + ); + phase = "clicking Join"; + await page.locator("#audio-check-join").click(); + phase = "waiting for room publication"; + await page.waitForFunction(() => window.testRoom?.published.length > 0); + phase = "startup scenario after room publication"; + return page; +} + +async function warnsOnLeave(page) { + return page.evaluate(() => { + const event = new Event("beforeunload", { cancelable: true }); + window.dispatchEvent(event); + return event.defaultPrevented; + }); +} + +async function assertPendingControls(page) { + assert.equal(await page.locator("#end").isDisabled(), true); + assert.equal(await page.locator("#pause").isDisabled(), true); + const result = await page.evaluate(() => { + document.querySelector("#end").click(); + document.querySelector("#pause").click(); + return { + ending: !document.querySelector("#ending-overlay").hidden, + paused: document.querySelector("#pause").textContent.trim() !== "Pause", + endRequests: window.testRoom.published + .map((bytes) => JSON.parse(new TextDecoder().decode(bytes))) + .filter((message) => message.type === "end_interview").length, + }; + }); + assert.deepEqual(result, { ending: false, paused: false, endRequests: 0 }); +} + +test( + "Strict startup failure does not promise offline editing or start the clock", + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t, { strict: true }); + await assertPendingControls(page); + const timer = await page.locator("#timer").textContent(); + await page.locator("#editor").fill("// Work while waiting for readiness"); + assert.equal(await warnsOnLeave(page), true); + await page.evaluate( + (topic) => + window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: "failed", + reason: "billing", + }, + topic, + ), + topics.control, + ); + await page.waitForFunction(() => window.testRoom.disconnectCalls === 1); + await page.waitForFunction( + () => !document.querySelector("#audio-check").hidden, + ); + const message = await page.locator("#audio-check-status").textContent(); + assert.match(message, /billing/); + assert.match(message, /lobby or retry media preflight/); + assert.doesNotMatch(message, /work the problem offline/); + assert.equal(await page.locator("#editor").isDisabled(), true); + assert.equal(await page.locator("#run-tests").isDisabled(), true); + assert.equal(await page.locator("#end").isDisabled(), true); + assert.equal(await page.locator("#pause").isDisabled(), true); + assert.equal(await page.locator("#timer").textContent(), timer); + assert.equal(await warnsOnLeave(page), true); + }, +); + +test( + "editable startup work and untimed offline work both retain leave protection", + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t); + await assertPendingControls(page); + await page.locator("#editor").fill("// Work written before Gemini answers"); + assert.equal(await warnsOnLeave(page), true); + const timer = await page.locator("#timer").textContent(); + await page.evaluate( + (topic) => + window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: "failed", + reason: "billing", + }, + topic, + ), + topics.control, + ); + await page.waitForFunction(() => window.testRoom.disconnectCalls === 1); + await page.locator("#editor").fill("// Still working offline"); + assert.equal(await warnsOnLeave(page), true); + assert.equal(await page.locator("#timer").textContent(), timer); + assert.equal(await page.locator("#end").isDisabled(), false); + assert.equal(await page.locator("#pause").isDisabled(), false); + assert.equal(await page.locator("#ending-overlay").isVisible(), false); + }, +); + +test( + "departure before the room is assigned ends readiness without waiting for its timeout", + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t, { departDuringConnect: true }); + await page.waitForFunction(() => window.testRoom.disconnectCalls === 1); + assert.match( + await page.locator("#presence-banner").textContent(), + /timer has not started/, + ); + assert.equal(await warnsOnLeave(page), true); + const media = await page.evaluate(() => + window.testMediaTracks.map((track) => track.readyState), + ); + assert.ok(media.length > 0); + assert.ok(media.every((ready) => ready === "ended")); + }, +); + +test( + "startup failure releases local media while the recording response is still pending", + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + let recordingRoute; + let requested; + const recordingRequested = new Promise((resolve) => { + requested = resolve; + }); + const page = await joinedPage(t, { + recording: true, + setup: async (page) => { + await page.route( + "**/api/interviews/startup-recording/recording", + (route) => { + recordingRoute = route; + requested(); + }, + ); + }, + }); + await recordingRequested; + await page.evaluate( + (topic) => + window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: "failed", + reason: "quota", + }, + topic, + ), + topics.control, + ); + await page.waitForFunction(() => window.testRoom.disconnectCalls === 1); + assert.match(await page.locator("#presence-banner").textContent(), /quota/); + const media = await page.evaluate(() => + window.testMediaTracks.map((track) => track.readyState), + ); + assert.ok(media.length > 0); + assert.ok(media.every((ready) => ready === "ended")); + assert.equal(await warnsOnLeave(page), true); + await recordingRoute.fulfill({ json: { state: "ready" } }); + await page.waitForFunction(() => + document.querySelector("#recording-state").textContent.includes("saved"), + ); + assert.match(await page.locator("#presence-banner").textContent(), /quota/); + assert.equal(await page.evaluate(() => window.testRoom.disconnectCalls), 1); + }, +); + +test("the vendored SDK delivers a post-departure packet without its sender", async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await browser.newPage(); + t.after(() => page.close()); + await page.goto( + `${base}/interview.html?problem=chargeback-pair-match&duration=45`, + ); + const events = await page.evaluate(async () => { + const sdk = window.LivekitClient; + const room = new sdk.Room(); + const agent = { + identity: "agent-ordering", + trackPublications: new Map(), + emit() {}, + setDisconnected() {}, + }; + room.remoteParticipants.set(agent.identity, agent); + const seen = []; + room.on(sdk.RoomEvent.DataReceived, (_payload, sender) => + seen.push({ event: "data", sender: sender?.identity ?? null }), + ); + room.on(sdk.RoomEvent.ParticipantDisconnected, (sender) => + seen.push({ event: "departure", sender: sender.identity }), + ); + const packet = { + participantIdentity: agent.identity, + kind: 0, + value: { + case: "user", + value: { payload: new Uint8Array(), topic: "control" }, + }, + }; + room.handleDataPacket(packet); + room.handleParticipantDisconnected(agent.identity, agent); + room.handleDataPacket(packet); + await room.disconnect(); + return seen; + }); + assert.deepEqual(events, [ + { event: "data", sender: "agent-ordering" }, + { event: "departure", sender: "agent-ordering" }, + { event: "data", sender: null }, + ]); +}); + +for (const reason of ["billing", "quota", "unavailable"]) { + test( + `offline practice can pause and resume after startup ${reason}`, + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t, { clock: true }); + const waitingTimer = await page.locator("#timer").textContent(); + await page.evaluate( + ({ reason, topic }) => + window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: "failed", + reason, + }, + topic, + ), + { reason, topic: topics.control }, + ); + await page.waitForFunction(() => window.testRoom.disconnectCalls === 1); + const banner = await page.locator("#presence-banner").textContent(); + assert.match(banner, /timer has not started/); + + await page.locator("#pause").click(); + await page.waitForFunction( + () => document.querySelector("#editor").disabled, + ); + assert.equal(await page.locator("#ending-overlay").isVisible(), false); + await page.clock.fastForward(46 * 60 * 1000); + await page.locator("#pause").click(); + await page.waitForFunction( + () => !document.querySelector("#editor").disabled, + ); + await page.locator("#editor").fill("// Offline practice after resume"); + assert.equal(await page.locator("#pause").textContent(), "Pause"); + assert.equal(await page.locator("#ending-overlay").isVisible(), false); + assert.equal(await page.locator("#timer").textContent(), waitingTimer); + assert.equal( + await page.locator("#presence-banner").textContent(), + banner, + ); + assert.equal( + await page.evaluate(() => window.testRoom.disconnectCalls), + 1, + ); + }, + ); +} + +test( + "a ready interview counts down and expires even while paused", + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t, { clock: true }); + // Let media preflight run before freezing wall-clock drift. Readiness and + // interview time then advance only when the test moves the clock. + await page.clock.pauseAt(new Date("2030-01-01T00:01:00Z")); + await assertPendingControls(page); + await page.evaluate( + (topic) => + window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: "ready", + }, + topic, + ), + topics.control, + ); + await page.waitForFunction( + () => document.querySelector("#timer").textContent === "45:00", + ); + assert.equal(await page.locator("#end").isDisabled(), false); + assert.equal(await page.locator("#pause").isDisabled(), false); + await page.clock.fastForward(60_000); + assert.equal(await page.locator("#timer").textContent(), "44:00"); + await page.locator("#pause").click(); + await page.evaluate( + (topic) => + window.testRoom.startup({ type: "pause_state", paused: true }, topic), + topics.control, + ); + await page.waitForFunction( + () => document.querySelector("#editor").disabled, + ); + await page.clock.fastForward(60_000); + assert.equal(await page.locator("#timer").textContent(), "43:00"); + assert.equal(await page.locator("#ending-overlay").isVisible(), false); + await page.clock.fastForward(43 * 60 * 1000); + await page.waitForFunction( + () => !document.querySelector("#ending-overlay").hidden, + ); + const endings = await page.evaluate(() => + window.testRoom.published + .map((bytes) => JSON.parse(new TextDecoder().decode(bytes))) + .filter((message) => message.type === "end_interview"), + ); + assert.equal(endings.length, 1); + assert.equal(endings[0].reason, "time_up"); + assert.equal(await page.locator("#timer").textContent(), "00:00"); + }, +); + +for (const reason of ["billing", "quota"]) { + test( + `a received ${reason} failure survives departure while its receipt is flushing`, + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t); + const waitingTimer = await page.locator("#timer").textContent(); + await page.evaluate( + ({ reason, topic }) => { + window.testRoom.holdReceipt = true; + return window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: "failed", + reason, + }, + topic, + ); + }, + { reason, topic: topics.control }, + ); + await page.waitForFunction(() => !!window.testRoom.releaseReceipt); + assert.equal( + await page.evaluate(() => window.testRoom.disconnectCalls), + 0, + "cleanup must give the receipt its bounded chance", + ); + const receipt = await page.evaluate(() => window.testRoom.receipts[0]); + assert.equal(receipt.options.topic, topics.control); + assert.equal(receipt.options.reliable, true); + assert.deepEqual(receipt.options.destinationIdentities, [ + "agent-startup", + ]); + await page.evaluate(() => window.testRoom.depart()); + await page.evaluate(() => window.testRoom.releaseReceipt()); + await page.waitForFunction(() => window.testRoom.disconnectCalls === 1); + const banner = await page.locator("#presence-banner").textContent(); + assert.match(banner, new RegExp(reason)); + assert.match(banner, /timer has not started/); + assert.equal(await page.locator("#timer").textContent(), waitingTimer); + await page.evaluate(() => window.testRoom.emit("Disconnected")); + assert.equal( + await page.locator("#presence-banner").textContent(), + banner, + ); + assert.equal( + await page.evaluate(() => window.testRoom.disconnectCalls), + 1, + ); + }, + ); +} + +test( + "a permanently stalled startup receipt cannot hold cleanup", + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t, { clock: true }); + const waitingTimer = await page.locator("#timer").textContent(); + await page.evaluate((topic) => { + window.testRoom.holdReceipt = true; + return window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: "failed", + reason: "billing", + }, + topic, + ); + }, topics.control); + await page.waitForFunction(() => !!window.testRoom.releaseReceipt); + assert.equal(await page.evaluate(() => window.testRoom.disconnectCalls), 0); + await page.clock.fastForward(1001); + await page.waitForFunction(() => window.testRoom.disconnectCalls === 1); + assert.match( + await page.locator("#presence-banner").textContent(), + /billing/, + ); + assert.equal(await page.locator("#timer").textContent(), waitingTimer); + await page.evaluate(() => window.testRoom.releaseReceipt()); + await page.clock.fastForward(1000); + assert.equal(await page.evaluate(() => window.testRoom.disconnectCalls), 1); + }, +); + +for (const late of [false, true]) { + test( + `departure with ${late ? "an unattributed late" : "no"} failure notice ends startup once`, + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t); + const waitingTimer = await page.locator("#timer").textContent(); + await page.evaluate(() => window.testRoom.depart()); + if (late) + await page.evaluate( + (topic) => + window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: "failed", + reason: "billing", + }, + topic, + ), + topics.control, + ); + await page.waitForFunction(() => window.testRoom.disconnectCalls === 1); + const banner = await page.locator("#presence-banner").textContent(); + assert.match(banner, /timer has not started/); + assert.doesNotMatch(banner, /billing|quota/); + assert.equal(await page.locator("#timer").textContent(), waitingTimer); + assert.equal( + await page.evaluate(() => window.testRoom.receipts.length), + 0, + ); + await page.evaluate(() => window.testRoom.emit("Disconnected")); + assert.equal( + await page.locator("#presence-banner").textContent(), + banner, + ); + assert.equal( + await page.evaluate(() => window.testRoom.disconnectCalls), + 1, + ); + }, + ); +} + +test( + "a permanently missing startup notice expires without a countdown", + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t, { clock: true }); + const waitingTimer = await page.locator("#timer").textContent(); + await page.clock.fastForward(180_000); + await page.waitForFunction(() => window.testRoom.disconnectCalls === 1); + assert.match( + await page.locator("#presence-banner").textContent(), + /timer has not started/, + ); + assert.equal(await page.locator("#timer").textContent(), waitingTimer); + await page.clock.fastForward(180_000); + assert.equal(await page.evaluate(() => window.testRoom.disconnectCalls), 1); + }, +); + +for (const startup of ["ready", "failed"]) { + test( + `the other agent's startup ${startup} cannot settle the pinned interviewer's gate`, + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t); + const waitingTimer = await page.locator("#timer").textContent(); + await page.evaluate( + ({ startup, topic }) => + window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup, + reason: "billing", + }, + topic, + "agent-other", + ), + { startup, topic: topics.control }, + ); + assert.equal( + await page.evaluate(() => window.testRoom.disconnectCalls), + 0, + ); + assert.equal(await page.locator("#timer").textContent(), waitingTimer); + await page.evaluate( + ({ startup, topic }) => + window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: startup === "ready" ? "failed" : "ready", + reason: "quota", + }, + topic, + ), + { startup, topic: topics.control }, + ); + if (startup === "ready") { + await page.waitForFunction(() => window.testRoom.disconnectCalls === 1); + assert.match( + await page.locator("#presence-banner").textContent(), + /quota/, + ); + assert.equal(await page.locator("#timer").textContent(), waitingTimer); + } else { + await page.waitForFunction( + (waiting) => document.querySelector("#timer").textContent !== waiting, + waitingTimer, + ); + assert.equal( + await page.evaluate(() => window.testRoom.disconnectCalls), + 0, + ); + } + }, + ); +} + +for (const reason of ["billing", "quota"]) { + test( + `a delayed startup disconnect preserves the ${reason} banner`, + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t); + await page.evaluate( + ({ reason, topic }) => + window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: "failed", + reason, + }, + topic, + ), + { reason, topic: topics.control }, + ); + await page.waitForFunction( + (reason) => + document + .querySelector("#presence-banner") + .textContent.includes(reason), + reason, + ); + const before = await page.locator("#presence-banner").textContent(); + assert.equal( + await page.evaluate(() => window.testRoom.disconnectCalls), + 1, + ); + await page.evaluate(() => window.testRoom.emit("Disconnected")); + assert.equal( + await page.locator("#presence-banner").textContent(), + before, + ); + assert.match(before, /timer has not started/); + }, + ); +} + +test( + "a real disconnect after readiness keeps the connection-drop banner", + { timeout: 15_000 }, + async (t) => { + if (!browser) return t.skip("playwright chromium unavailable"); + const page = await joinedPage(t); + const waitingTimer = await page.locator("#timer").textContent(); + await page.evaluate( + (topic) => + window.testRoom.startup( + { + type: "interviewer_state", + reconnecting: false, + startup: "ready", + }, + topic, + ), + topics.control, + ); + await page.waitForFunction( + (waiting) => document.querySelector("#timer").textContent !== waiting, + waitingTimer, + ); + await page.evaluate(() => window.testRoom.emit("Disconnected")); + assert.match( + await page.locator("#presence-banner").textContent(), + /provider could not be reached/, + ); + }, +); diff --git a/tests/browser/lib.test.js b/tests/browser/lib.test.js index 59368243..2957a989 100644 --- a/tests/browser/lib.test.js +++ b/tests/browser/lib.test.js @@ -3621,6 +3621,32 @@ test("interviewer reconnect reasons use bounded wording and optional estimates", } }); +test("startup errors name only classified provider reasons", () => { + for (const [reason, expected] of [ + ["billing", /billing or credit/], + ["quota", /quota or rate limit/], + ["credentials", /API credentials/], + ["refused", /refused this Live request/], + ]) { + const text = interviewerReconnectMessage({ startup: "failed", reason }); + assert.match(text, expected); + assert.match(text, /timer has not started/); + assert.doesNotMatch(text, /Retrying in/); + } + for (const reason of [ + undefined, + "1011", + "503", + "secret provider body", + "__proto__", + "toString", + ]) { + const text = interviewerReconnectMessage({ startup: "failed", reason }); + assert.match(text, /did not answer/); + assert.doesNotMatch(text, /billing|quota|secret provider body/); + } +}); + // Each side was tested against a hand-written copy of the other, so a reason // renamed in either place passed both and showed the generic banner. test("every reconnect reason the agent can send has its own wording", () => { diff --git a/tests/browser/report-recovery.test.js b/tests/browser/report-recovery.test.js index b4949d6c..eaebac5e 100644 --- a/tests/browser/report-recovery.test.js +++ b/tests/browser/report-recovery.test.js @@ -479,6 +479,7 @@ test("terminal room disconnect finalizes recovery before clearing the room", asy "room.on(livekit.RoomEvent.Disconnected, () => {".length; const body = source.slice(start, source.indexOf("\n });", start)); page.scope.reportRecovery = page.reportRecovery; + page.scope.room = page.room; page.scope.console = { warn() {} }; const playbackRooms = []; page.scope.room = page.room; diff --git a/tests/browser/restore-starter.test.js b/tests/browser/restore-starter.test.js index a3c7ea0b..a16235ed 100644 --- a/tests/browser/restore-starter.test.js +++ b/tests/browser/restore-starter.test.js @@ -72,7 +72,8 @@ window.LivekitClient = { Track: { Source: { Microphone: "microphone", Camera: "camera" } }, Room: class { constructor() { - this.remoteParticipants = new Map(); + const agent = { identity: "agent-fake-room", kind: "AGENT", attributes: {} }; + this.remoteParticipants = new Map([[agent.identity, agent]]); this.name = "fake-room"; this.localParticipant = { identity: "candidate", @@ -82,6 +83,16 @@ window.LivekitClient = { topic: options.topic, payload: JSON.parse(new TextDecoder().decode(bytes)), }); + // The first integrity publication follows pinning the interviewer. + if (options.topic === "integrity" && !this.ready) { + this.ready = true; + window.__handlers.DataReceived( + new TextEncoder().encode(JSON.stringify({ + type: "interviewer_state", reconnecting: false, startup: "ready", + })), + agent, undefined, "control", + ); + } }, streamBytes: async () => ({ write: async () => {}, close: async () => {} }), }; diff --git a/tests/browser/strict-selection.test.js b/tests/browser/strict-selection.test.js index 85daba61..caa5cf21 100644 --- a/tests/browser/strict-selection.test.js +++ b/tests/browser/strict-selection.test.js @@ -90,10 +90,23 @@ function mediaAndRoom() { window.LivekitClient = { RoomEvent: new Proxy({}, { get: (_, key) => key }), Room: class { - remoteParticipants = new Map(); + agent = { identity: "agent-test", kind: "AGENT", attributes: {} }; + remoteParticipants = new Map([[this.agent.identity, this.agent]]); localParticipant = { publishTrack: async () => {}, - publishData: async () => {}, + publishData: async (_bytes, options) => { + if (options.topic !== "integrity" || this.ready) return; + this.ready = true; + const payload = new TextEncoder().encode( + JSON.stringify({ + type: "interviewer_state", + reconnecting: false, + startup: "ready", + }), + ); + for (const handler of window.roomEvents.DataReceived ?? []) + handler(payload, this.agent, undefined, "control"); + }, }; on(event, callback) { (window.roomEvents[event] ??= []).push(callback); diff --git a/tests/fixtures/startup-receipt.json b/tests/fixtures/startup-receipt.json new file mode 100644 index 00000000..5953583e --- /dev/null +++ b/tests/fixtures/startup-receipt.json @@ -0,0 +1,12 @@ +{ + "topic": "control", + "cases": [ + { + "name": "received startup failure", + "payload": { + "type": "startup_received", + "deliveryId": "2294ce784e0250bb254239cbf01f1cadc64d5ce20dbdeaba8ef1c6105b7b33b7" + } + } + ] +} diff --git a/tests/unit/gemini.rs b/tests/unit/gemini.rs index 9a970912..9f74ccd7 100644 --- a/tests/unit/gemini.rs +++ b/tests/unit/gemini.rs @@ -344,6 +344,48 @@ async fn report_still_runs_after_live_quota_exhaustion() { assert_eq!(remaining, MAX_REPORT_HTTP_ATTEMPTS - 1); } +#[test] +fn startup_failure_identifies_exhausted_rotations() { + for (prefix, failures, expected) in [ + ( + "startup-exhausted-quota", + [CredentialFailure::Quota, CredentialFailure::Quota], + "quota", + ), + ( + "startup-exhausted-billing", + [CredentialFailure::Billing, CredentialFailure::Billing], + "billing", + ), + ( + "startup-exhausted-mixed", + [CredentialFailure::Quota, CredentialFailure::Billing], + "billing", + ), + ( + "startup-exhausted-refused", + [CredentialFailure::Invalid, CredentialFailure::Refused], + "unavailable", + ), + ] { + let first = format!("{prefix}-first"); + let second = format!("{prefix}-second"); + let configured = format!("{first},{second}"); + let config = live_config(&[("GOOGLE_API_KEYS", &configured)]); + let keys = GeminiKeys::from_config(&config); + for (key, failure) in [(&first, failures[0]), (&second, failures[1])] { + keys.failed(key, failure, ApiSurface::Live); + } + let error = keys.select().unwrap_err(); + assert_eq!(credential_failure(&error), None); + assert_eq!(startup_failure_reason(&error), expected, "{prefix}"); + } + assert_eq!( + startup_failure_reason(&io::Error::other("1011 Internal error")), + "unavailable" + ); +} + #[tokio::test] async fn report_failover_reuses_the_existing_call_budget() { let (result, seen, remaining) = @@ -2941,6 +2983,208 @@ async fn open_live_session_sends_setup_and_waits_for_setup_complete() { session.close().await.unwrap(); } +#[tokio::test] +async fn startup_go_away_is_replayed_before_output_with_original_timestamps() { + let config = live_config(&[]); + let boot = bootstrap(&config, "interview-fixed", Some("two-sum"), 45); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut socket = accept_async(socket).await.unwrap(); + socket.next().await.unwrap().unwrap(); + socket + .send(Message::Text(r#"{"setupComplete":{}}"#.into())) + .await + .unwrap(); + let request = socket.next().await.unwrap().unwrap(); + let request: Value = serde_json::from_str(request.to_text().unwrap()).unwrap(); + assert_eq!(request["clientContent"]["turnComplete"], true); + socket + .send(Message::Text(r#"{"goAway":{"timeLeft":"9.5s"}}"#.into())) + .await + .unwrap(); + socket + .send(Message::Text( + r#"{"serverContent":{"modelTurn":{"parts":[{"text":"Hello"}]}}}"#.into(), + )) + .await + .unwrap(); + let mut generations = 1; + while let Some(Ok(message)) = socket.next().await { + if message.is_close() { + break; + } + if message.is_text() { + generations += 1; + } + } + generations + }); + let mut session = open_live_session_at(&url, &boot, None).await.unwrap(); + + // Tap the real reader, retaining its timestamps rather than inventing the + // observations the replay is supposed to preserve. + let (send, recv) = channel(GEMINI_EVENT_QUEUE); + let mut events = std::mem::replace(&mut session.events, recv); + let observed = Arc::new(Mutex::new(Vec::new())); + let captured = observed.clone(); + let forwarding = tokio::spawn(async move { + while let Some(event) = events.recv().await { + captured.lock().unwrap().push(event.clone()); + if send.send(event).await.is_err() { + break; + } + } + }); + session.send_context("opening", true).await.unwrap(); + tokio::time::timeout( + Duration::from_secs(5), + session.wait_for_model_output(&mut tokio::time::interval(KEEPALIVE_INTERVAL)), + ) + .await + .unwrap() + .unwrap(); + let expected = observed.lock().unwrap().clone(); + assert_eq!(expected.len(), 2); + assert!(matches!(&expected[0].1, GeminiEvent::GoAway { time_left } if time_left == "9.5s")); + assert_eq!(expected[1].1, GeminiEvent::Text("Hello".into())); + // Replay later, so stamping at consumption cannot accidentally pass. + tokio::time::sleep(Duration::from_millis(20)).await; + for (received_at, event) in expected { + assert_eq!(session.next_event().await, Some(event)); + assert_eq!(session.event_received_at(), Some(received_at)); + } + session.shutdown().await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), forwarding) + .await + .unwrap() + .unwrap(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(), + 1 + ); +} + +#[tokio::test] +async fn startup_output_requires_content_and_keeps_a_bounded_replay() { + let config = live_config(&[]); + let boot = bootstrap(&config, "interview-fixed", Some("two-sum"), 45); + let (address, _frames) = stalled_live_server(false).await; + let mut session = open_live_session_at(&format!("ws://{address}"), &boot, None) + .await + .unwrap(); + let empty = [ + GeminiEvent::Audio { + bytes: vec![], + mime_type: "audio/pcm".into(), + }, + GeminiEvent::Text(" ".into()), + GeminiEvent::OutputTranscript("\n".into()), + GeminiEvent::ToolCall(vec![]), + GeminiEvent::InputTranscript("candidate".into()), + GeminiEvent::CandidateVoiceActivity { + active: true, + audio_offset: None, + }, + GeminiEvent::CandidateVoiceActivity { + active: false, + audio_offset: None, + }, + GeminiEvent::UsageRecorded, + GeminiEvent::TurnComplete, + GeminiEvent::Interrupted, + ]; + let (send, recv) = channel(16); + session.events = recv; + for event in &empty { + send.send((std::time::Instant::now(), event.clone())) + .await + .unwrap(); + } + drop(send); + assert!( + session + .wait_for_model_output(&mut tokio::time::interval(KEEPALIVE_INTERVAL)) + .await + .is_err() + ); + for output in [ + GeminiEvent::Audio { + bytes: vec![1, 2], + mime_type: "audio/pcm".into(), + }, + GeminiEvent::Text("answer".into()), + GeminiEvent::OutputTranscript("answer".into()), + GeminiEvent::ToolCall(vec![GeminiFunctionCall { + id: "1".into(), + name: "read_editor".into(), + args: json!({}), + }]), + ] { + session.pending_events.clear(); + let (send, recv) = channel(16); + session.events = recv; + let received_at = std::time::Instant::now() - Duration::from_secs(2); + for event in &empty { + send.send((received_at, event.clone())).await.unwrap(); + } + send.send((received_at + Duration::from_millis(10), output.clone())) + .await + .unwrap(); + drop(send); + session + .wait_for_model_output(&mut tokio::time::interval(KEEPALIVE_INTERVAL)) + .await + .unwrap(); + for expected in &empty { + assert_eq!(session.next_event().await.as_ref(), Some(expected)); + assert_eq!(session.event_received_at(), Some(received_at)); + } + assert_eq!(session.next_event().await, Some(output)); + assert_eq!( + session.event_received_at(), + Some(received_at + Duration::from_millis(10)) + ); + // The live queue after replay must retain its own receive time too. + let (send, recv) = channel(1); + session.events = recv; + let next_received_at = received_at + Duration::from_millis(20); + send.send((next_received_at, GeminiEvent::TurnComplete)) + .await + .unwrap(); + assert_eq!(session.next_event().await, Some(GeminiEvent::TurnComplete)); + assert_eq!(session.event_received_at(), Some(next_received_at)); + } + let (send, recv) = channel(GEMINI_EVENT_QUEUE + 1); + session.events = recv; + for _ in 0..=GEMINI_EVENT_QUEUE { + send.send((std::time::Instant::now(), GeminiEvent::TurnComplete)) + .await + .unwrap(); + } + drop(send); + assert!( + session + .wait_for_model_output(&mut tokio::time::interval(KEEPALIVE_INTERVAL)) + .await + .unwrap_err() + .to_string() + .contains("event limit") + ); + assert_eq!(session.pending_events.len(), GEMINI_EVENT_QUEUE); + let reader = session.reader.abort_handle(); + drop(session); + tokio::task::yield_now().await; + assert!( + reader.is_finished(), + "startup cancellation must retire the reader" + ); +} + /// The clock the restart budget reads. It has to start at the socket, not /// at zero and not at the interview: a session that always reports no age /// makes every close look like an endpoint that will not stay up, and the @@ -3255,6 +3499,116 @@ async fn open_live_session_delivers_server_events_from_reader_task() { server.await.unwrap(); } +#[tokio::test] +async fn startup_output_wait_keeps_a_late_first_reply_alive() { + let config = live_config(&[]); + let boot = bootstrap(&config, "interview-fixed", Some("two-sum"), 45); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let (asked, request) = tokio::sync::oneshot::channel(); + let (release, mut output) = tokio::sync::oneshot::channel(); + let (acknowledged, mut acks) = tokio::sync::mpsc::unbounded_channel(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut socket = accept_async(socket).await.unwrap(); + socket.next().await.unwrap().unwrap(); + socket + .send(Message::Text(r#"{"setupComplete":{}}"#.into())) + .await + .unwrap(); + let frame = socket.next().await.unwrap().unwrap(); + let input: Value = serde_json::from_str(frame.to_text().unwrap()).unwrap(); + assert_eq!(input["realtimeInput"]["text"], "opening"); + asked.send(()).unwrap(); + let mut generations = 1; + let mut pings = 0; + let mut sent_output = false; + loop { + tokio::select! { + _ = &mut output, if !sent_output => { + socket.send(Message::Text(r#"{"serverContent":{"modelTurn":{"parts":[{"text":"late answer"}]}}}"#.into())).await.unwrap(); + sent_output = true; + } + frame = socket.next() => match frame.unwrap().unwrap() { + Message::Ping(payload) => { + socket.send(Message::Pong(payload)).await.unwrap(); + + // The observation acknowledges receipt by the reader, + // not just completion of the server's loopback write. + socket.send(Message::Text(r#"{"usageMetadata":{"promptTokenCount":0}}"#.into())).await.unwrap(); + pings += 1; + acknowledged.send(pings).unwrap(); + } + Message::Text(_) => generations += 1, + Message::Close(_) => break, + _ => {}, + } + } + } + (generations, pings) + }); + let mut session = open_live_session_at(&format!("ws://{address}"), &boot, None) + .await + .unwrap(); + session.send_text("opening").await.unwrap(); + request.await.unwrap(); + let observed = Arc::clone(&session.usage); + let waiting = tokio::spawn(async move { + let result = tokio::time::timeout( + Duration::from_secs(90), + session.wait_for_model_output(&mut tokio::time::interval(KEEPALIVE_INTERVAL)), + ) + .await; + (session, result) + }); + tokio::task::yield_now().await; + let began = tokio::time::Instant::now(); + for count in 1..=3 { + tokio::time::pause(); + tokio::time::advance(KEEPALIVE_INTERVAL).await; + tokio::time::resume(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), acks.recv()) + .await + .unwrap(), + Some(count) + ); + tokio::time::timeout(Duration::from_secs(5), async { + while observed.lock().unwrap().len() < count { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(!waiting.is_finished(), "no model output has been sent yet"); + } + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(10)).await; + tokio::time::resume(); + assert!(began.elapsed() > READ_IDLE_LIMIT); + release.send(()).unwrap(); + let (mut session, result) = tokio::time::timeout(Duration::from_secs(5), waiting) + .await + .unwrap() + .unwrap(); + result.unwrap().unwrap(); + for _ in 0..3 { + assert_eq!(session.next_event().await, Some(GeminiEvent::UsageRecorded)); + } + assert_eq!( + session.next_event().await, + Some(GeminiEvent::Text("late answer".into())) + ); + session.shutdown().await.unwrap(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(), + (1, 3) + ); +} + /// The whole point of reading the update: the handle has to still be /// there once the socket is gone, because that is when the caller asks. #[tokio::test] diff --git a/tests/unit/livekit.rs b/tests/unit/livekit.rs index d95abfcb..3176e044 100644 --- a/tests/unit/livekit.rs +++ b/tests/unit/livekit.rs @@ -16,6 +16,39 @@ use tokio::sync::mpsc; use crate::config::load_from_pairs; +#[path = "livekit/startup.rs"] +mod startup; + +async fn probe_live_output( + gemini: &mut GeminiLiveSession, + state: &mut RuntimeState, + greeting: &str, + limit: Duration, +) -> Result> { + let (_tx, mut input) = mpsc::unbounded_channel(); + super::probe_live_output( + gemini, + state, + greeting, + limit, + &mut StartupContext { + media: &mut CandidateMedia::new(), + candidate_identity: "candidate", + candidate_video: false, + deferred_events: &mut Vec::new(), + room_closed: &mut false, + }, + &mut input, + ) + .await +} + +impl StartupInputs for mpsc::UnboundedReceiver { + async fn next(&mut self, _media: &mut CandidateMedia) -> StartupInput { + self.recv().await.unwrap() + } +} + // 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. @@ -277,32 +310,237 @@ fn only_a_replacement_publishes_its_reconnect_wait() { } } -/// The interview starts once, when the candidate joined. -/// -/// Stamping a second `Instant::now()` after the room and the Gemini session -/// came up put this process a second or two behind the tab for the rest of -/// the interview, and the browser announces the round boundary exactly once -/// off its own clock. The gap has to be absorbed by a tolerance nobody can -/// justify, or removed here. Read as source because the two stamps are only -/// distinguishable in a room that really connected. -#[test] -fn the_interview_clock_starts_when_the_candidate_joined() { - let source = include_str!("../../src/livekit.rs"); - let opener = source - .split("async fn open_session") - .nth(1) - .expect("open_session is still defined here"); - let bound = opener - .split_once("let started_at =") - .expect("open_session still stamps the interview clock") - .1 - .lines() - .next() - .unwrap_or_default(); - assert!( - bound.contains("setup_began"), - "the clock must start where the candidate's does, not after setup: {bound}" +#[test] +fn startup_results_use_the_existing_control_path() { + assert_eq!( + startup_state(None), + serde_json::json!({ + "type": "interviewer_state", "reconnecting": false, "startup": "ready" + }) + ); + let error = std::io::Error::other("1011 Internal error, 503, or silence"); + assert_eq!(startup_state(Some(&error))["reason"], "unavailable"); +} + +#[tokio::test] +async fn startup_probe_waits_for_output_and_preserves_the_opening_events() { + use futures_util::{SinkExt, StreamExt}; + use tokio_tungstenite::tungstenite::Message; + let config = load_from_pairs([ + ("LIVEKIT_URL", "wss://example.livekit.cloud"), + ("LIVEKIT_API_KEY", "devkey"), + ("LIVEKIT_API_SECRET", "devsecret"), + ("GOOGLE_API_KEY", "startup-success"), + ]) + .unwrap(); + let keys = GeminiKeys::from_config(&config); + let boot = crate::runtime::bootstrap(&config, "interview-fixed", Some("two-sum"), 45); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}/", listener.local_addr().unwrap()); + let (asked, request) = tokio::sync::oneshot::channel(); + let (release, output) = tokio::sync::oneshot::channel(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut socket = tokio_tungstenite::accept_async(socket).await.unwrap(); + socket.next().await.unwrap().unwrap(); + socket + .send(Message::Text(r#"{"setupComplete":{}}"#.into())) + .await + .unwrap(); + let clock = socket.next().await.unwrap().unwrap(); + let clock: serde_json::Value = serde_json::from_str(clock.to_text().unwrap()).unwrap(); + assert_eq!(clock["clientContent"]["turnComplete"], false); + assert!( + clock["clientContent"]["turns"][0]["parts"][0]["text"] + .as_str() + .unwrap() + .contains("TIMER:") + ); + let message = socket.next().await.unwrap().unwrap(); + let message: serde_json::Value = serde_json::from_str(message.to_text().unwrap()).unwrap(); + assert_eq!(message["clientContent"]["turnComplete"], true); + assert_eq!( + message["clientContent"]["turns"][0]["parts"][0]["text"], + "opening" + ); + asked.send(()).unwrap(); + output.await.unwrap(); + socket + .send(Message::Text( + r#"{"serverContent":{"inputTranscription":{"text":"Candidate"}}}"#.into(), + )) + .await + .unwrap(); + socket + .send(Message::Text( + r#"{"serverContent":{"modelTurn":{"parts":[{"text":"Hello"}]}}}"#.into(), + )) + .await + .unwrap(); + let mut generations = 1; + while let Some(Ok(message)) = socket.next().await { + if message.is_close() { + break; + } + if message.is_text() { + generations += 1; + } + } + generations + }); + let mut gemini = crate::gemini::live_session_with_keys_at(&url, &keys, &boot, None) + .await + .unwrap(); + let mut state = initial_runtime_state(&boot, Instant::now()); + let greeting = with_timer(&state, "opening".to_string()); + let before = Instant::now(); + let ready = { + let probe = probe_live_output(&mut gemini, &mut state, &greeting, Duration::from_secs(5)); + tokio::pin!(probe); + tokio::select! { + result = &mut probe => panic!("setup alone must not start the interview: {result:?}"), + _ = request => {} + } + tokio::select! { + result = &mut probe => panic!("no model output yet: {result:?}"), + _ = tokio::time::sleep(Duration::from_millis(20)) => {} + } + release.send(()).unwrap(); + probe.await.unwrap() + }; + assert!(ready.duration_since(before) >= Duration::from_millis(20)); + assert_eq!( + gemini.next_event().await, + Some(GeminiEvent::InputTranscript("Candidate".into())) + ); + assert_eq!( + gemini.next_event().await, + Some(GeminiEvent::Text("Hello".into())) + ); + gemini.shutdown().await.unwrap(); + assert_eq!(server.await.unwrap(), 1); +} + +#[test] +fn startup_activity_preserves_opening_latency_and_restarts_the_reply_watch() { + let config = load_from_pairs([ + ("LIVEKIT_URL", "wss://example.livekit.cloud"), + ("LIVEKIT_API_KEY", "devkey"), + ("LIVEKIT_API_SECRET", "devsecret"), + ("GOOGLE_API_KEY", "startup-latency"), + ("CODETRIAL_GEMINI_FIRST_OUTPUT_S", "5"), + ("CODETRIAL_GEMINI_REPLY_TIMEOUT_S", "90"), + ]) + .unwrap(); + let requested = Instant::now(); + let received = requested + Duration::from_secs(55); + let ready = requested + Duration::from_secs(60); + let mut activity = + ready_runtime_activity(&config, "startup-room", ready, requested, 7, "opening"); + assert!(!activity.reply_timed_out(ready, false)); + assert_eq!( + activity.expired_reply_wait(ready + Duration::from_secs(5), false), + Some(Duration::from_secs(5)), ); + assert_eq!(activity.reply_timeout, Duration::from_secs(90)); + let record = activity.latency.audio(received).unwrap(); + assert_eq!(record["room"], "startup-room"); + assert_eq!(record["session"], 7); + assert_eq!(record["origin"], "prompt"); + assert_eq!(record["elapsed_ms"], 55_000); +} + +#[tokio::test] +async fn startup_probe_bounds_silence_without_retrying_or_guessing_quota() { + use futures_util::{SinkExt, StreamExt}; + use tokio_tungstenite::tungstenite::Message; + for frame in [ + r#"{}"#, + r#"{"goAway":{"timeLeft":"9.5s"}}"#, + r#"{"serverContent":{"turnComplete":true}}"#, + r#"{"serverContent":{"modelTurn":{"parts":[{"text":" "}]}}}"#, + r#"{"error":{"code":402,"message":"billing is not enabled"}}"#, + r#"{"error":{"code":429,"message":"resource exhausted"}}"#, + r#"{"error":{"code":401,"message":"invalid API key"}}"#, + r#"{"error":{"code":403,"message":"permission denied"}}"#, + r#"{"error":{"code":503,"message":"internal error"}}"#, + ] { + let config = load_from_pairs([ + ("LIVEKIT_URL", "wss://example.livekit.cloud"), + ("LIVEKIT_API_KEY", "devkey"), + ("LIVEKIT_API_SECRET", "devsecret"), + ("GOOGLE_API_KEY", "startup-failure"), + ]) + .unwrap(); + let keys = GeminiKeys::from_config(&config); + let boot = crate::runtime::bootstrap(&config, "interview-fixed", Some("two-sum"), 45); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}/", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut socket = tokio_tungstenite::accept_async(socket).await.unwrap(); + socket.next().await.unwrap().unwrap(); + socket + .send(Message::Text(r#"{"setupComplete":{}}"#.into())) + .await + .unwrap(); + let clock = socket.next().await.unwrap().unwrap(); + let clock: serde_json::Value = serde_json::from_str(clock.to_text().unwrap()).unwrap(); + assert_eq!(clock["clientContent"]["turnComplete"], false); + let request = socket.next().await.unwrap().unwrap(); + let request: serde_json::Value = + serde_json::from_str(request.to_text().unwrap()).unwrap(); + assert_eq!(request["clientContent"]["turnComplete"], true); + socket.send(Message::Text(frame.into())).await.unwrap(); + let mut generations = 1; + while let Some(Ok(message)) = socket.next().await { + if message.is_close() { + break; + } + if message.is_text() { + generations += 1; + } + } + generations + }); + let mut gemini = crate::gemini::live_session_with_keys_at(&url, &keys, &boot, None) + .await + .unwrap(); + let mut state = initial_runtime_state(&boot, Instant::now()); + let greeting = with_timer(&state, "opening".to_string()); + let error = probe_live_output( + &mut gemini, + &mut state, + &greeting, + Duration::from_millis(100), + ) + .await + .unwrap_err(); + let reason = crate::gemini::startup_failure_reason(error.as_ref()); + assert_eq!( + reason, + if frame.contains("402") { + "billing" + } else if frame.contains("429") { + "quota" + } else if frame.contains("401") { + "credentials" + } else if frame.contains("403") { + "refused" + } else { + "unavailable" + } + ); + gemini.shutdown().await.unwrap(); + drop(gemini); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(), + 1 + ); + } } #[test] @@ -324,6 +562,381 @@ fn only_the_interview_candidate_can_drive_the_runtime() { ); } +#[tokio::test] +async fn startup_ready_publication_failure_preserves_observed_usage() { + use futures_util::{SinkExt, StreamExt}; + use tokio_tungstenite::tungstenite::Message; + for stalled in [false, true] { + let config = load_from_pairs([ + ("LIVEKIT_URL", "wss://example.livekit.cloud"), + ("LIVEKIT_API_KEY", "devkey"), + ("LIVEKIT_API_SECRET", "devsecret"), + ("GOOGLE_API_KEY", "startup-accounting"), + ]) + .unwrap(); + let keys = GeminiKeys::from_config(&config); + let boot = crate::runtime::bootstrap(&config, "interview-fixed", Some("two-sum"), 45); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}/", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut socket = tokio_tungstenite::accept_async(socket).await.unwrap(); + socket.next().await.unwrap().unwrap(); + socket + .send(Message::Text(r#"{"setupComplete":{}}"#.into())) + .await + .unwrap(); + socket.next().await.unwrap().unwrap(); + socket.next().await.unwrap().unwrap(); + socket.send(Message::Text(r#"{"usageMetadata":{"promptTokenCount":42,"responseTokenCount":7,"totalTokenCount":49}}"#.into())).await.unwrap(); + socket + .send(Message::Text( + r#"{"serverContent":{"modelTurn":{"parts":[{"text":"Hello"}]}}}"#.into(), + )) + .await + .unwrap(); + socket + .next() + .await + .is_some_and(|message| message.unwrap().is_close()) + }); + let mut gemini = crate::gemini::live_session_with_keys_at(&url, &keys, &boot, None) + .await + .unwrap(); + let mut state = initial_runtime_state(&boot, Instant::now()); + let greeting = with_timer(&state, "opening".to_string()); + let mut summaries = Vec::new(); + let error = account_startup( + &config, + "interview-fixed", + 7, + |line| summaries.push(line), + async |usage| { + probe_live_output(&mut gemini, &mut state, &greeting, Duration::from_secs(5)) + .await?; + publish_startup_readiness(&mut gemini, usage, async { + if stalled { + std::future::pending::<()>().await; + } + Err(std::io::Error::other("notice rejected").into()) + }) + .await + }, + ) + .await + .unwrap_err(); + assert!( + gemini.drain_usage().is_empty(), + "observations are accounted once" + ); + assert_eq!(error.to_string().contains("timed out"), stalled); + assert_eq!( + summaries.len(), + 1, + "the production error exit logs exactly once" + ); + let summary = &summaries[0]; + assert!( + summary.contains("phase=startup prompt_tokens=42 response_tokens=7"), + "{summary}" + ); + assert!( + summary.contains("usage_samples=1 total_tokens=49"), + "{summary}" + ); + assert!( + tokio::time::timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(), + "the failed startup closes its socket" + ); + } +} + +#[tokio::test] +async fn startup_room_join_failure_is_accounted_exactly_once() { + let mut config = load_from_pairs([ + ("LIVEKIT_URL", "wss://example.livekit.cloud"), + ("LIVEKIT_API_KEY", "devkey"), + ("LIVEKIT_API_SECRET", "devsecret"), + ("GOOGLE_API_KEY", "startup-join-accounting"), + ]) + .unwrap(); + // Fail in the SDK before room setup, without reaching any external host. + config.livekit_url = "not-a-livekit-url".to_string(); + let keys = Arc::new(GeminiKeys::from_config(&config)); + let mut summaries = Vec::new(); + let result = tokio::time::timeout( + Duration::from_secs(5), + open_session(&config, &keys, "interview-startup-failure", 1, 7, |line| { + summaries.push(line); + }), + ) + .await + .expect("an invalid URL must fail before any network wait"); + assert!( + matches!(result, Ok(OpenSessionOutcome::FailedBeforeCandidate(_))), + "a failed room join must preserve the pre-candidate release outcome" + ); + assert_eq!(summaries.len(), 1, "every startup failure is counted once"); + let summary = &summaries[0]; + assert!(summary.starts_with("codetrial live_usage room=interview-startup-failure session=7 ")); + assert!( + summary.contains("elapsed_s=0 outcome=error phase=startup"), + "{summary}" + ); + assert!(summary.contains("usage_samples=0"), "{summary}"); + assert!(summary.contains("prompt_tokens=0"), "{summary}"); +} + +#[tokio::test(start_paused = true)] +async fn startup_failure_waits_for_receipt_before_leaving() { + let (received, receipt) = tokio::sync::oneshot::channel(); + let published = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let count = published.clone(); + let delivery = tokio::spawn(async move { + publish_startup_failure( + async { + count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(()) + }, + async { + receipt.await.unwrap(); + }, + ) + .await; + }); + tokio::task::yield_now().await; + assert_eq!(published.load(std::sync::atomic::Ordering::SeqCst), 1); + assert!( + !delivery.is_finished(), + "SDK publication only queues the notice; leaving now loses a delayed classified packet's sender" + ); + received.send(()).unwrap(); + delivery.await.unwrap(); + assert_eq!(published.load(std::sync::atomic::Ordering::SeqCst), 1); +} + +#[tokio::test(start_paused = true)] +async fn startup_failure_missing_receipt_shares_the_publication_deadline() { + let began = tokio::time::Instant::now(); + let delivery = tokio::spawn(async { + publish_startup_failure( + async { + tokio::time::sleep(RECONNECT_NOTICE_LIMIT / 2).await; + Ok(()) + }, + std::future::pending(), + ) + .await; + }); + tokio::task::yield_now().await; + tokio::time::advance(RECONNECT_NOTICE_LIMIT / 2).await; + tokio::task::yield_now().await; + assert!( + !delivery.is_finished(), + "a delayed publish must still allow a receipt" + ); + tokio::time::advance(RECONNECT_NOTICE_LIMIT / 2 - Duration::from_millis(1)).await; + tokio::task::yield_now().await; + assert!(!delivery.is_finished()); + tokio::time::advance(Duration::from_millis(1)).await; + delivery.await.unwrap(); + assert_eq!(began.elapsed(), RECONNECT_NOTICE_LIMIT); +} + +#[test] +fn startup_receipt_matches_browser_bytes_and_only_the_candidate() { + let wire: serde_json::Value = + serde_json::from_str(include_str!("../fixtures/startup-receipt.json")).unwrap(); + assert_eq!(wire["cases"].as_array().unwrap().len(), 1); + let payload = serde_json::to_vec(&wire["cases"][0]["payload"]).unwrap(); + let packet = br#"{"type":"interviewer_state","reconnecting":false,"startup":"failed","reason":"billing"}"#; + let id = crate::sha256_hex(&[packet]); + let topic = wire["topic"].as_str(); + assert!(is_startup_receipt( + topic, + Some("candidate"), + "candidate", + &payload, + &id + )); + assert!(!is_startup_receipt( + topic, + Some("other"), + "candidate", + &payload, + &id + )); + assert!(!is_startup_receipt(topic, None, "candidate", &payload, &id)); + assert!(!is_startup_receipt( + None, + Some("candidate"), + "candidate", + &payload, + &id + )); + assert!(!is_startup_receipt( + topic, + Some("candidate"), + "candidate", + &payload, + "different-notice" + )); + assert!(!is_startup_receipt( + topic, + Some("candidate"), + "candidate", + b"invalid json", + &id + )); + let report = + serde_json::to_vec(&serde_json::json!({"type":"report_received", "deliveryId":id})) + .unwrap(); + assert!(!is_startup_receipt( + topic, + Some("candidate"), + "candidate", + &report, + &id + )); +} + +#[tokio::test(start_paused = true)] +async fn startup_failure_rejected_or_stalled_publication_does_not_wait_for_receipt() { + for stalled in [false, true] { + let began = tokio::time::Instant::now(); + publish_startup_failure( + async { + if stalled { + std::future::pending::<()>().await; + } + Err(std::io::Error::other("notice rejected").into()) + }, + async { + panic!("a rejected publication cannot wait for a receipt"); + }, + ) + .await; + assert_eq!( + began.elapsed(), + if stalled { + RECONNECT_NOTICE_LIMIT + } else { + Duration::ZERO + } + ); + } +} + +#[tokio::test] +async fn startup_open_failover_does_not_repeat_the_readiness_generation() { + use futures_util::{SinkExt, StreamExt}; + use tokio_tungstenite::tungstenite::Message; + for answers in [false, true] { + let primary = format!("startup-open-{answers}-first"); + let backup = format!("startup-open-{answers}-backup"); + let configured = format!("{primary},{backup}"); + let config = load_from_pairs([ + ("LIVEKIT_URL", "wss://example.livekit.cloud"), + ("LIVEKIT_API_KEY", "devkey"), + ("LIVEKIT_API_SECRET", "devsecret"), + ("GOOGLE_API_KEYS", configured.as_str()), + ]) + .unwrap(); + let keys = GeminiKeys::from_config(&config); + let boot = crate::runtime::bootstrap(&config, "interview-fixed", Some("two-sum"), 45); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}/", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut refused = tokio_tungstenite::accept_async(socket).await.unwrap(); + refused.next().await.unwrap().unwrap(); + refused + .send(Message::Text( + r#"{"error":{"code":401,"message":"invalid API key"}}"#.into(), + )) + .await + .unwrap(); + drop(refused); + let (socket, _) = listener.accept().await.unwrap(); + let mut socket = tokio_tungstenite::accept_async(socket).await.unwrap(); + socket.next().await.unwrap().unwrap(); + socket + .send(Message::Text(r#"{"setupComplete":{}}"#.into())) + .await + .unwrap(); + let clock: serde_json::Value = + serde_json::from_str(socket.next().await.unwrap().unwrap().to_text().unwrap()) + .unwrap(); + assert_eq!(clock["clientContent"]["turnComplete"], false); + let request: serde_json::Value = + serde_json::from_str(socket.next().await.unwrap().unwrap().to_text().unwrap()) + .unwrap(); + assert_eq!(request["clientContent"]["turnComplete"], true); + socket + .send(Message::Text( + if answers { + r#"{"serverContent":{"modelTurn":{"parts":[{"text":"Hello"}]}}}"# + } else { + r#"{"serverContent":{"turnComplete":true}}"# + } + .into(), + )) + .await + .unwrap(); + let mut generations = 1; + while let Some(Ok(frame)) = socket.next().await { + if frame.is_close() { + break; + } + if frame.is_text() { + generations += 1; + } + } + generations + }); + let mut restarts = 0; + let mut gemini = crate::gemini::first_open_within( + crate::gemini::FIRST_OPEN_LIMIT, + retry_cold_open( + &keys, + &mut restarts, + Some(tokio::time::Instant::now() + crate::gemini::FIRST_OPEN_LIMIT), + || crate::gemini::live_session_with_keys_at(&url, &keys, &boot, None), + |_| async {}, + ), + ) + .await + .unwrap(); + assert_eq!(restarts, 1); + assert_eq!(keys.select().unwrap(), backup); + let mut state = initial_runtime_state(&boot, Instant::now()); + let greeting = with_timer(&state, "opening".to_string()); + assert_eq!( + probe_live_output( + &mut gemini, + &mut state, + &greeting, + Duration::from_millis(100) + ) + .await + .is_ok(), + answers + ); + let _ = gemini.shutdown().await; + drop(gemini); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(), + 1 + ); + } +} + /// The three guards a data packet passes before it can drive anything, in /// one place because they used to be three `continue`s that looked alike. #[test] @@ -1728,6 +2341,7 @@ async fn first_cold_open_retries_a_503_on_the_selected_key() { let session = retry_cold_open( &keys, &mut restarts, + None, || crate::gemini::live_session_with_keys_at(&url, &keys, &boot, None), |notice| { // A room slow to take the estimate must not stretch the wait it @@ -1813,6 +2427,7 @@ async fn a_sole_key_rate_limit_is_announced_as_quota() { retry_cold_open( &keys, &mut 0, + None, || crate::gemini::live_session_with_keys_at(&url, &keys, &boot, None), |notice| { notices.push(notice); @@ -1885,6 +2500,7 @@ async fn a_first_open_gives_up_at_its_time_limit() { retry_cold_open( &keys, &mut 0, + None, || crate::gemini::live_session_with_keys_at(&url, &keys, &boot, None), |_| async {}, ), @@ -1898,6 +2514,79 @@ async fn a_first_open_gives_up_at_its_time_limit() { assert!(started.elapsed() < COLD_OPEN_BACKOFF); } +#[tokio::test] +#[allow(clippy::result_large_err)] +async fn startup_open_preserves_quota_when_cooldown_outlasts_its_budget() { + use tokio_tungstenite::tungstenite::handshake::server; + + let config = load_from_pairs([ + ("LIVEKIT_URL", "wss://example.livekit.cloud"), + ("LIVEKIT_API_KEY", "devkey"), + ("LIVEKIT_API_SECRET", "devsecret"), + ("GOOGLE_API_KEYS", "startup-cooled-a,startup-cooled-b"), + ]) + .unwrap(); + let keys = GeminiKeys::from_config(&config); + let boot = crate::runtime::bootstrap(&config, "interview-fixed", Some("two-sum"), 45); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}/", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + for _ in 0..2 { + let (socket, _) = listener.accept().await.unwrap(); + let _ = tokio_tungstenite::accept_hdr_async( + socket, + |_: &server::Request, _: server::Response| { + let mut refusal = server::ErrorResponse::new(None); + *refusal.status_mut() = axum::http::StatusCode::TOO_MANY_REQUESTS; + Err(refusal) + }, + ) + .await; + } + }); + for _ in 0..2 { + assert!( + crate::gemini::live_session_with_keys_at(&url, &keys, &boot, None) + .await + .is_err() + ); + } + server.await.unwrap(); + + let mut restarts = 0; + let mut attempts = 0; + let mut notices = Vec::new(); + let limit = COLD_OPEN_BACKOFF / 4; + let error = crate::gemini::first_open_within( + limit, + retry_cold_open( + &keys, + &mut restarts, + Some(tokio::time::Instant::now() + limit), + || { + attempts += 1; + let error = keys.select().unwrap_err(); + async move { Err(error.into()) } + }, + |notice| { + notices.push(notice); + async {} + }, + ), + ) + .await + .err() + .unwrap(); + assert_eq!( + crate::gemini::startup_failure_reason(error.as_ref()), + "quota" + ); + assert!(crate::gemini::exhausted_until(error.as_ref()).is_some()); + assert_eq!(attempts, 1); + assert_eq!(restarts, 0); + assert!(notices.is_empty()); +} + /// Every key's project out of credit: the rotation tries each once and then /// runs dry, and the error it hands back is still the billing failure, which /// is what the room's summary names, not the empty rotation it left behind. @@ -1937,6 +2626,7 @@ async fn a_rotation_emptied_by_billing_failures_reports_billing() { let error = retry_cold_open( &keys, &mut 0, + None, || { attempts += 1; crate::gemini::live_session_with_keys_at(&url, &keys, &boot, None) @@ -2030,6 +2720,7 @@ async fn an_exhausted_rotation_waits_only_for_a_key_out_on_quota() { retry_cold_open( &keys, &mut restarts, + None, || { attempts += 1; let error = keys.select().unwrap_err(); diff --git a/tests/unit/livekit/media.rs b/tests/unit/livekit/media.rs index ed104751..2a9c8f03 100644 --- a/tests/unit/livekit/media.rs +++ b/tests/unit/livekit/media.rs @@ -204,6 +204,9 @@ async fn a_failed_flush_is_an_error_the_caller_can_see() { fn an_ended_audio_track_is_released_even_when_the_flush_fails() { let loop_source = include_str!("../../../src/livekit.rs"); let arm = loop_source + .split("pub(crate) async fn run_room_with_slot") + .nth(1) + .expect("the room loop entry point still exists") .split("frame = next_audio_frame") .nth(1) .expect("the room loop still has an audio arm"); diff --git a/tests/unit/livekit/startup.rs b/tests/unit/livekit/startup.rs new file mode 100644 index 00000000..b779871d --- /dev/null +++ b/tests/unit/livekit/startup.rs @@ -0,0 +1,401 @@ +use super::*; +use futures_util::{SinkExt, StreamExt}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use tokio_tungstenite::tungstenite::Message; + +struct StartupWire { + audio: Vec, + generations: usize, + pings: usize, +} + +async fn startup_socket( + answer_after_bytes: Option, +) -> ( + GeminiLiveSession, + tokio::task::JoinHandle, + Arc, +) { + let config = startup_config("60"); + let keys = GeminiKeys::from_config(&config); + let boot = crate::runtime::bootstrap(&config, "interview-fixed", Some("two-sum"), 45); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}/", listener.local_addr().unwrap()); + let received = Arc::new(AtomicUsize::new(0)); + let observed = received.clone(); + let server = tokio::spawn(async move { + use base64::Engine; + let (socket, _) = listener.accept().await.unwrap(); + let mut socket = tokio_tungstenite::accept_async(socket).await.unwrap(); + socket.next().await.unwrap().unwrap(); + socket + .send(Message::Text(r#"{"setupComplete":{}}"#.into())) + .await + .unwrap(); + let mut wire = StartupWire { + audio: Vec::new(), + generations: 0, + pings: 0, + }; + while let Some(Ok(message)) = socket.next().await { + match message { + Message::Close(_) => break, + Message::Ping(bytes) => { + wire.pings += 1; + socket.send(Message::Pong(bytes)).await.unwrap(); + } + Message::Text(text) => { + let value: serde_json::Value = serde_json::from_str(&text).unwrap(); + if value["clientContent"]["turnComplete"] == true { + wire.generations += 1; + if answer_after_bytes.is_none() { + observed.store(1, Ordering::SeqCst); + } + } + if let Some(data) = value["realtimeInput"]["audio"]["data"].as_str() { + wire.audio.extend( + base64::engine::general_purpose::STANDARD + .decode(data) + .unwrap(), + ); + observed.store(wire.audio.len(), Ordering::SeqCst); + if Some(wire.audio.len()) == answer_after_bytes { + socket.send(Message::Text(r#"{"serverContent":{"modelTurn":{"parts":[{"text":"Hello"}]}}}"#.into())).await.unwrap(); + } + } + } + _ => {} + } + } + wire + }); + let gemini = crate::gemini::live_session_with_keys_at(&url, &keys, &boot, None) + .await + .unwrap(); + (gemini, server, received) +} + +fn startup_config(first_output: &str) -> AgentConfig { + load_from_pairs([ + ("LIVEKIT_URL", "wss://example.livekit.cloud"), + ("LIVEKIT_API_KEY", "devkey"), + ("LIVEKIT_API_SECRET", "devsecret"), + ("GOOGLE_API_KEY", "startup-media"), + ("CODETRIAL_GEMINI_FIRST_OUTPUT_S", first_output), + ("CODETRIAL_GEMINI_REPLY_TIMEOUT_S", "90"), + ]) + .unwrap() +} + +fn startup_state_for(config: &AgentConfig) -> RuntimeState { + let boot = crate::runtime::bootstrap(config, "interview-fixed", Some("two-sum"), 45); + initial_runtime_state(&boot, Instant::now()) +} + +#[tokio::test] +async fn startup_pumps_more_than_the_audio_queue_before_a_late_first_output() { + let wire = startup_with_audio_until_output("60", 5_500).await; + assert!((3..=5).contains(&wire.pings), "pings={}", wire.pings); +} + +#[tokio::test] +async fn startup_accepts_output_after_a_short_recovery_deadline() { + let wire = startup_with_audio_until_output("5", 600).await; + assert!(wire.pings <= 1, "pings={}", wire.pings); +} + +#[tokio::test] +async fn startup_preserves_participant_departure_and_return_until_ready() { + startup_with_audio_until_output("5", 2).await; +} + +async fn startup_with_audio_until_output(first_output: &str, frames: usize) -> StartupWire { + use ::livekit::webrtc::audio_frame::AudioFrame; + let expected: Vec = (0..frames) + .flat_map(|i| (0..160).flat_map(move |_| (i as i16).to_le_bytes())) + .collect(); + let (mut gemini, server, received) = startup_socket(Some(expected.len())).await; + let config = startup_config(first_output); + assert_eq!( + config.gemini_first_output_s, + first_output.parse::().unwrap() + ); + let mut state = startup_state_for(&config); + let greeting = with_timer(&state, "opening".into()); + let mut media = CandidateMedia::new(); + let mut deferred = Vec::new(); + let mut left = false; + let (tx, mut rx) = mpsc::unbounded_channel(); + + // LiveKit owns RemoteParticipant construction. Distinct opaque events prove + // that classified departures and later room events retain their order. + tx.send(StartupInput::ParticipantLeft(RoomEvent::Reconnecting)) + .unwrap(); + tx.send(StartupInput::Room(Some(RoomEvent::Reconnected))) + .unwrap(); + let mut context = StartupContext { + media: &mut media, + candidate_identity: "candidate", + candidate_video: false, + deferred_events: &mut deferred, + room_closed: &mut left, + }; + tokio::time::pause(); + let began = tokio::time::Instant::now(); + let probe = crate::livekit::probe_live_output( + &mut gemini, + &mut state, + &greeting, + startup_output_limit(&config), + &mut context, + &mut rx, + ); + let feed = async { + let real_deadline = Instant::now() + Duration::from_secs(10); + for batch in 0..frames / 2 { + for offset in 0..2 { + let value = (batch * 2 + offset) as i16; + tx.send(StartupInput::Audio(Some(AudioFrame { + data: vec![value; 160].into(), + sample_rate: 16_000, + num_channels: 1, + samples_per_channel: 160, + }))) + .unwrap(); + } + + // Keep the scheduler awake while the real loopback socket drains; + // paused Tokio time must not jump to the deadline during kernel IO. + while received.load(Ordering::SeqCst) < (batch + 1) * 640 { + assert!( + Instant::now() < real_deadline, + "startup stopped pumping audio" + ); + tokio::task::yield_now().await; + } + tokio::time::advance(Duration::from_millis(20)).await; + } + for _ in 0..1_000 { + tokio::task::yield_now().await; + } + }; + let (ready, ()) = tokio::join!(probe, feed); + ready.unwrap(); + assert!(began.elapsed() >= Duration::from_millis(frames as u64 * 10)); + assert!(!left); + assert!(matches!( + deferred.as_slice(), + [RoomEvent::Reconnecting, RoomEvent::Reconnected] + )); + assert_eq!( + gemini.next_event().await, + Some(GeminiEvent::Text("Hello".into())) + ); + gemini.shutdown().await.unwrap(); + tokio::time::resume(); + let wire = server.await.unwrap(); + assert_eq!(wire.audio, expected); + assert_eq!(wire.generations, 1); + wire +} + +#[tokio::test] +async fn startup_room_loss_still_cancels_the_wait() { + for event in [ + StartupInput::Room(None), + StartupInput::Room(Some(RoomEvent::Disconnected { + reason: DisconnectReason::ClientInitiated, + })), + ] { + let (mut gemini, server, _) = startup_socket(None).await; + let config = startup_config("60"); + let mut state = startup_state_for(&config); + let greeting = with_timer(&state, "opening".into()); + let mut media = CandidateMedia::new(); + let mut deferred = Vec::new(); + let mut left = false; + let (_tx, mut input) = mpsc::unbounded_channel(); + _tx.send(StartupInput::Room(Some(RoomEvent::Reconnecting))) + .unwrap(); + _tx.send(event).unwrap(); + let before = tokio::time::Instant::now(); + let error = crate::livekit::probe_live_output( + &mut gemini, + &mut state, + &greeting, + startup_output_limit(&config), + &mut StartupContext { + media: &mut media, + candidate_identity: "candidate", + candidate_video: false, + deferred_events: &mut deferred, + room_closed: &mut left, + }, + &mut input, + ) + .await + .unwrap_err(); + assert!(left); + assert!(before.elapsed() < Duration::from_secs(1)); + assert_eq!( + error.downcast_ref::().unwrap().kind(), + std::io::ErrorKind::Interrupted + ); + assert!(matches!( + input.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + )); + gemini.shutdown().await.unwrap(); + assert_eq!(server.await.unwrap().generations, 1); + } +} + +#[tokio::test] +async fn startup_uses_the_first_output_limit_despite_repeated_media_wakeups() { + for (setting, seconds) in [("", 10), ("5", 10), ("10", 10), ("23", 23), ("120", 90)] { + let (mut gemini, server, requested) = startup_socket(None).await; + let config = startup_config(setting); + assert_eq!(startup_output_limit(&config), Duration::from_secs(seconds)); + let mut state = startup_state_for(&config); + let greeting = with_timer(&state, "opening".into()); + let mut media = CandidateMedia::new(); + let mut deferred = Vec::new(); + let mut left = false; + let (tx, mut input) = mpsc::unbounded_channel(); + tx.send(StartupInput::ParticipantLeft(RoomEvent::Reconnecting)) + .unwrap(); + tx.send(StartupInput::Room(Some(RoomEvent::Reconnected))) + .unwrap(); + let mut context = StartupContext { + media: &mut media, + candidate_identity: "candidate", + candidate_video: false, + deferred_events: &mut deferred, + room_closed: &mut left, + }; + let done = std::sync::atomic::AtomicBool::new(false); + tokio::time::pause(); + let run = async { + let result = crate::livekit::probe_live_output( + &mut gemini, + &mut state, + &greeting, + startup_output_limit(&config), + &mut context, + &mut input, + ) + .await; + done.store(true, Ordering::SeqCst); + result + }; + let drive = async { + let real_deadline = Instant::now() + Duration::from_secs(5); + while requested.load(Ordering::SeqCst) == 0 { + assert!(Instant::now() < real_deadline, "opening was never sent"); + tokio::task::yield_now().await; + } + for _ in 0..seconds { + assert!( + !done.load(Ordering::SeqCst), + "deadline ended early: {setting}" + ); + tx.send(StartupInput::Audio(None)).unwrap(); + for _ in 0..100 { + tokio::task::yield_now().await; + } + tokio::time::advance(Duration::from_secs(1)).await; + } + // Tokio's timer wheel rounds a non-aligned deadline up to a ms. + tokio::time::advance(Duration::from_millis(1)).await; + for _ in 0..100 { + tokio::task::yield_now().await; + } + assert!( + done.load(Ordering::SeqCst), + "media restarted the deadline: {setting}" + ); + }; + let (result, ()) = tokio::join!(run, drive); + let error = result.unwrap_err(); + assert_eq!( + error.downcast_ref::().unwrap().kind(), + std::io::ErrorKind::TimedOut + ); + assert_eq!( + crate::gemini::startup_failure_reason(error.as_ref()), + "unavailable" + ); + assert!(!left); + assert!(matches!( + deferred.as_slice(), + [RoomEvent::Reconnecting, RoomEvent::Reconnected] + )); + gemini.shutdown().await.unwrap(); + tokio::time::resume(); + assert_eq!(server.await.unwrap().generations, 1); + } +} + +#[tokio::test] +async fn startup_reads_native_microphone_frames_and_preserves_other_room_events() { + use ::livekit::prelude::{LocalAudioTrack, RtcAudioSource}; + use ::livekit::webrtc::audio_frame::AudioFrame; + use ::livekit::webrtc::audio_stream::native::{NativeAudioStream, NativeAudioStreamOptions}; + let expected = vec![0u8; 80 * 320]; + let (mut gemini, server, _) = startup_socket(Some(expected.len())).await; + let config = startup_config("5"); + let mut state = startup_state_for(&config); + let greeting = with_timer(&state, "opening".into()); + let source = NativeAudioSource::new(AudioSourceOptions::default(), 16_000, 1, 0); + let track = LocalAudioTrack::create_audio_track( + "startup-microphone", + RtcAudioSource::Native(source.clone()), + ); + let mut media = CandidateMedia::new(); + media.audio = Some(NativeAudioStream::with_options( + track.rtc_track(), + 16_000, + 1, + NativeAudioStreamOptions { + queue_size_frames: Some(4000), + }, + )); + for _ in 0..80 { + source + .capture_frame(&AudioFrame { + data: vec![0; 160].into(), + sample_rate: 16_000, + num_channels: 1, + samples_per_channel: 160, + }) + .await + .unwrap(); + } + let (tx, events) = mpsc::unbounded_channel(); + tx.send(RoomEvent::Reconnecting).unwrap(); + let events = tokio::sync::Mutex::new(events); + let mut deferred = Vec::new(); + let mut left = false; + crate::livekit::probe_live_output( + &mut gemini, + &mut state, + &greeting, + startup_output_limit(&config), + &mut StartupContext { + media: &mut media, + candidate_identity: "candidate", + candidate_video: false, + deferred_events: &mut deferred, + room_closed: &mut left, + }, + &mut &events, + ) + .await + .unwrap(); + assert!(!left); + assert!(matches!(deferred.as_slice(), [RoomEvent::Reconnecting])); + gemini.shutdown().await.unwrap(); + let wire = server.await.unwrap(); + assert_eq!(wire.audio, expected); + assert_eq!(wire.generations, 1); +} diff --git a/web/before-unload.js b/web/before-unload.js index 3b5077b4..40e92acc 100644 --- a/web/before-unload.js +++ b/web/before-unload.js @@ -1,7 +1,13 @@ -/// A positive deadline means the timed interview has started; keep warning -/// until a report copy is available from Past attempts. -export function shouldWarnBeforeUnload({ endsAt, reportPersisted }) { - return Number.isFinite(endsAt) && endsAt > 0 && !reportPersisted; +/// Readiness can leave editable work without a deadline. Neither that work nor +/// a timed attempt is safe to leave until a report copy is in Past attempts. +export function shouldWarnBeforeUnload({ + endsAt, + workStarted, + reportPersisted, +}) { + return ( + (workStarted || (Number.isFinite(endsAt) && endsAt > 0)) && !reportPersisted + ); } /// The browser owns the confirmation text; the page can only request it. diff --git a/web/interview.js b/web/interview.js index 6b6bc289..5fcb9fd9 100644 --- a/web/interview.js +++ b/web/interview.js @@ -72,6 +72,7 @@ import { boardStreamOptions, clamp, codeUpdatePayload, + createInterviewerReadiness, codingLoop, countdown, FRAMEWORKS, @@ -87,6 +88,7 @@ import { modeIsWhiteboard, providerUiState, receiveReportDelivery, + sendStartupReceipt, roomInterviewer, sanitizeReport, sessionReport, @@ -459,6 +461,7 @@ const state = { // derived from it rather than accumulated, so throttling and suspend cannot // bend it. endsAt: 0, + workStarted: false, // True only after the latest report save confirms a local or account copy // the Past attempts flow can reopen. reportPersisted: false, @@ -482,7 +485,9 @@ const state = { /// interviewer's own evaluation is missing from it. reportUnreadable: false, room: null, + startupCleanupRoom: null, connected: false, + interviewerReadiness: null, /// The consent this interview is running under, `null` where the server /// records nothing. The withdrawal button needs it and has nowhere else to /// read it from. @@ -518,6 +523,7 @@ window.addEventListener("pageshow", (event) => { !event.persisted || shouldWarnBeforeUnload({ endsAt: state.endsAt, + workStarted: state.workStarted, reportPersisted: state.reportPersisted, }) ) @@ -530,6 +536,7 @@ function updateBeforeUnloadGuard() { beforeUnloadGuard.setEnabled( shouldWarnBeforeUnload({ endsAt: state.endsAt, + workStarted: state.workStarted, reportPersisted: state.reportPersisted, }), ); @@ -905,14 +912,19 @@ async function init() { // is clear by the time the candidate is actually working rather than holding // a caption nobody said. showCaptions(); - const started = await connect(preflight, presenting); - if (started === false) return; - if (strict) { - nodes.end.disabled = false; - nodes.pause.disabled = false; + if (strict) nodes.end.disabled = true; + nodes.pause.disabled = true; + if (!strict) { + state.workStarted = true; + updateBeforeUnloadGuard(); } + const startup = await connect(preflight, presenting); + if (state.phase !== "live") return; + nodes.end.disabled = strict && !startup; + nodes.pause.disabled = strict && !startup; + if (!startup) return; publishCode(); - state.endsAt = Date.now() + durationMin * 60 * 1000; + state.endsAt = startup.startedAt + durationMin * 60 * 1000; updateBeforeUnloadGuard(); tickTimer(); setInterval(tickTimer, 1000); @@ -1928,6 +1940,8 @@ function runAudioCheck() { async function connect(preflight, presenting = false) { setAgentStateLabel(providerUiState("connecting").label); + state.interviewerReadiness = createInterviewerReadiness(); + let offlineStartup = null; try { // Before the token, because the token is what precedes an Egress call. The // server refuses a recorded room without this row, so the ordering is @@ -2012,6 +2026,8 @@ async function connect(preflight, presenting = false) { state.connected = true; if (strict) updateRunAvailability(); if (strict && !whiteboard) nodes.editor.disabled = false; + state.workStarted = true; + updateBeforeUnloadGuard(); // After the room, not before it. The server refuses to record an interview // that has not claimed a room, so starting here rather than alongside the // consent request is the difference between a recording and a failure @@ -2022,7 +2038,9 @@ async function connect(preflight, presenting = false) { if (recordingEnabled && state.interviewId) { nodes.withdrawConsent.hidden = false; nodes.endingWithdrawConsent.hidden = false; - await startRecording(); + // Recording status has its own recovery. It must not hold readiness or + // keep local devices running after a startup failure. + void startRecording(); // First event of the replay, so a template that joins late knows the // interview was already running rather than inferring it from silence. recordReplay("lifecycle", { @@ -2047,6 +2065,18 @@ async function connect(preflight, presenting = false) { language: state.language, }); } + const startup = await state.interviewerReadiness.wait; + if (startup.startup !== "ready" || state.phase !== "live") { + const error = new Error("Gemini startup check failed"); + error.startup = startup; + throw error; + } + board.ready = true; + exampleBoard?.refresh(); + if (whiteboard) { + paintBoard(); + } + return startup; } catch (error) { const failure = !strict && error?.name === "TimeoutError" @@ -2067,11 +2097,19 @@ async function connect(preflight, presenting = false) { const joined = state.room; state.connected = false; state.room = null; + if (joined && error?.startup) { + state.startupCleanupRoom = joined; + await error.startup.flushed; + } if (joined) void joined.disconnect?.()?.catch?.(() => {}); stopPreflight(preflight); - const reason = - strict && - (error?.name === "TimeoutError" || [408, 504].includes(error?.status)) + stopLocalMedia(); + state.interviewerReadiness.fail(); + if (state.phase !== "live") return false; + const reason = error?.startup + ? interviewerReconnectMessage(error.startup, { strict }) + : strict && + (error?.name === "TimeoutError" || [408, 504].includes(error?.status)) ? "Strict Interview connection timed out. Return to the lobby or retry media preflight." : failure?.message; if (strict) { @@ -2118,6 +2156,16 @@ async function connect(preflight, presenting = false) { return false; } const degraded = providerUiState("degraded", reason); + if (error?.startup) { + degraded.message = interviewerReconnectMessage(error.startup); + } + // A deployment with no live session still offers the old timed practice. + // No failure after joining grants a replacement local clock. + if (!error?.startup && !state.joinedRoom) { + offlineStartup = { startedAt: Date.now() }; + } + state.workStarted = true; + updateBeforeUnloadGuard(); setAgentStateLabel(degraded.label); // A degraded start without a reason reads as the product working. The server // says why it refused, in words written for a candidate, and a busy server @@ -2130,13 +2178,13 @@ async function connect(preflight, presenting = false) { true, ); } - // Joined and recording, or offline with nothing to send: either way the - // board's first stroke now reaches everything that should see it. + // Offline practice keeps its board even after a failed readiness check. board.ready = true; exampleBoard?.refresh(); if (whiteboard) { paintBoard(); } + return offlineStartup; } // The replay: what the candidate was looking at, sent to the server so the @@ -2170,7 +2218,7 @@ async function connectLiveKit(connection, preflight, presenting = false) { livekit.RoomEvent.DataReceived, async (payload, participant, kind, topic) => { if (topic === topics.control && isAgent(participant)) { - receiveControl(payload); + receiveControl(payload, participant, room); return; } if (!acceptsReport(topic, participant)) return; @@ -2206,7 +2254,14 @@ async function connectLiveKit(connection, preflight, presenting = false) { // the agent's participant record existed, and nothing ever rewrote it. An // agent that leaves now updates the label too, instead of leaving it claiming // an interviewer who is gone. - room.on(livekit.RoomEvent.ParticipantDisconnected, updateAgentState); + room.on(livekit.RoomEvent.ParticipantDisconnected, (participant) => { + if ( + isAgent(participant) && + !roomInterviewer(roomParticipants(), state.agentIdentity) + ) + state.interviewerReadiness?.fail(); + updateAgentState(); + }); room.on(livekit.RoomEvent.TrackSubscribed, updateAgentState); // The room's own connection had no handlers at all, so a mid-interview drop // left `state.connected` true, the UI silent, and every `publish` quietly @@ -2245,6 +2300,9 @@ async function connectLiveKit(connection, preflight, presenting = false) { }); room.on(livekit.RoomEvent.Disconnected, () => { audioPlayback.release(room); + const startupCleanup = state.startupCleanupRoom === room; + if (startupCleanup) state.startupCleanupRoom = null; + state.interviewerReadiness?.fail(); state.connected = false; if (strict) updateRunAvailability(); // Terminal, not transient: LiveKit fires Reconnecting for a recoverable @@ -2257,13 +2315,14 @@ async function connectLiveKit(connection, preflight, presenting = false) { if (state.phase === "report_recovery") reportRecovery.finish(); state.room = null; if (strict) setAgentStateLabel("Not connected"); - setBanner( - "connection", - strict - ? "The Strict Interview connection dropped." - : providerUiState("degraded", "The interview connection dropped.") - .message, - ); + if (!startupCleanup) + setBanner( + "connection", + strict + ? "The Strict Interview connection dropped." + : providerUiState("degraded", "The interview connection dropped.") + .message, + ); console.warn("codetrial room_disconnected"); }); // The participant is needed to tell Jim from any other remote audio, so both @@ -2974,7 +3033,7 @@ function updatePresenceBanner(eventType) { /// a lid close stopped it entirely; this file already reasons about exactly /// that hazard for the face sampler. function tickTimer() { - if (state.phase !== "live") return; + if (state.phase !== "live" || !state.endsAt) return; const tick = countdown(state.endsAt, Date.now()); nodes.timer.textContent = formatTime(tick.remaining); nodes.timer.classList.toggle("urgent", tick.urgent); @@ -3106,7 +3165,7 @@ function showFrameworkHint() { }, FRAMEWORK_HINT_MS); } -function receiveControl(bytes) { +function receiveControl(bytes, participant, room) { try { const message = JSON.parse(new TextDecoder().decode(bytes)); if (message.type === "report_retry") { @@ -3125,6 +3184,24 @@ function receiveControl(bytes) { message.type === "interviewer_state" && typeof message.reconnecting === "boolean" ) { + if (message.startup) { + if (!isCurrentAgent(participant)) return; + const flushed = + message.startup === "failed" + ? sendStartupReceipt(bytes, (receipt) => + room.localParticipant.publishData( + new TextEncoder().encode(JSON.stringify(receipt)), + { + reliable: true, + topic: topics.control, + destinationIdentities: [participant.identity], + }, + ), + ) + : undefined; + state.interviewerReadiness?.receive(message, flushed); + return; + } // Gemini caps how long one connection lasts, so this arrives several // times in a normal interview. It is the only notice the candidate gets // that the interviewer went deaf: the agent stays in the room across it, diff --git a/web/lib.js b/web/lib.js index cc26563e..6082a6c6 100644 --- a/web/lib.js +++ b/web/lib.js @@ -338,18 +338,18 @@ export async function receiveReportDelivery( publishReceipt, receive, ) { - const flushed = sendReportReceipt(payload, publishReceipt); + const flushed = sendReceipt(reportReceipt(payload), publishReceipt); if (first) await receive(payload, flushed); await flushed; } /// Hashing and publication share one bound, so neither can hold the page. -async function sendReportReceipt(payload, publishReceipt) { +async function sendReceipt(receipt, publishReceipt) { let timer; let queued; try { queued = await Promise.race([ - reportReceipt(payload) + receipt .then(async (receipt) => { if (!receipt) return false; await publishReceipt(receipt); @@ -371,6 +371,18 @@ export function reportReceiptPayload(deliveryId) { return { type: "report_received", deliveryId }; } +export async function startupReceipt(payload) { + try { + return { type: "startup_received", deliveryId: await sha256Hex(payload) }; + } catch { + return null; + } +} + +export function sendStartupReceipt(payload, publishReceipt) { + return sendReceipt(startupReceipt(payload), publishReceipt); +} + // Data-channel payload builders. The agent decodes these by key, so they are // a wire contract; web/tests/lib.test.js pins the exact shapes. @@ -1648,7 +1660,26 @@ const RECONNECT_CAUSES = { retrying: "could not be reached", }; -export function interviewerReconnectMessage(message) { +export function interviewerReconnectMessage(message, { strict = false } = {}) { + if (message.startup === "failed") { + const causes = { + quota: + "Gemini reported a quota or rate limit. Check the project's limits before starting again.", + billing: + "Gemini reported a billing or credit problem. Check the project's billing before starting again.", + credentials: + "Gemini rejected the API credentials. Check the configured key before starting again.", + refused: + "Gemini refused this Live request. Check the project's access before starting again.", + }; + const cause = Object.hasOwn(causes, message.reason) + ? causes[message.reason] + : "Gemini did not answer the startup check. Try starting again when the provider is available."; + const next = strict + ? "Return to the lobby or retry media preflight." + : "You can still work the problem offline."; + return `${cause} The interview timer has not started. ${next}`; + } if (!Object.hasOwn(RECONNECT_CAUSES, message.reason)) { return providerUiState("interviewer_reconnecting").message; } @@ -1661,6 +1692,45 @@ export function interviewerReconnectMessage(message) { return `The interviewer ${cause} and cannot hear you right now.${estimate} Keep working; nothing is lost.`; } +/// Installed before joining: a fast agent may publish readiness before the +/// caller finishes publishing media or starts recording. Replacement notices +/// never settle this one-shot gate, and a missing notice has a finite wait. +export function createInterviewerReadiness({ + timeoutMs = 180_000, + now = Date.now, + schedule = setTimeout, + cancel = clearTimeout, +} = {}) { + let finish; + let settled = false; + const wait = new Promise((resolve) => { + finish = resolve; + }); + let timer; + const receive = (message, flushed) => { + if ( + settled || + message.type !== "interviewer_state" || + !["ready", "failed"].includes(message.startup) + ) + return; + settled = true; + cancel(timer); + finish( + message.startup === "ready" + ? { startup: "ready", startedAt: now() } + : { + startup: "failed", + reason: message.reason, + ...(flushed ? { flushed } : {}), + }, + ); + }; + const fail = () => receive({ type: "interviewer_state", startup: "failed" }); + timer = schedule(fail, timeoutMs); + return { wait, receive, fail }; +} + export function providerUiState(kind, detail = "") { const detailText = String(detail).toLowerCase(); const reason = /429|rate limit|too many|busy|capacity/.test(detailText)