feat: animate local speaker ring from MoQ frame transmission

Wire the local user's avatar into the existing speaking-ring animation so
the host/speaker sees the same green ring + amplitude glow on their own
cell that remote speakers already get.

Ground truth is the broadcaster's `publisher.send(opus)` success: the new
`onLevel` callback fires only after a frame actually leaves on the wire,
computed from the raw PCM peak (shared `peakAmplitude` util used by both
the player decode loop and the broadcaster capture loop). This means the
animation reflects what the relay sees, not any UI-side mute / role
state that could be stale or buggy — if frames are flowing, the ring
plays.

Plumbing: `NestsSpeaker.startBroadcasting` gains an optional `onLevel`
that the moq-lite + IETF broadcasters both honour, the reconnecting
wrapper replays it on every session reissue (alongside the existing
mute-intent replay), and `NestViewModel.startBroadcast` hands it a
lambda that calls the existing `onSpeakerActivity` / `onAudioLevel`
plumbing keyed on the local pubkey. The 250 ms speaking-timeout fades
the ring naturally when the user mutes or stops broadcasting.

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt

@@ -65,7 +65,7 @@ class MoqLiteNestsSpeaker internal constructor(
     private val gate = Mutex()
     private var activeHandle: MoqLiteBroadcastHandle? = null
 
-    override suspend fun startBroadcasting(): BroadcastHandle {
+    override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle {
         gate.withLock {
             val current = state.value
             check(current is NestsSpeakerState.Connected) {
@@ -104,6 +104,7 @@ class MoqLiteNestsSpeaker internal constructor(
                                 // and the room is silently mute.
                                 onBroadcastTerminalFailure()
                             },
+                            onLevel = onLevel,
                         )
                     }
                 } catch (t: Throwable) {

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt

@@ -260,7 +260,7 @@ private fun failedSpeaker(state: MutableStateFlow<NestsSpeakerState>): NestsSpea
     object : NestsSpeaker {
         override val state = state
 
-        override suspend fun startBroadcasting(): BroadcastHandle = error("speaker never connected: ${state.value}")
+        override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle = error("speaker never connected: ${state.value}")
 
         override suspend fun close() {
             if (state.value !is NestsSpeakerState.Closed) {

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt

@@ -51,11 +51,20 @@ interface NestsSpeaker {
      * Begin announcing our speaker track and pumping mic frames out as
      * OBJECT_DATAGRAMs.
      *
+     * [onLevel] fires once per Opus frame that successfully reaches
+     * the wire, with the peak amplitude of the underlying PCM frame
+     * normalized to `[0, 1]`. Default no-op so callers that don't
+     * render a local-speaking ring pay zero cost. The callback runs
+     * on the speaker's coroutine scope (typically the caller's
+     * `viewModelScope`) so the consumer must keep its handler
+     * lightweight. See [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster.start]
+     * for the ground-truth-vs-UI-state rationale.
+     *
      * @throws IllegalStateException if a broadcast is already running on
      *   this speaker or the session is not [NestsSpeakerState.Connected].
      * @throws MoqProtocolException if the peer rejects the ANNOUNCE.
      */
-    suspend fun startBroadcasting(): BroadcastHandle
+    suspend fun startBroadcasting(onLevel: (Float) -> Unit = { /* no-op */ }): BroadcastHandle
 
     /** Tear down the MoQ session + transport. Idempotent. */
     suspend fun close()
@@ -151,7 +160,7 @@ class DefaultNestsSpeaker internal constructor(
     private val gate = Mutex()
     private var activeHandle: DefaultBroadcastHandle? = null
 
-    override suspend fun startBroadcasting(): BroadcastHandle {
+    override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle {
         gate.withLock {
             val current = state.value
             check(current is NestsSpeakerState.Connected) {
@@ -174,7 +183,7 @@ class DefaultNestsSpeaker internal constructor(
                     publisher = publisher,
                     scope = scope,
                 )
-            broadcaster.start()
+            broadcaster.start(onLevel = onLevel)
             mutableState.value =
                 NestsSpeakerState.Broadcasting(
                     room = current.room,

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt

@@ -273,7 +273,7 @@ private class ReconnectingSpeakerHandle(
 
     @Volatile private var activeBroadcast: ReissuingBroadcastHandle? = null
 
-    override suspend fun startBroadcasting(): BroadcastHandle =
+    override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle =
         gate.withLock {
             check(state.value !is NestsSpeakerState.Closed) {
                 "startBroadcasting on a closed speaker"
@@ -291,7 +291,7 @@ private class ReconnectingSpeakerHandle(
                 ?: error("no live session — wait for state == Connected before startBroadcasting")
 
             val handle =
-                ReissuingBroadcastHandle(activeSpeaker, scope) { closed ->
+                ReissuingBroadcastHandle(activeSpeaker, scope, onLevel) { closed ->
                     if (activeBroadcast === closed) activeBroadcast = null
                 }
             handle.start()
@@ -325,6 +325,13 @@ private class ReconnectingSpeakerHandle(
 private class ReissuingBroadcastHandle(
     private val activeSpeaker: StateFlow<NestsSpeaker?>,
     private val scope: CoroutineScope,
+    /**
+     * Forwarded to every underlying [NestsSpeaker.startBroadcasting]
+     * the pump opens, so the local-speaking ring keeps animating
+     * across session recycles. Mirrors how [desiredMuted] is
+     * replayed — the user-observed signal is monotonic.
+     */
+    private val onLevel: (Float) -> Unit,
     private val onClose: (ReissuingBroadcastHandle) -> Unit,
 ) : BroadcastHandle {
     @Volatile private var desiredMuted: Boolean = false
@@ -362,7 +369,7 @@ private class ReissuingBroadcastHandle(
                     if (closed) return@collectLatest
                     val handle =
                         try {
-                            sp.startBroadcasting()
+                            sp.startBroadcasting(onLevel)
                         } catch (ce: kotlinx.coroutines.CancellationException) {
                             // Don't `return@collectLatest` on cancel —
                             // propagate so the launched pumpJob actually

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Amplitude.kt

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.nestsclient.audio
+
+/**
+ * Peak amplitude of a 16-bit PCM frame, normalized to `[0, 1]`. Peak
+ * (vs RMS) is jittery on its own but responds instantly to onsets,
+ * which suits a visual ring that the UI smooths via animateDpAsState.
+ *
+ * Shared between [NestPlayer] (decoded remote audio) and the speaker
+ * broadcasters (raw mic capture) so the on-screen ring around the
+ * local avatar uses the same scale as remote speakers' rings.
+ */
+internal fun peakAmplitude(pcm: ShortArray): Float {
+    var maxAbs = 0
+    for (s in pcm) {
+        val abs = if (s.toInt() < 0) -s.toInt() else s.toInt()
+        if (abs > maxAbs) maxAbs = abs
+    }
+    return (maxAbs / 32768f).coerceIn(0f, 1f)
+}

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt

@@ -74,6 +74,8 @@ class NestBroadcaster(
          */
         onTerminalFailure: () -> Unit = { /* swallow */ },
         onError: (AudioException) -> Unit = { /* swallow */ },
+        /** See [NestMoqLiteBroadcaster.start]'s `onLevel` kdoc. */
+        onLevel: (Float) -> Unit = { /* no-op */ },
     ) {
         check(!stopped) { "NestBroadcaster already stopped" }
         check(job == null) { "NestBroadcaster.start already called" }
@@ -120,6 +122,7 @@ class NestBroadcaster(
                         sendOutcome
                             .onSuccess {
                                 consecutiveSendErrors = 0
+                                onLevel(peakAmplitude(pcm))
                             }.onFailure { t ->
                                 if (t is CancellationException) throw t
                                 consecutiveSendErrors += 1

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt

@@ -116,6 +116,23 @@ class NestMoqLiteBroadcaster(
     fun start(
         onTerminalFailure: () -> Unit = { /* swallow */ },
         onError: (AudioException) -> Unit = { /* swallow */ },
+        /**
+         * Fires once per Opus frame that successfully reaches
+         * [MoqLitePublisherHandle.send] — i.e. the frame is on the
+         * wire from this client's perspective. The argument is the
+         * peak amplitude of the underlying PCM frame, normalized to
+         * `[0, 1]` (same scale [NestPlayer] reports for remote
+         * speakers).
+         *
+         * This deliberately taps **after** the mute gate and **after**
+         * a successful send, so the local "I'm speaking" ring on the
+         * UI tracks ground truth (frames actually leaving) rather
+         * than any UI-side mute / role state that could lie.
+         *
+         * Default no-op so callers that don't render a local ring
+         * (tests, headless interop) pay zero cost.
+         */
+        onLevel: (Float) -> Unit = { /* no-op */ },
     ) {
         check(!stopped) { "NestMoqLiteBroadcaster already stopped" }
         check(job == null) { "NestMoqLiteBroadcaster.start already called" }
@@ -180,6 +197,12 @@ class NestMoqLiteBroadcaster(
                         sendOutcome
                             .onSuccess {
                                 consecutiveSendErrors = 0
+                                // Fire the speaking-ring tap only on
+                                // a successful send. If the publisher
+                                // throws (transport gone, peer dead),
+                                // the frame didn't actually go out and
+                                // the UI shouldn't claim we're talking.
+                                onLevel(peakAmplitude(pcm))
                             }.onFailure { t ->
                                 if (t is CancellationException) throw t
                                 consecutiveSendErrors += 1

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt

@@ -112,20 +112,6 @@ class NestPlayer(
             }
     }
 
-    /**
-     * Peak amplitude of a 16-bit PCM frame, normalized to `[0, 1]`. Peak
-     * (vs RMS) is jittery on its own but responds instantly to onsets,
-     * which suits a visual ring that the UI smooths via animateDpAsState.
-     */
-    private fun peakAmplitude(pcm: ShortArray): Float {
-        var maxAbs = 0
-        for (s in pcm) {
-            val abs = if (s.toInt() < 0) -s.toInt() else s.toInt()
-            if (abs > maxAbs) maxAbs = abs
-        }
-        return (maxAbs / 32768f).coerceIn(0f, 1f)
-    }
-
     /**
      * Stop playback, cancel the decode loop, release the decoder. Idempotent.
      *

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeakerTest.kt

@@ -138,7 +138,7 @@ class ReconnectingNestsSpeakerTest {
         val handles: MutableList<ScriptedBroadcastHandle> =
             java.util.concurrent.CopyOnWriteArrayList()
 
-        override suspend fun startBroadcasting(): BroadcastHandle {
+        override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle {
             val handle = ScriptedBroadcastHandle()
             handles += handle
             // Increment AFTER appending to `handles` so the volatile
@@ -176,7 +176,7 @@ class ReconnectingNestsSpeakerTest {
         override val state: StateFlow<NestsSpeakerState> =
             MutableStateFlow(NestsSpeakerState.Failed("scripted-fail")).asStateFlow()
 
-        override suspend fun startBroadcasting(): BroadcastHandle = error("never connected")
+        override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle = error("never connected")
 
         override suspend fun close() {}
     }

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcasterTest.kt

@@ -103,7 +103,7 @@ class NestBroadcasterTest {
             val broadcaster = NestBroadcaster(capture, encoder, publisher, backgroundScope)
             val errors = mutableListOf<AudioException>()
 
-            broadcaster.start { errors.add(it) }
+            broadcaster.start(onError = { errors.add(it) })
             capture.awaitDrained()
 
             assertEquals(1, errors.size, "exactly one encode error should have been reported")