Teams meetings: mark the newest speaker, and log more of a nameless participant

docs/research/teams-calls.md

@@ -1612,7 +1612,7 @@ last), the meeting took them (`sendonly`) but then neither answered our
 sent its acceptance twice. Without them, both worked. Until a recording
 shows what else the web client does, we offer the one camera line. It
 follows whoever speaks with a camera on, by the data channel's `dsh`
-(the speakers' audio source ids, newest first), matched to the roster's
+(the speakers' audio source ids, the newest last), matched to the roster's
 `main-audio` `sourceId`.
 
 **Several cameras, recorded (`teams.live.com10.har` and its data channel
@@ -1639,9 +1639,14 @@ for one camera on each.
 
 **Who speaks.** The meeting mixes everyone's sound into one stream, with
 no audio level header, so loudness alone says that *someone* speaks. The
-data channel's `dsh` says who spoke last (newest first, by audio source
-id). The bar and the tiles mark that person as speaking while the mixed
-sound is loud, as Teams' own speaking ring roughly does.
+data channel's `dsh` says who spoke (a history, the newest last, by audio
+source id), but it came **once**, right after the `ack`, and never again
+in a meeting of two minutes with people talking: a snapshot on joining.
+The audio line's SDP offers `dsh` as an RTCP application message too
+(`a=rtcp-fb:* x-message app send:dsh,x-gain recv:dsh,x-gain`), and the
+call registers a `csrcInfo` callback for the media controller; both are
+now logged, to find the live signal. The bar and the tiles mark the last
+speaker known as speaking while the mixed sound is loud.
 
 ### H.10 Sharing a screen in a meeting (recorded: `teams.live.com11.har` and its data channel log)
 

src/teams/calling/call.rs

@@ -189,11 +189,17 @@ impl People {
                 && !self.by_mri.contains_key(mri)
                 && let Some(details) = &them.details
             {
+                let devices: Vec<serde_json::Value> = them
+                    .endpoints
+                    .values()
+                    .map(|e| serde_json::Value::Object(e.rest.clone().into_iter().collect()))
+                    .collect();
                 log::info!(
-                    "Teams meeting: no name for {mri} ({} participant; role {:?}; details also say {})",
+                    "Teams meeting: no name for {mri} ({} participant; role {:?}; details also say {}; its devices {})",
                     details.participant_type.as_deref().unwrap_or("untyped"),
                     them.role,
                     serde_json::Value::Object(details.rest.clone().into_iter().collect()),
+                    serde_json::Value::Array(devices),
                 );
             }
             let attendee = Attendee {
@@ -300,6 +306,7 @@ async fn run_meeting(
         watching_share: None,
         syns: 0,
         heard: 0,
+        feedback: 0,
         sharing: false,
         endpoint: endpoint.clone(),
         own_camera: None,
@@ -570,6 +577,8 @@ struct InMeeting {
     syns: u32,
     /// How many messages came on it since it opened.
     heard: u32,
+    /// How many application messages came in RTCP.
+    feedback: u32,
     /// Whether we share our screen, as the meeting was last told.
     sharing: bool,
     /// Our endpoint id, by which the roster lists our own streams.
@@ -891,6 +900,11 @@ impl Call {
                         }
                     }
                     Some(MediaEvent::ChannelData(message)) => self.on_channel(&message, session, tell),
+                    Some(MediaEvent::Feedback {
+                        sender,
+                        media,
+                        payload,
+                    }) => self.on_feedback(sender, media, &payload, session, tell),
                     Some(MediaEvent::AudioFlowing) => {
                         log::info!("Teams call: audio flows both ways");
                         tell(CallEvent::AudioFlowing);
@@ -1360,6 +1374,11 @@ impl Call {
                     self.offer_share(local).await;
                 }
             }
+            Push::CsrcInfo(body) => {
+                let text = body.to_string();
+                let cut = text.char_indices().nth(2000).map_or(text.len(), |(i, _)| i);
+                log::info!("Teams meeting: csrcInfo: {}", &text[..cut]);
+            }
             Push::Other(name) => log::debug!("Teams call: push {name} not acted on"),
         }
         None
@@ -1573,30 +1592,93 @@ impl Call {
                     }
                 }
                 super::channel::Heard::Speakers(history) => {
-                    let Some(meeting) = &mut self.meeting else {
-                        return;
-                    };
-                    // The newest first; ourselves (not in the list) aside.
-                    let speaker = history.iter().find_map(|&source| {
-                        meeting
-                            .people
-                            .list()
-                            .into_iter()
-                            .find(|a| a.audio == Some(source))
-                            .map(|a| a.mri)
-                    });
-                    if speaker.is_some() && speaker != meeting.speaker {
-                        meeting.speaker = speaker.clone();
-                        if let Some(mri) = speaker {
-                            tell(CallEvent::Speaker(mri));
-                        }
-                        self.ask_for_video(session, tell);
-                    }
+                    self.spoke(&history, "the data channel", session, tell);
                 }
             }
         }
     }
 
