Bound interview token startup and no-show cleanup

src/accounts/interviews.rs

@@ -198,13 +198,16 @@ pub fn interview_for_account(
 ///
 /// The claim is spent before the interviewer is dispatched, so that two
 /// requests racing on one interview cannot both start a room. A dispatcher that
-/// refuses then leaves a consent bound to a room nobody will join, and the
-/// candidate is told to retry a request that would be refused. This is that
-/// retry working.
+/// refuses or a dispatched room whose candidate never arrives would otherwise
+/// leave consent bound to a room nobody will join. This gives that claim back.
 ///
 /// Guarded on the room it is giving back, so a release arriving late cannot
 /// unbind a room some other request has since claimed.
 ///
+/// Returns whether the claim was released. `false` means the room no longer
+/// matches, another pending interview holds this consent version's pending
+/// slot, or recording has started for this interview.
+///
 /// The ceiling: a process that dies between the claim and the release leaves
 /// the interview bound to a room that never ran, and the candidate has to
 /// reload. That is one page reload against a lock nobody can take back, which
@@ -214,16 +217,29 @@ pub fn release_interview_room(
     id: &str,
     account_id: i64,
     room_name: &str,
-) -> rusqlite::Result<()> {
+) -> rusqlite::Result<bool> {
     accounts.with(|connection| {
-        connection.execute(
+        let changed = connection.execute(
             "
         UPDATE interviews SET room_name = NULL
-        WHERE id = ?1 AND account_id = ?2 AND room_name = ?3
+        WHERE id = ?1
+          AND account_id = ?2
+          AND room_name = ?3
+          AND NOT EXISTS (
+              SELECT 1 FROM recordings
+              WHERE recordings.interview_id = interviews.id
+          )
+          AND NOT EXISTS (
+              SELECT 1 FROM interviews AS pending
+              WHERE pending.account_id = interviews.account_id
+                AND pending.consent_version = interviews.consent_version
+                AND pending.room_name IS NULL
+                AND pending.consent_withdrawn_at IS NULL
+          )
         ",
             (id, account_id, room_name),
         )?;
-        Ok(())
+        Ok(changed > 0)
     })
 }
 

src/dispatch.rs

@@ -9,7 +9,7 @@ use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, Mutex};
 
 use crate::config::{AgentConfig, Provider};
-use crate::web::{DispatchRefusal, RoomDispatcher};
+use crate::web::{DispatchRefusal, NoShowConsent, RoomDispatcher};
 
 const MAX_CONCURRENT_GUEST_INTERVIEWS: usize = 2;
 
@@ -46,15 +46,24 @@ impl RoomDispatcher for LocalDispatcher {
     }
 
     fn ensure_agent(&self, room_name: &str, provider: &Provider) -> Result<(), DispatchRefusal> {
-        self.ensure_agent_with_kind(room_name, provider, false)
+        self.ensure_agent_with_kind(room_name, provider, false, None)
+    }
+
+    fn ensure_agent_with_no_show_release(
+        &self,
+        room_name: &str,
+        provider: &Provider,
+        no_show: Option<NoShowConsent>,
+    ) -> Result<(), DispatchRefusal> {
+        self.ensure_agent_with_kind(room_name, provider, false, no_show)
     }
 
     fn ensure_guest_agent(
         &self,
         room_name: &str,
         provider: &Provider,
     ) -> Result<(), DispatchRefusal> {
-        self.ensure_agent_with_kind(room_name, provider, true)
+        self.ensure_agent_with_kind(room_name, provider, true, None)
     }
 }
 
