From 1746f1e65a80539ac69370735cccd59e8632be56 Mon Sep 17 00:00:00 2001 From: Jim Huang Date: Mon, 2 Nov 2026 02:20:23 +0800 Subject: [PATCH] Adopt livekit 0.10 SendTextOptions The livekit 0.10.0 bump renamed the options type send_text takes from StreamTextOptions to SendTextOptions, which broke the build on every lane. The constructor and setters are unchanged, so the type is renamed at the two sites that name it. The test Recorder now records sender_identity so a test can check a candidate segment is published under the candidate's identity, and the broken-stream board test covers a length mismatch that reaches drain as an error item. --- src/livekit/room_link.rs | 6 ++++-- src/livekit/session.rs | 6 +++--- tests/unit/livekit/board.rs | 4 ++++ tests/unit/livekit/transcripts.rs | 24 ++++++++++++++++++++++++ 4 files changed, 35 insertions(+), 5 deletions(-) diff --git a/src/livekit/room_link.rs b/src/livekit/room_link.rs index 4c331d1bd..58b5693fd 100644 --- a/src/livekit/room_link.rs +++ b/src/livekit/room_link.rs @@ -16,7 +16,7 @@ use std::collections::HashMap; -use ::livekit::data_stream::api::StreamTextOptions; +use ::livekit::data_stream::api::SendTextOptions; use ::livekit::prelude::{DataPacket, LocalParticipant, Room}; type Error = Box; @@ -153,7 +153,7 @@ impl ParticipantLink { pub(super) async fn send_text( &self, text: &str, - options: StreamTextOptions, + options: SendTextOptions, ) -> Result<(), Error> { match self { Self::Live(participant) => { @@ -166,6 +166,7 @@ impl ParticipantLink { topic: options.topic, text: text.to_string(), attributes: options.attributes, + sender_identity: options.sender_identity.map(|identity| identity.to_string()), }); Ok(()) } @@ -186,6 +187,7 @@ pub(super) enum Sent { topic: String, text: String, attributes: HashMap, + sender_identity: Option, }, Left, } diff --git a/src/livekit/session.rs b/src/livekit/session.rs index b6c374f79..2b5b4d2bc 100644 --- a/src/livekit/session.rs +++ b/src/livekit/session.rs @@ -24,7 +24,7 @@ pub(super) use recovery::{ use std::collections::HashMap; use std::time::{Duration, Instant}; -use ::livekit::data_stream::api::StreamTextOptions; +use ::livekit::data_stream::api::SendTextOptions; use crate::agent::{ CANDIDATE_SPEAKER, INTERVIEWER_SPEAKER, ModelInputKind, RecordOutcome, RuntimeState, @@ -2607,8 +2607,8 @@ pub(super) fn transcript_stream_options( segment_id: String, final_segment: bool, sender_identity: Option<&str>, -) -> StreamTextOptions { - let options = StreamTextOptions::new_with_topic(TOPIC_TRANSCRIPTION) +) -> SendTextOptions { + let options = SendTextOptions::new_with_topic(TOPIC_TRANSCRIPTION) .with_attribute("lk.segment_id", &segment_id) .with_attribute( "lk.transcription_final", diff --git a/tests/unit/livekit/board.rs b/tests/unit/livekit/board.rs index 02fe25653..1be469dbe 100644 --- a/tests/unit/livekit/board.rs +++ b/tests/unit/livekit/board.rs @@ -342,6 +342,10 @@ async fn a_broken_or_empty_stream_is_no_board() { drained(vec![Ok(vec![1, 2]), Err("reset")]).await.is_none(), "half a JPEG is not shown" ); + + // A length mismatch against the declared total reaches `drain` as an error + // item, with no bytes or after all of them. + assert!(drained(vec![Err("length exceeded")]).await.is_none()); assert!(drained(Vec::new()).await.is_none()); } diff --git a/tests/unit/livekit/transcripts.rs b/tests/unit/livekit/transcripts.rs index ca8c1723a..aefd4fd91 100644 --- a/tests/unit/livekit/transcripts.rs +++ b/tests/unit/livekit/transcripts.rs @@ -88,3 +88,27 @@ async fn a_flush_behind_queued_segments_returns_once_they_are_sent() { .collect(); assert_eq!(sent.last().map(String::as_str), Some("I will use a map")); } + +/// A candidate's own speech is published under their identity, an interviewer +/// turn under the agent's own. +#[tokio::test] +async fn a_segment_is_sent_under_the_identity_it_carries() { + let (room, recorder) = crate::livekit::room_link::RoomLink::recording("room"); + let mut writer = TranscriptWriter::default(); + let mut candidate = segment("candidate-1", "I will use a map", true); + candidate.identity = Some("candidate-fixed".into()); + writer.publish(&room, candidate); + writer.publish(&room, segment("interviewer-1", "Go ahead.", true)); + writer.flush(Duration::from_secs(5)).await; + let identities: Vec<_> = recorder + .take() + .into_iter() + .filter_map(|sent| match sent { + crate::livekit::room_link::Sent::Text { + sender_identity, .. + } => Some(sender_identity), + _ => None, + }) + .collect(); + assert_eq!(identities, [Some("candidate-fixed".to_string()), None]); +}