+    /// Who spoke, as a speaker history (`dsh`) from `way` says: the
+    /// newest last (seen: the first marked whoever spoke long ago);
+    /// ourselves (not in the list) aside.
+    fn spoke(
+        &mut self,
+        history: &[i64],
+        way: &str,
+        session: &MediaSession,
+        tell: &(impl Fn(CallEvent) + Send),
+    ) {
+        let Some(meeting) = &mut self.meeting else {
+            return;
+        };
+        let speaker = history.iter().rev().find_map(|&source| {
+            meeting
+                .people
+                .list()
+                .into_iter()
+                .find(|a| a.audio == Some(source))
+                .map(|a| a.mri)
+        });
+        log::info!(
+            "Teams meeting: spoke, by audio source ({way}): {history:?}; marking {}",
+            speaker.as_deref().unwrap_or("no one")
+        );
+        if speaker.is_some() && speaker != meeting.speaker {
+            meeting.speaker = speaker.clone();
+            if let Some(mri) = speaker {
+                tell(CallEvent::Speaker(mri));
+            }
+            self.ask_for_video(session, tell);
+        }
+    }
+
+    /// An application message in RTCP: the audio line offers `dsh` this
+    /// way too. Its shape is not recorded, so the first few are logged
+    /// whole; one with a JSON `dsh` in it counts as a speaker history.
+    fn on_feedback(
+        &mut self,
+        sender: u32,
+        media: u32,
+        payload: &[u8],
+        session: &MediaSession,
+        tell: &(impl Fn(CallEvent) + Send),
+    ) {
+        let Some(meeting) = &mut self.meeting else {
+            return;
+        };
+        meeting.feedback += 1;
+        if meeting.feedback <= 5 || meeting.feedback % 100 == 0 {
+            let head: Vec<String> = payload
+                .iter()
+                .take(32)
+                .map(|b| format!("{b:02x}"))
+                .collect();
+            log::info!(
+                "Teams meeting: RTCP application message {} from {sender} about {media}, {} bytes: {} / {:?}",
+                meeting.feedback,
+                payload.len(),
+                head.join(" "),
+                String::from_utf8_lossy(payload)
+            );
+        }
+        // JSON, if any, from its first brace.
+        let Some(start) = payload.iter().position(|&b| b == b'{') else {
+            return;
+        };
+        let Ok(message) = serde_json::from_slice::<serde_json::Value>(&payload[start..]) else {
+            return;
+        };
+        if message.get("type").and_then(|t| t.as_str()) != Some("dsh") {
+            return;
+        }
+        let history: Vec<i64> = message
+            .get("history")
+            .and_then(|h| h.as_array())
+            .map(|h| h.iter().filter_map(serde_json::Value::as_i64).collect())
+            .unwrap_or_default();
+        self.spoke(&history, "RTCP", session, tell);
+    }
+
     /// Asks the meeting for the cameras to show, where that changed: one
     /// camera on of those in the call per camera line, each kept on its
     /// line while it stays on; none on a line once no one is left for it.