@@ -64,6 +73,7 @@ impl LocalDispatcher {
         room_name: &str,
         provider: &Provider,
         guest: bool,
+        no_show: Option<NoShowConsent>,
     ) -> Result<(), DispatchRefusal> {
         let slot = match self.reserve_with_kind(room_name, guest) {
             Reservation::Existing => return Ok(()),
@@ -93,21 +103,61 @@ impl LocalDispatcher {
         // outlives the response by the length of the interview.
         self.runtime.spawn(async move {
             eprintln!("codetrial dispatch room={}", slot.room_name);
-            if let Err(error) = crate::livekit::run_room_with_slot(
+            let outcome = crate::livekit::run_room_with_slot(
                 &config,
                 &slot.room_name,
                 crate::web::current_epoch_seconds(),
                 Some(&slot),
             )
-            .await
-            {
-                eprintln!("codetrial agent_failed room={}: {error}", slot.room_name);
+            .await;
+            if outcome.as_ref().is_ok_and(no_show_release_required) {
+                release_no_show(no_show, &slot.room_name).await;
+            }
+            match outcome {
+                Ok(crate::livekit::RoomRunOutcome::NoCandidate) => {}
+                Ok(crate::livekit::RoomRunOutcome::FailedBeforeCandidate(error)) => {
+                    eprintln!("codetrial agent_failed room={}: {error}", slot.room_name);
+                }
+                Ok(crate::livekit::RoomRunOutcome::Completed) => {}
+                Err(error) => {
+                    eprintln!("codetrial agent_failed room={}: {error}", slot.room_name);
+                }
             }
         });
         Ok(())
     }
 }
 
+fn no_show_release_required(outcome: &crate::livekit::RoomRunOutcome) -> bool {
+    matches!(
+        outcome,
+        crate::livekit::RoomRunOutcome::NoCandidate
+            | crate::livekit::RoomRunOutcome::FailedBeforeCandidate(_)
+    )
+}
+
+async fn release_no_show(no_show: Option<NoShowConsent>, room_name: &str) -> bool {
+    let Some(no_show) = no_show else {
+        return false;
+    };
+    match no_show.release().await {
+        Ok(true) => true,
+        Ok(false) => {
+            eprintln!(
+                "codetrial consent_release_skipped room={room_name} \
+                 reason=claim_changed_pending_or_recording_started"
+            );
+            false
+        }
+        Err(error) => {
+            eprintln!(
+                "could not release interview consent after a no-show in room={room_name}: {error}"
+            );
+            false
+        }
+    }
+}
+
 enum Reservation {
     Existing,
     New(Slot),

src/livekit.rs

@@ -1531,36 +1531,53 @@ struct OpenSession<'a> {
     restarts: usize,
 }
 
+enum OpenSessionOutcome<'a> {
+    Started(Box<OpenSession<'a>>),
+    NoCandidate,
+    FailedBeforeCandidate(Box<dyn std::error::Error + Send + Sync>),
+}
+
 /// Joins, waits for a candidate, brings up audio and Gemini, and greets.
 ///
 /// Split from [`run_room`] because none of it repeats: it runs once, in order,
 /// and every line of it is a step that either succeeds or ends the interview
 /// before it starts. What is left in `run_room` is the part that runs many
 /// times, which is the part worth reading as a loop.
 ///
-/// `Ok(None)` is nobody joined. That is not an error: the room was named, the
-/// candidate never arrived, and there is nothing to run or report.
+/// `NoCandidate` is nobody joined. That is not an error: the room was named,
+/// the candidate never arrived, and there is nothing to run or report.
 async fn open_session<'a>(
     config: &'a AgentConfig,
     keys: &Arc<GeminiKeys>,
     room_name: &'a str,
     now_seconds: u64,
     session_id: u64,
-) -> Result<Option<OpenSession<'a>>, Box<dyn std::error::Error + Send + Sync>> {
+) -> Result<OpenSessionOutcome<'a>, Box<dyn std::error::Error + Send + Sync>> {
     let agent_identity = agent_identity(room_name);
-    let (room, mut events) = join_room(config, room_name, &agent_identity, now_seconds).await?;
+    let (room, mut events) = match join_room(config, room_name, &agent_identity, now_seconds).await
+    {
+        Ok(joined) => joined,
+        Err(error) => return Ok(OpenSessionOutcome::FailedBeforeCandidate(error)),
+    };
     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?;
+    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 Some((candidate_identity, candidate_metadata)) =
-        candidate_metadata(room.connected(), &mut events, &mut deferred_events).await?
-    else {
+    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(None);
+        return Ok(OpenSessionOutcome::NoCandidate);
     };
     let events = tokio::sync::Mutex::new(events);
     let boot = candidate_bootstrap(config, room_name, Some(&candidate_metadata));
@@ -1699,7 +1716,7 @@ async fn open_session<'a>(
     turn.activity
         .mark_prompted(Instant::now(), Some(&greeting), false);
 
-    Ok(Some(OpenSession {
+    Ok(OpenSessionOutcome::Started(Box::new(OpenSession {
         room,
         events,
         agent_identity,
@@ -1711,7 +1728,7 @@ async fn open_session<'a>(
         turn,
         started_at,
         restarts,
-    }))
+    })))
 }
 
 /// The room loop's own bookkeeping, distinct from the session it runs on.
