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)