Play an acknowledgement while a tool round runs

A tool round holds the reply a candidate is waiting for back by a whole
model round trip, during which they hear nothing. With
CODETRIAL_GEMINI_TOOL_ACK set, off by default, the agent plays a short
acknowledgement generated once per interview by Gemini's speech model in
the session's voice and accent. It plays at most once per thing the
candidate said, a wait kept across an interruption included, never while
they are speaking again, the floor is held or a goodbye is due, and only
into idle output, so it is never heard over audio. The clip is trimmed
of its padding to whole output frames, and the reply's first audio cuts
whatever of it is still queued or playing, so the reply never waits for
the filler. The output worker now checks for cancellation before
capturing a frame, so a frame cut by a barge-in is not published either.

The reply's own latency is unchanged in meaning: the trace stamps the
acknowledgement as tool_ack_ms, and the analyzer's tool_ack_lead is how
much sooner than the reply's captured audio the candidate heard
something.

Author: jserv

.cargo/mutants.toml

@@ -1,7 +1,7 @@
 # Functions the mutation gate cannot judge, because `cargo test` cannot reach
 # them. Matched against the mutant names that `cargo mutants --list` prints.
 #
-# EXCLUSIONS: 71
+# EXCLUSIONS: 74
 #
 # That number is checked by `scripts/test.sh`, so adding an entry means editing
 # this line too. The point is not the count, it is that the list only ever grows
@@ -102,7 +102,10 @@
 # `NativeAudioSource`, which is libwebrtc: it returns nothing a test can read,
 # so replacing the worker with nothing at all looks the same from outside as
 # playing the audio. What can be checked is checked, in `OutputAudio::capture`,
-# which is where the framing and the playout pacing are decided.
+# which is where the framing and the playout pacing are decided. The
+# `OutputAudioSink` impl for `NativeAudioSource` is the same libwebrtc call
+# under a trait, there so the worker's ordering of capture and cancellation
+# can be tested against a sink that records it.
 #
 # The rest of the room loop's helpers are here for the same reason: every one
 # of them takes `&Room` or the live Gemini session and answers by writing to it.
@@ -299,6 +302,16 @@
 # builds the Google URL for `generate_phase_judgment_at`, which is tested
 # against a local server; reaching the real one needs a live key.
 #
+# `play_tool_ack` is the room's half of the tool-round acknowledgement: it
+# queues the clip on the live `GeminiEventContext` and stamps the trace, so
+# `Ok(())` looks the same from outside as the work, for the reason
+# `send_model_text` does. Whether it plays is decided by `tool_ack_fits` and
+# `RuntimeActivity::take_tool_ack`, and the trace's stamp by
+# `ReplyLatency::tool_ack`; all three are tested, and the
+# browser check with `CODETRIAL_GEMINI_TOOL_ACK` set is what heard it.
+# `tool_ack_clip` only names the Google URL for `tool_ack_clip_at`, which is
+# tested against a local server, as `generate_phase_judgment_with_keys` does.
+#
 # `log` in the latency module and `log_first_frame` beside the output worker
 # only write a line to stderr; `()` leaves nothing a test can read. What the
 # line holds is built and tested apart from the printing: the trace's own