@@ -2113,7 +2195,21 @@ mod tests {
         let details = bot.details.as_mut().expect("details");
         details.hidden = true;
         details.display_name = None;
-        update.participants.insert("28:recorder".into(), bot);
+        update
+            .participants
+            .insert("28:recorder".into(), bot.clone());
+        // The recorder says so on its device instead (recorded).
+        let mut transcriber = bot;
+        transcriber.details.as_mut().expect("details").hidden = false;
+        for endpoint in transcriber.endpoints.values_mut() {
+            endpoint.rest.insert(
+                "endpointMetadata".into(),
+                serde_json::json!({"__platform": {"ui": {"hidden": true}}}),
+            );
+        }
+        update
+            .participants
+            .insert("28:transcriber".into(), transcriber);
         let mut people = People::default();
         assert!(!people.take(&update, "8:live:organizer"));
         assert!(people.list().is_empty());

src/teams/calling/channel.rs

@@ -145,7 +145,7 @@ pub enum Heard {
     Ready,
     /// A source request of ours was answered.
     SourceAnswered { sequence: u64, ok: bool },
-    /// Who spoke last, by their audio's source id, newest first.
+    /// Who spoke, by their audio's source id: a history, the newest last.
     Speakers(Vec<i64>),
 }
 

src/teams/calling/media.rs

@@ -471,6 +471,14 @@ pub enum MediaEvent {
     ChannelOpen,
     /// A message on the meeting's data channel.
     ChannelData(Vec<u8>),
+    /// An application message in RTCP from the far end (`x-message app`:
+    /// the audio line offers `dsh` this way), with the sender's and the
+    /// media's SSRCs.
+    Feedback {
+        sender: u32,
+        media: u32,
+        payload: Vec<u8>,
+    },
     /// The media broke; nothing more comes.
     Failed(Failure),
     /// Stopped as asked.
@@ -1706,6 +1714,13 @@ impl Session {
             RtcEvent::ChannelClose(id) if Some(id) == self.channel => {
                 log::info!("media: the meeting's data channel closed");
             }
+            RtcEvent::AppSpecificFeedback(feedback) => {
+                let _ = self.tell.send(MediaEvent::Feedback {
+                    sender: *feedback.sender_ssrc,
+                    media: *feedback.media_ssrc,
+                    payload: feedback.payload.to_vec(),
+                });
+            }
             _ => {}
         }
     }

src/teams/calling/types.rs

@@ -525,9 +525,12 @@ impl RosterParticipant {
 
     /// Whether the client would not show them: a hidden participant (a
     /// recording or transcription bot, say) is in the roster but not in
-    /// the room.
+    /// the room. The details may say so; a bot's device says so in its
+    /// metadata (recorded: `endpointMetadata.__platform.ui.hidden` on the
+    /// call recorder and on the transcription bot).
     pub fn is_hidden(&self) -> bool {
         self.details.as_ref().is_some_and(|d| d.hidden)
+            || self.endpoints.values().any(RosterEndpoint::is_hidden)
     }
 
     /// Whether the participant is muted: an endpoint of theirs says so.
@@ -625,9 +628,22 @@ pub struct RosterEndpoint {
     /// in, `Call` for one never kept waiting (recorded).
     #[serde(deserialize_with = "nullable")]
     pub modality_joined: String,
+    /// Whatever else the roster says of the device, kept so a participant
+    /// without a name can be logged in full and the name found.
+    #[serde(flatten)]
+    pub rest: BTreeMap<String, serde_json::Value>,
 }
 
 impl RosterEndpoint {
+    /// Whether the client hides this device (a bot's, recorded).
+    pub fn is_hidden(&self) -> bool {
+        self.rest
+            .get("endpointMetadata")
+            .and_then(|m| m.pointer("/__platform/ui/hidden"))
+            .and_then(serde_json::Value::as_bool)
+            .unwrap_or(false)
+    }
+
     /// Whether this device waits in the lobby: it is there, it has never
     /// been in the call, and it is not in it now. One let in and gone
     /// again can still show its lobby media.
@@ -1041,6 +1057,9 @@ pub enum Push {
     /// A renegotiation of ours was refused (`rejection`,
     /// `mediaNegotiationFailure`), by its event name.
     Refused(String),
+    /// The media controller says which contributing sources (CSRCs) are
+    /// whose; not yet understood, kept whole to be logged.
+    CsrcInfo(serde_json::Value),
     /// A callback a call does not act on (`progress`,
     /// `admitParticipantSuccess`, …), by its event name.
     Other(String),
@@ -1071,6 +1090,7 @@ impl Push {
             "end" => Self::CallEnd(CallEndPush::deserialize(body)?.call_end),
             "conversationEnd" => Self::ConversationEnd(ConversationEnd::deserialize(body)?),
             "rejection" | "mediaNegotiationFailure" => Self::Refused(event.to_owned()),
+            "csrcInfo" => Self::CsrcInfo(body.clone()),
             other => Self::Other(other.to_owned()),
         })
     }