Adopt livekit 0.10 SendTextOptions

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<dyn std::error::Error + Send + Sync>;
@@ -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<String, String>,
+        sender_identity: Option<String>,
     },
     Left,
 }

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",

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());
 }
 

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