@@ -336,6 +349,7 @@ exclude_re = [
     "send_model_text",
     "publish_output_audio",
     "output_audio_worker",
+    "impl OutputAudioSink for NativeAudioSource",
     "publish_framework_progress",
     "send_wrap_up_and_wait",
     "drop_stale_playout",
@@ -376,6 +390,8 @@ exclude_re = [
     "settle_phase_judge",
     "flush_framework_progress",
     "generate_phase_judgment_with_keys",
+    "play_tool_ack",
+    "replace tool_ack_clip ->",
     "livekit/latency\\.rs:.*replace log with \\(\\)$",
     "replace log_first_frame with \\(\\)$",
 ]

README.md

@@ -282,6 +282,7 @@ The common ones:
 | `CODETRIAL_MAX_INTERIM_REVIEWS` | `6` | Quiet-pause report-model reviews per interview; `0` disables them and `72` is the maximum |
 | `CODETRIAL_GEMINI_REPLY_TIMEOUT_S` | `45` | Seconds a started interviewer reply may go without further output before the Live socket is replaced (20–120); see [degradation controls](docs/provider-cost-and-degradation.md) |
 | `CODETRIAL_GEMINI_FIRST_OUTPUT_S` | `10` | Seconds an owed reply that nothing has started toward may go with nothing back before the Live socket is replaced (5–120, never above the reply timeout) |
+| `CODETRIAL_GEMINI_TOOL_ACK` | `false` | Experimental: while a tool call holds back a reply the candidate is waiting for, play a short acknowledgement generated once per interview in the interviewer's voice; see [latency observability](docs/latency-observability.md) |
 | `CODETRIAL_GEMINI_CANDIDATE_VIDEO_ENABLED` | `false` | Forward candidate video to Gemini, one low-resolution frame in five seconds; ignored in a whiteboard interview |
 | `CODETRIAL_COMPILER_EXPLORER_ENABLED` | `true` | Enable remote C, C++, and Java runs |
 | `CODETRIAL_MAX_CONCURRENT_INTERVIEWS` | `16` | Interviews one `web` process hosts agents for |

config/codetrial.env.example

@@ -23,6 +23,9 @@ CODETRIAL_WEB_DIR=web
 CODETRIAL_WEB_ADDR=127.0.0.1:3000
 CODETRIAL_COMPILER_EXPLORER_ENABLED=true
 CODETRIAL_GEMINI_CANDIDATE_VIDEO_ENABLED=false
+# Experimental: play a short spoken acknowledgement in the interviewer's voice
+# while a tool call holds back the reply; one speech-model call per interview.
+# CODETRIAL_GEMINI_TOOL_ACK=false
 # Optional pair for Live context compression experiments. Unset keeps the
 # provider defaults; smaller windows discard older dialogue. See
 # docs/provider-cost-and-degradation.md before choosing production values, and

docs/latency-observability.md

@@ -57,6 +57,20 @@ barge-in that caused it is the input still waiting for an answer.
 
 ## Waterfall
 
+With `CODETRIAL_GEMINI_TOOL_ACK` set, a tool round that holds back the reply a
+candidate is waiting for, and has heard nothing of, queues a short
+acknowledgement generated once per interview in the session's voice and accent
+(`gemini-3.1-flash-tts-preview`, on the report keys). It plays at most once per
+thing the candidate said, never while they are speaking again, and never
+before a goodbye. It is not the reply: the watchdog and this record still wait
+for the reply's own audio. Its first accepted chunk cancels the remaining clip
+and clears buffered filler before the reply plays. Audio already transmitted
+cannot be recalled. Cancelled queue entries are skipped without capture;
+`output_queue_frames` can include them until the worker drains them. The record
+stamps it as `tool_ack_ms`, and the summary's `tool_ack_lead` is the reply's native
+capture less that stamp: how much sooner the candidate heard something. An
+interview whose clip is not ready yet, or failed to generate, plays nothing.
+
 Offsets share the origin and use the agent's monotonic clock. A milestone seen
 before an origin that later moved is negative, and one never observed is
 `null`. `observed_ms` may be negative when provider receipt precedes the first

scripts/analyze-latency.py

@@ -42,6 +42,7 @@
     "candidate_yield_ms",
     "first_provider_output_ms",
     "first_tool_call_ms",
+    "tool_ack_ms",
     "last_tool_response_ms",
     "tool_handler_ms",
     "provider_event_queue_ms",
@@ -470,6 +471,17 @@ def ordering(event):
         "beyond_window_with_stalls": statistics(
             [r["beyond_window_ms"] for r in comparable]
         ),
+        # With an acknowledgement played during a tool round, the candidate
+        # hears something before the reply: how much sooner. The clip went out
+        # on an empty queue, so its queueing is its capture; the reply's is its
+        # native capture, which waited behind the clip when the clip was longer.
+        "tool_ack_lead": statistics(
+            [
+                r["server_total_ms"] - r["first_audio"]["tool_ack_ms"]
+                for r in typical
+                if r["first_audio"].get("tool_ack_ms") is not None
+            ]
+        ),
         "beyond_window_by_kind": {
             kind: statistics(
                 [r["beyond_window_ms"] for r in typical if reply_kind(r) == kind]
@@ -630,6 +642,7 @@ def summary_lines(report):
         "beyond_window",
         "beyond_window_with_stalls",
         "transcript_before_end",
+        "tool_ack_lead",
     ):
         if report[label]["count"]:
             out.append(stat_line(label, report[label]))

src/config.rs

@@ -67,6 +67,9 @@ pub const DEFAULT_WEB_DIR: &str = "web";
 pub const DEFAULT_WEB_ADDR: &str = "127.0.0.1:3000";
 pub const DEFAULT_COMPILER_EXPLORER_ENABLED: bool = true;
 pub const DEFAULT_GEMINI_CANDIDATE_VIDEO_ENABLED: bool = false;
+/// Off until measured in real interviews: a short spoken acknowledgement
+/// played while a tool round holds the reply back. See `gemini::tool_ack_clip`.
+pub const DEFAULT_GEMINI_TOOL_ACK: bool = false;
 
 /// How many interviews one `codetrial web` process will host agents for at
 /// once. Each is a LiveKit room plus a metered Gemini Live session, so an
@@ -564,6 +567,7 @@ pub struct AgentConfig {
     pub room_prefix: String,
     pub default_duration_min: u32,
     pub gemini_candidate_video_enabled: bool,
+    pub gemini_tool_ack: bool,
     pub max_interim_reviews: usize,
     pub gemini_reply_timeout_s: u32,
     pub gemini_first_output_s: u32,
@@ -782,6 +786,10 @@ pub fn load_from_pairs(
                 .get("CODETRIAL_GEMINI_CANDIDATE_VIDEO_ENABLED")
                 .map(String::as_str),
         ),
+        gemini_tool_ack: parse_bool(
+            values.get("CODETRIAL_GEMINI_TOOL_ACK").map(String::as_str),
+            DEFAULT_GEMINI_TOOL_ACK,
+        ),
         max_interim_reviews: optional_u32(
             &values,
             "CODETRIAL_MAX_INTERIM_REVIEWS",

src/gemini.rs

@@ -1528,6 +1528,25 @@ async fn generate_content_once(
     timeout: Duration,
     what: &str,
 ) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
+    let response = post_generate_content(api_key, url, request, timeout, what).await?;
+    gemini_text(&response).ok_or_else(|| {
+        io::Error::new(
+            io::ErrorKind::InvalidData,
+            format!("Gemini {what} response had no text"),
+        )
+        .into()
+    })
+}
+
+/// The transport half of `generate_content_once`, for a caller that reads
+/// something other than text out of the answer.
+async fn post_generate_content(
+    api_key: &str,
+    url: &str,
+    request: &Value,
+    timeout: Duration,
+    what: &str,
+) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
     let started = std::time::Instant::now();
     let response = crate::http_client()
         .post(url)
@@ -1553,13 +1572,7 @@ async fn generate_content_once(
         .map(TokenUsage::from_metadata)
         .unwrap_or_default();
     eprintln!("gemini {what} usage {}", usage.log_fields());
-    gemini_text(&response).ok_or_else(|| {
-        io::Error::new(
-            io::ErrorKind::InvalidData,
-            format!("Gemini {what} response had no text"),
-        )
-        .into()
-    })
+    Ok(response)
 }
 
 async fn generate_report_once(
@@ -1750,6 +1763,86 @@ fn live_websocket_url_at(endpoint: &str, api_key: &str) -> String {
     format!("{endpoint}?key={}", percent_encode_component(api_key))
 }
 
+/// The model and words of the acknowledgement a tool round may play. Generated
+/// once per interview in the session's own voice and accent, so it sounds like
+/// Jim.
+const TOOL_ACK_TTS_MODEL: &str = "gemini-3.1-flash-tts-preview";
+const TOOL_ACK_WORDS: &str = "Mm, okay.";
+const TOOL_ACK_TIMEOUT: Duration = Duration::from_secs(20);
+
+/// What the speech model is asked to say, in the interview's accent when it
+/// has one.
+fn tool_ack_text(accent: Option<&str>) -> String {
+    let accent = accent
+        .map(|accent| format!(", with a natural {accent} accent"))
+        .unwrap_or_default();
+    format!("Say briefly, in a natural, thinking tone{accent}: {TOOL_ACK_WORDS}")
+}
+
+/// A short spoken acknowledgement in `voice`, as 16-bit mono PCM at
+/// `sample_rate`, the output track's own format. One attempt on the report
+/// keys: an interview without it only loses a filler.
+pub(crate) async fn tool_ack_clip(
+    keys: &GeminiKeys,
+    voice: &str,
+    accent: Option<&str>,
+    sample_rate: u32,
+) -> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
+    let url = gemini_generate_content_url(TOOL_ACK_TTS_MODEL);
+    tool_ack_clip_at(keys, &url, voice, accent, sample_rate).await
+}
+
+async fn tool_ack_clip_at(
+    keys: &GeminiKeys,
+    url: &str,
+    voice: &str,
+    accent: Option<&str>,
+    sample_rate: u32,
+) -> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
+    let api_key = keys.select_report()?;
+    let request = json!({
+        "contents": [{ "parts": [{ "text": tool_ack_text(accent) }] }],
+        "generationConfig": {
+            "responseModalities": ["AUDIO"],
+            "speechConfig": { "voiceConfig": { "prebuiltVoiceConfig": { "voiceName": voice } } }
+        }
+    });
+    let response = post_generate_content(
+        &api_key,
+        url,
+        &request,
+        TOOL_ACK_TIMEOUT,
+        "tool acknowledgement",
+    )
+    .await?;
+    tool_ack_pcm(&response, sample_rate)
+}
+
+/// The PCM in a speech response, refusing anything but raw 16-bit samples at
+/// the output track's rate: a container or a codec would play as noise.
+fn tool_ack_pcm(
+    response: &Value,
+    sample_rate: u32,
+) -> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
+    let data = &response["candidates"][0]["content"]["parts"][0]["inlineData"];
+
+    // Media types are case-insensitive, and the speech model answers in lower
+    // case: "audio/l16; rate=24000; channels=1".
+    let mime = data["mimeType"].as_str().unwrap_or_default();
+    let lower = mime.to_ascii_lowercase();
+    let raw = lower.starts_with("audio/l16") || lower.starts_with("audio/pcm");
+    let rate = format!("rate={sample_rate}");
+    if !raw || !lower.split(';').any(|param| param.trim() == rate) {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidData,
+            format!("acknowledgement audio is {mime:?}, not {sample_rate} Hz PCM"),
+        )
+        .into());
+    }
+    let encoded = data["data"].as_str().unwrap_or_default();
+    Ok(STANDARD.decode(encoded)?)
+}
+
 /// No `?key=` here on purpose. A `reqwest` error Displays the URL it was built
 /// from, and this call's errors reach the candidate's browser in the report
 /// failure note, so the credential travels in a header instead.

