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]); +}