@@ -2585,15 +2602,45 @@ pub async fn run_room(
     room_name: &str,
     now_seconds: u64,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    run_room_with_slot(config, room_name, now_seconds, None).await
+    match run_room_with_slot(config, room_name, now_seconds, None).await? {
+        RoomRunOutcome::Completed | RoomRunOutcome::NoCandidate => Ok(()),
+        RoomRunOutcome::FailedBeforeCandidate(error) => Err(error),
+    }
+}
+
+#[derive(Debug)]
+pub(crate) enum RoomRunOutcome {
+    Completed,
+    NoCandidate,
+    FailedBeforeCandidate(Box<dyn std::error::Error + Send + Sync>),
+}
+
+fn startup_failure_line(
+    model: &str,
+    room_name: &str,
+    session_id: u64,
+    error: &(dyn std::error::Error + Send + Sync + 'static),
+) -> String {
+    // A socket may have opened before the failure, so no count is claimed; no
+    // generation was asked for, so there is no usage.
+    let outcome = LiveOutcome::from_error(error, LiveOutcome::Error);
+    live_usage_line(
+        room_name,
+        session_id,
+        model,
+        0,
+        outcome,
+        None,
+        &crate::gemini::TokenUsage::default(),
+    )
 }
 
 pub(crate) async fn run_room_with_slot(
     config: &AgentConfig,
     room_name: &str,
     now_seconds: u64,
     slot: Option<&crate::dispatch::Slot>,
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+) -> Result<RoomRunOutcome, Box<dyn std::error::Error + Send + Sync>> {
     // Names this run on every usage line. Milliseconds rather than the
     // dispatch's seconds, so a room retried within the same second is not
     // summed with the run before it.
@@ -2603,7 +2650,36 @@ 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 Some(OpenSession {
+    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 = 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));
+        }
+    };
+    let OpenSession {
         room,
         events,
         agent_identity,
@@ -2615,30 +2691,7 @@ pub(crate) async fn run_room_with_slot(
         mut turn,
         started_at,
         restarts,
-    }) = open_session(config, &keys, room_name, now_seconds, session_id)
-        .await
-        .inspect_err(|error| {
-            // The interview never started, and the summary is where an operator
-            // counts those, a first open a depleted project refused above all.
-            // A socket may have opened before the failure, so no count is
-            // claimed; no generation was asked for, so there is no usage.
-            let outcome = LiveOutcome::from_error(error.as_ref(), LiveOutcome::Error);
-            eprintln!(
-                "{}",
-                live_usage_line(
-                    room_name,
-                    session_id,
-                    &config.gemini_live_model,
-                    0,
-                    outcome,
-                    None,
-                    &crate::gemini::TokenUsage::default(),
-                )
-            );
-        })?
-    else {
-        return Ok(());
-    };
+    } = *session;
 
     let interview = InterviewContext {
         config,
@@ -2910,7 +2963,7 @@ pub(crate) async fn run_room_with_slot(
             &turn.activity.live_usage,
         )
     );
-    result
+    result.map(|()| RoomRunOutcome::Completed)
 }
 
 /// A session's usage summary, in the one spelling the log analyzer reads.