src/livekit.rs

@@ -1555,6 +1555,31 @@ async fn open_session<'a>(
         boot.room_name, boot.problem.id, boot.duration_min
     );
 
+    // Generated beside the setup below and picked up whenever it lands: a tool
+    // round before it is ready simply plays nothing.
+    let tool_ack = Arc::new(std::sync::OnceLock::new());
+    let mut tool_ack_task = None;
+    if config.gemini_tool_ack {
+        let (keys, slot) = (keys.clone(), tool_ack.clone());
+        let (voice, accent) = (boot.voice.to_string(), boot.accent.map(str::to_string));
+        let task = tokio::spawn(async move {
+            match crate::gemini::tool_ack_clip(
+                &keys,
+                &voice,
+                accent.as_deref(),
+                GEMINI_OUTPUT_AUDIO_SAMPLE_RATE,
+            )
+            .await
+            {
+                Ok(pcm) => {
+                    let _ = slot.set(media::trim_silence(&pcm, GEMINI_OUTPUT_AUDIO_SAMPLE_RATE));
+                }
+                Err(error) => eprintln!("tool acknowledgement not generated: {error}"),
+            }
+        });
+        tool_ack_task = Some(media::AbortOnDrop(task.abort_handle()));
+    }
+
     // Only the Gemini session overlaps, and only because it touches nothing in
     // the room. Isolation and publication cannot: isolation exists so that one
     // agent is in the room when the candidate starts subscribing, and running
@@ -1568,7 +1593,7 @@ async fn open_session<'a>(
     let mut restarts = 0;
     let mut room_ready = Duration::ZERO;
     let mut gemini_ready = Duration::ZERO;
-    let (output_audio, mut gemini) = tokio::try_join!(
+    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, GEMINI_OUTPUT_AUDIO_SAMPLE_RATE).await;
@@ -1595,6 +1620,9 @@ async fn open_session<'a>(
         },
     )?;
 
+    output_audio.tool_ack = tool_ack;
+    output_audio.tool_ack_task = tool_ack_task;
+
     // Which branch held the greeting back: an earlier Gemini connect can only
     // help when Gemini is the later of the two.
     eprintln!(

src/livekit/latency.rs

@@ -95,6 +95,8 @@ struct Trace {
     yielded: Option<Instant>,
     first_output: Option<Instant>,
     first_tool_call: Option<Instant>,
+    /// A local acknowledgement queued while a tool round held the reply.
+    tool_ack: Option<Instant>,
     last_tool_response: Option<Instant>,
     tool_batches: u64,
     tool_handler_ms: u64,
@@ -262,6 +264,7 @@ impl ReplyLatency {
             yielded: None,
             first_output: None,
             first_tool_call: None,
+            tool_ack: None,
             last_tool_response: None,
             tool_batches: 0,
             tool_handler_ms: 0,
@@ -627,6 +630,16 @@ impl ReplyLatency {
         }
     }
 
+    /// The candidate hears something before the reply itself: the first
+    /// sound, which the reply's own audio record still measures past.
+    pub(super) fn tool_ack(&mut self, now: Instant) {
+        if let Some(trace) = self.trace.as_mut()
+            && !trace.audio_accepted
+        {
+            trace.tool_ack.get_or_insert(now);
+        }
+    }
+
     /// A batch that failed to send is recorded but does not move the
     /// continuation's start: nothing reached the model for it to resume on.
     pub(super) fn tool_response(&mut self, now: Instant, handler_ms: u64, sent: bool) -> Value {
@@ -736,6 +749,7 @@ impl ReplyLatency {
             "candidate_yield_ms": at(trace.yielded),
             "first_provider_output_ms": at(trace.first_output),
             "first_tool_call_ms": at(trace.first_tool_call),
+            "tool_ack_ms": at(trace.tool_ack),
             "last_tool_response_ms": at(trace.last_tool_response),
             "tool_batches": trace.tool_batches,
             "tool_handler_ms": trace.tool_handler_ms,