feat(audio-rooms): M6 + M7 — AudioRoomBroadcaster + NestsSpeaker API

M6 (AudioRoomBroadcaster): Inverse of AudioRoomPlayer. Pulls PCM frames from an AudioCapture, runs them through an OpusEncoder, and pushes the resulting Opus packets into a MoqSession.TrackPublisher as OBJECT_DATAGRAMs. setMuted keeps the capture + encoder running so unmute is sample-accurate; encode failures are reported via onError but don't tear the loop down. M7 (NestsSpeaker): Mirror of NestsListener for the host / speaker path. - `NestsSpeaker.startBroadcasting()` — sends ANNOUNCE for the room's namespace (`["nests", roomId]`), opens a TrackPublisher named after this user's pubkey hex, wires an AudioRoomBroadcaster onto it. - `BroadcastHandle.setMuted` / `close()` — speaker-side equivalents of the listener's mute / disconnect controls. - `NestsSpeakerState` sealed hierarchy (Idle / Connecting{step} / Connected / Broadcasting{isMuted} / Failed / Closed). - `connectNestsSpeaker` orchestration mirrors `connectNestsListener` end-to-end (HTTP → WebTransport → MoQ setup), then returns a `DefaultNestsSpeaker` ready for `startBroadcasting`. Tests: - AudioRoomBroadcasterTest (5 tests): pcm-flow ordering, mute behavior, encoder warmup skip, encode-error tolerance, idempotent stop. - NestsSpeakerTest (2 tests): start/mute/close state transitions, double-start rejection. Bug caught + fixed during M7 testing: `DefaultNestsSpeaker.close()` held its `gate` mutex while calling `handle.close()` which then called back through `parent.broadcastClosed()` → self-deadlock. Fixed by making `broadcastClosed` lockless (StateFlow updates are atomic; the activeHandle compare-and-set is benign under the gate's exclusion of concurrent startBroadcasting calls). Verified: `:nestsClient:jvmTest` (78 tests, all green) + `./gradlew spotlessApply` clean.
2026-04-26 03:22:02 +00:00
parent 5c32996f93
commit 0afde79afd
5 changed files with 868 additions and 0 deletions
@@ -20,6 +20,8 @@
 */
 package com.vitorpamplona.nestsclient

+import com.vitorpamplona.nestsclient.audio.AudioCapture
+import com.vitorpamplona.nestsclient.audio.OpusEncoder
 import com.vitorpamplona.nestsclient.moq.MoqSession
 import com.vitorpamplona.nestsclient.moq.MoqVersion
 import com.vitorpamplona.nestsclient.moq.SubscribeHandle
@@ -141,6 +143,114 @@ private fun failedListener(state: MutableStateFlow<NestsListenerState>): NestsLi
        }
    }

+/**
+ * Speaker / host counterpart of [connectNestsListener]. Walks the same
+ * HTTP → WebTransport → MoQ handshake; the difference is the post-setup
+ * step is `announce(...)` (driven by [NestsSpeaker.startBroadcasting])
+ * instead of `subscribe(...)`.
+ *
+ * @param speakerPubkeyHex this user's pubkey hex, used as the MoQ track
+ *   name when we ANNOUNCE — listeners look us up by exactly that name.
+ * @param captureFactory builds an [AudioCapture] (one per broadcast).
+ *   Android passes `{ AudioRecordCapture() }`.
+ * @param encoderFactory builds an [OpusEncoder] (one per broadcast).
+ *   Android passes `{ MediaCodecOpusEncoder() }`.
+ */
+suspend fun connectNestsSpeaker(
+    httpClient: NestsClient,
+    transport: WebTransportFactory,
+    scope: CoroutineScope,
+    serviceBase: String,
+    roomId: String,
+    signer: NostrSigner,
+    speakerPubkeyHex: String,
+    captureFactory: () -> AudioCapture,
+    encoderFactory: () -> OpusEncoder,
+    supportedMoqVersions: List<Long> = listOf(MoqVersion.DRAFT_17),
+): NestsSpeaker {
+    val state =
+        MutableStateFlow<NestsSpeakerState>(
+            NestsSpeakerState.Connecting(NestsSpeakerState.Connecting.ConnectStep.ResolvingRoom),
+        )
+
+    val roomInfo =
+        try {
+            httpClient.resolveRoom(serviceBase = serviceBase, roomId = roomId, signer = signer)
+        } catch (t: NestsException) {
+            state.value = NestsSpeakerState.Failed("Room resolution failed: ${t.message}", t)
+            return failedSpeaker(state)
+        }
+
+    state.value = NestsSpeakerState.Connecting(NestsSpeakerState.Connecting.ConnectStep.OpeningTransport)
+
+    val (authority, path) =
+        try {
+            parseEndpoint(roomInfo.endpoint)
+        } catch (t: Throwable) {
+            state.value =
+                NestsSpeakerState.Failed(
+                    "Malformed MoQ endpoint URL '${roomInfo.endpoint}': ${t.message}",
+                    t,
+                )
+            return failedSpeaker(state)
+        }
+
+    val webTransport =
+        try {
+            transport.connect(authority = authority, path = path, bearerToken = roomInfo.token)
+        } catch (t: WebTransportException) {
+            state.value =
+                NestsSpeakerState.Failed(
+                    "WebTransport ${t.kind.name}: ${t.message}",
+                    t,
+                )
+            return failedSpeaker(state)
+        }
+
+    state.value = NestsSpeakerState.Connecting(NestsSpeakerState.Connecting.ConnectStep.MoqHandshake)
+
+    val moq =
+        try {
+            MoqSession.client(webTransport, scope).also { it.setup(supportedMoqVersions) }
+        } catch (t: Throwable) {
+            runCatching { webTransport.close(0, "moq setup failed") }
+            state.value = NestsSpeakerState.Failed("MoQ handshake failed: ${t.message}", t)
+            return failedSpeaker(state)
+        }
+
+    val negotiatedVersion =
+        moq.selectedVersion ?: run {
+            runCatching { moq.close() }
+            state.value = NestsSpeakerState.Failed("MoQ session reported no negotiated version")
+            return failedSpeaker(state)
+        }
+
+    state.value = NestsSpeakerState.Connected(roomInfo, negotiatedVersion)
+    return DefaultNestsSpeaker(
+        session = moq,
+        roomNamespace = TrackNamespace.of("nests", roomId),
+        speakerTrackName = speakerPubkeyHex.encodeToByteArray(),
+        captureFactory = captureFactory,
+        encoderFactory = encoderFactory,
+        scope = scope,
+        mutableState = state,
+    )
+}
+
+/** Mirror of [failedListener] for the speaker path. */
+private fun failedSpeaker(state: MutableStateFlow<NestsSpeakerState>): NestsSpeaker =
+    object : NestsSpeaker {
+        override val state = state
+
+        override suspend fun startBroadcasting(): BroadcastHandle = error("speaker never connected: ${state.value}")
+
+        override suspend fun close() {
+            if (state.value !is NestsSpeakerState.Closed) {
+                state.value = NestsSpeakerState.Closed
+            }
+        }
+    }
+
 /**
 * Split a typical nests endpoint URL such as `https://relay.example.com/moq`
 * or `https://relay.example.com:4443/api/v1/moq?room=abc` into the
@@ -0,0 +1,237 @@
+/*
+ * 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
+
+import com.vitorpamplona.nestsclient.audio.AudioCapture
+import com.vitorpamplona.nestsclient.audio.AudioRoomBroadcaster
+import com.vitorpamplona.nestsclient.audio.OpusEncoder
+import com.vitorpamplona.nestsclient.moq.MoqSession
+import com.vitorpamplona.nestsclient.moq.TrackNamespace
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+
+/**
+ * High-level handle for the speaker / host audio path. Mirror of
+ * [NestsListener]: hides HTTP + WebTransport + MoQ wiring under one
+ * observable state machine, with the added ability to broadcast our own
+ * speaker track.
+ *
+ * Open one [NestsSpeaker] per audio-room screen where we have host or
+ * speaker permission. Call [startBroadcasting] when the user taps "Talk";
+ * call [BroadcastHandle.close] when they stop talking or leave the room.
+ */
+interface NestsSpeaker {
+    /** Connection / broadcast state — drives the speaker UI's status chip. */
+    val state: StateFlow<NestsSpeakerState>
+
+    /**
+     * Begin announcing our speaker track and pumping mic frames out as
+     * OBJECT_DATAGRAMs.
+     *
+     * @throws IllegalStateException if a broadcast is already running on
+     *   this speaker or the session is not [NestsSpeakerState.Connected].
+     * @throws com.vitorpamplona.nestsclient.moq.MoqProtocolException if the
+     *   peer rejects the ANNOUNCE.
+     */
+    suspend fun startBroadcasting(): BroadcastHandle
+
+    /** Tear down the MoQ session + transport. Idempotent. */
+    suspend fun close()
+}
+
+/** Active broadcast on one [NestsSpeaker]. Returned from [NestsSpeaker.startBroadcasting]. */
+interface BroadcastHandle {
+    /**
+     * Toggle whether mic frames reach the wire. The capture pipeline keeps
+     * running, so unmute is sample-accurate. Default unmuted.
+     */
+    suspend fun setMuted(muted: Boolean)
+
+    /** Whether the next OBJECT_DATAGRAM the peer would receive is silenced. */
+    val isMuted: Boolean
+
+    /**
+     * Stop publishing. Releases the mic + encoder, sends UNANNOUNCE +
+     * SUBSCRIBE_DONE on the wire. Idempotent.
+     */
+    suspend fun close()
+}
+
+/** Lifecycle states of a [NestsSpeaker]. */
+sealed class NestsSpeakerState {
+    data object Idle : NestsSpeakerState()
+
+    data class Connecting(
+        val step: ConnectStep,
+    ) : NestsSpeakerState() {
+        enum class ConnectStep {
+            ResolvingRoom,
+            OpeningTransport,
+            MoqHandshake,
+        }
+    }
+
+    /** Connection live; ready for [NestsSpeaker.startBroadcasting]. */
+    data class Connected(
+        val roomInfo: NestsRoomInfo,
+        val negotiatedMoqVersion: Long,
+    ) : NestsSpeakerState()
+
+    /** Currently announcing + emitting OBJECT_DATAGRAMs for our track. */
+    data class Broadcasting(
+        val roomInfo: NestsRoomInfo,
+        val negotiatedMoqVersion: Long,
+        val isMuted: Boolean,
+    ) : NestsSpeakerState()
+
+    data class Failed(
+        val reason: String,
+        val cause: Throwable? = null,
+    ) : NestsSpeakerState()
+
+    data object Closed : NestsSpeakerState()
+}
+
+/**
+ * Default [NestsSpeaker] that wraps a connected [MoqSession] and plumbs an
+ * [AudioRoomBroadcaster] through it on [startBroadcasting].
+ *
+ * Construction does NOT open the transport — call [connectNestsSpeaker] to
+ * walk the full HTTP + transport + MoQ handshake.
+ */
+class DefaultNestsSpeaker internal constructor(
+    private val session: MoqSession,
+    private val roomNamespace: TrackNamespace,
+    private val speakerTrackName: ByteArray,
+    private val captureFactory: () -> AudioCapture,
+    private val encoderFactory: () -> OpusEncoder,
+    private val scope: CoroutineScope,
+    private val mutableState: MutableStateFlow<NestsSpeakerState>,
+) : NestsSpeaker {
+    override val state: StateFlow<NestsSpeakerState> = mutableState.asStateFlow()
+
+    private val gate = Mutex()
+    private var activeHandle: DefaultBroadcastHandle? = null
+
+    override suspend fun startBroadcasting(): BroadcastHandle {
+        gate.withLock {
+            val current = state.value
+            check(current is NestsSpeakerState.Connected) {
+                "startBroadcasting requires Connected state, was $current"
+            }
+            check(activeHandle == null) { "speaker is already broadcasting" }
+
+            val announce = session.announce(roomNamespace)
+            val publisher =
+                try {
+                    announce.openTrack(speakerTrackName)
+                } catch (t: Throwable) {
+                    runCatching { announce.unannounce() }
+                    throw t
+                }
+            val broadcaster =
+                AudioRoomBroadcaster(
+                    capture = captureFactory(),
+                    encoder = encoderFactory(),
+                    publisher = publisher,
+                    scope = scope,
+                )
+            broadcaster.start()
+            mutableState.value =
+                NestsSpeakerState.Broadcasting(
+                    roomInfo = current.roomInfo,
+                    negotiatedMoqVersion = current.negotiatedMoqVersion,
+                    isMuted = false,
+                )
+            val handle =
+                DefaultBroadcastHandle(
+                    broadcaster = broadcaster,
+                    announce = announce,
+                    parent = this,
+                )
+            activeHandle = handle
+            return handle
+        }
+    }
+
+    /**
+     * Lockless: callable from inside [close] (which already holds [gate]) and
+     * from [DefaultBroadcastHandle.close] (which doesn't). We compare-and-set
+     * on activeHandle and update state atomically via the StateFlow.
+     */
+    internal fun broadcastClosed(handle: DefaultBroadcastHandle) {
+        if (activeHandle !== handle) return
+        activeHandle = null
+        val current = mutableState.value
+        if (current is NestsSpeakerState.Broadcasting) {
+            mutableState.value =
+                NestsSpeakerState.Connected(current.roomInfo, current.negotiatedMoqVersion)
+        }
+    }
+
+    internal fun reportMuteState(muted: Boolean) {
+        val current = mutableState.value
+        if (current is NestsSpeakerState.Broadcasting) {
+            mutableState.value = current.copy(isMuted = muted)
+        }
+    }
+
+    override suspend fun close() {
+        gate.withLock {
+            if (state.value is NestsSpeakerState.Closed) return
+            activeHandle?.let { runCatching { it.close() } }
+            activeHandle = null
+            runCatching { session.close() }
+            mutableState.value = NestsSpeakerState.Closed
+        }
+    }
+}
+
+internal class DefaultBroadcastHandle(
+    private val broadcaster: AudioRoomBroadcaster,
+    private val announce: MoqSession.AnnounceHandle,
+    private val parent: DefaultNestsSpeaker,
+) : BroadcastHandle {
+    @Volatile private var muted: Boolean = false
+
+    @Volatile private var closed: Boolean = false
+
+    override val isMuted: Boolean get() = muted
+
+    override suspend fun setMuted(muted: Boolean) {
+        if (closed) return
+        this.muted = muted
+        broadcaster.setMuted(muted)
+        parent.reportMuteState(muted)
+    }
+
+    override suspend fun close() {
+        if (closed) return
+        closed = true
+        runCatching { broadcaster.stop() }
+        runCatching { announce.unannounce() }
+        parent.broadcastClosed(this)
+    }
+}
@@ -0,0 +1,141 @@
+/*
+ * 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
+
+import com.vitorpamplona.nestsclient.moq.MoqSession
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.launch
+
+/**
+ * Inverse of [AudioRoomPlayer]: pulls PCM from an [AudioCapture], runs it
+ * through an [OpusEncoder], and pushes the resulting Opus packets into a
+ * [MoqSession.TrackPublisher] as MoQ OBJECT_DATAGRAMs.
+ *
+ * One instance per outgoing track (typically one per local speaker, since
+ * Opus encoder state is per-stream).
+ *
+ * Lifecycle:
+ *   - [start] opens the mic, starts the capture-encode-publish loop.
+ *   - [setMuted] keeps the loop running but stops emitting frames so unmute
+ *     is instant. Default unmuted.
+ *   - [stop] cancels the loop, stops the mic, releases the encoder, and
+ *     closes the publisher.
+ *
+ * Encoder warm-up frames (encoder returns empty) are dropped silently.
+ * Encode failures are reported via [onError] but do NOT tear the loop down —
+ * one bad frame shouldn't end the broadcast.
+ */
+class AudioRoomBroadcaster(
+    private val capture: AudioCapture,
+    private val encoder: OpusEncoder,
+    private val publisher: MoqSession.TrackPublisher,
+    private val scope: CoroutineScope,
+) {
+    private var job: Job? = null
+    private var stopped = false
+
+    @Volatile private var muted: Boolean = false
+
+    /**
+     * Start capturing + encoding + publishing in the background. Returns
+     * immediately. Calling twice is an error.
+     */
+    fun start(onError: (AudioException) -> Unit = { /* swallow */ }) {
+        check(!stopped) { "AudioRoomBroadcaster already stopped" }
+        check(job == null) { "AudioRoomBroadcaster.start already called" }
+
+        capture.start()
+        job =
+            scope.launch {
+                try {
+                    while (true) {
+                        val pcm = capture.readFrame() ?: break
+                        val opus =
+                            try {
+                                encoder.encode(pcm)
+                            } catch (ce: CancellationException) {
+                                throw ce
+                            } catch (t: Throwable) {
+                                onError(
+                                    AudioException(
+                                        AudioException.Kind.DecoderError,
+                                        "Opus encode failed for a frame",
+                                        t,
+                                    ),
+                                )
+                                continue
+                            }
+                        if (opus.isEmpty()) continue
+                        if (muted) continue
+                        runCatching { publisher.send(opus) }
+                            .onFailure { t ->
+                                if (t is CancellationException) throw t
+                                // Network drop on send is recoverable — log via onError but
+                                // don't stop the loop; the next frame may go through.
+                                onError(
+                                    AudioException(
+                                        AudioException.Kind.PlaybackFailed,
+                                        "publisher.send failed",
+                                        t,
+                                    ),
+                                )
+                            }
+                    }
+                } catch (ce: CancellationException) {
+                    throw ce
+                } catch (t: Throwable) {
+                    onError(
+                        AudioException(
+                            AudioException.Kind.PlaybackFailed,
+                            "audio capture pipeline failed",
+                            t,
+                        ),
+                    )
+                }
+            }
+    }
+
+    /**
+     * Toggle whether captured frames reach the wire. The mic stays open and
+     * the encoder keeps state consistent so unmute resumes mid-conversation
+     * with no glitch. No-op once [stop] has been called.
+     */
+    fun setMuted(muted: Boolean) {
+        if (stopped) return
+        this.muted = muted
+    }
+
+    /**
+     * Stop the loop, release the mic, release the encoder, close the MoQ
+     * publisher (which fires SUBSCRIBE_DONE to every attached subscriber).
+     * Idempotent.
+     */
+    suspend fun stop() {
+        if (stopped) return
+        stopped = true
+        job?.cancel()
+        runCatching { capture.stop() }
+        runCatching { encoder.release() }
+        runCatching { publisher.close() }
+    }
+}
@@ -0,0 +1,170 @@
+/*
+ * 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
+
+import com.vitorpamplona.nestsclient.audio.AudioCapture
+import com.vitorpamplona.nestsclient.audio.OpusEncoder
+import com.vitorpamplona.nestsclient.moq.Announce
+import com.vitorpamplona.nestsclient.moq.AnnounceOk
+import com.vitorpamplona.nestsclient.moq.ClientSetup
+import com.vitorpamplona.nestsclient.moq.MoqCodec
+import com.vitorpamplona.nestsclient.moq.MoqSession
+import com.vitorpamplona.nestsclient.moq.MoqVersion
+import com.vitorpamplona.nestsclient.moq.ServerSetup
+import com.vitorpamplona.nestsclient.moq.TrackNamespace
+import com.vitorpamplona.nestsclient.transport.FakeWebTransport
+import kotlinx.coroutines.async
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertIs
+import kotlin.test.assertTrue
+
+class NestsSpeakerTest {
+    @Test
+    fun startBroadcasting_announces_then_state_is_Broadcasting() =
+        runTest {
+            val (speakerSide, peerSide) = FakeWebTransport.pair()
+            val speakerSession = MoqSession.client(speakerSide, backgroundScope)
+            val ns = TrackNamespace.of("nests", "test-room")
+
+            val peer =
+                async {
+                    val ctrl = peerSide.peerOpenedBidiStreams().first()
+                    val cs = MoqCodec.decode(ctrl.incoming().first())!!.message as ClientSetup
+                    ctrl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first())))
+                    val ann = MoqCodec.decode(ctrl.incoming().first())!!.message as Announce
+                    ctrl.write(MoqCodec.encode(AnnounceOk(ann.namespace)))
+                }
+
+            speakerSession.setup(listOf(MoqVersion.DRAFT_17))
+
+            val state =
+                MutableStateFlow<NestsSpeakerState>(
+                    NestsSpeakerState.Connected(
+                        roomInfo = NestsRoomInfo(endpoint = "https://relay.example/moq"),
+                        negotiatedMoqVersion = speakerSession.selectedVersion!!,
+                    ),
+                )
+            val speaker =
+                DefaultNestsSpeaker(
+                    session = speakerSession,
+                    roomNamespace = ns,
+                    speakerTrackName = "test-pubkey".encodeToByteArray(),
+                    captureFactory = { ConstantCapture() },
+                    encoderFactory = { ConstantEncoder() },
+                    scope = backgroundScope,
+                    mutableState = state,
+                )
+
+            val handle = speaker.startBroadcasting()
+            peer.await()
+
+            val now = speaker.state.value
+            assertIs<NestsSpeakerState.Broadcasting>(now)
+            assertEquals(false, now.isMuted)
+
+            // setMuted reflects through into the public state.
+            handle.setMuted(true)
+            assertEquals(true, (speaker.state.value as NestsSpeakerState.Broadcasting).isMuted)
+            assertEquals(true, handle.isMuted)
+
+            handle.close()
+            // Should fall back to Connected after broadcast ends.
+            assertIs<NestsSpeakerState.Connected>(speaker.state.value)
+
+            speaker.close()
+            assertEquals(NestsSpeakerState.Closed, speaker.state.value)
+        }
+
+    @Test
+    fun startBroadcasting_twice_throws_IllegalStateException() =
+        runTest {
+            val (speakerSide, peerSide) = FakeWebTransport.pair()
+            val speakerSession = MoqSession.client(speakerSide, backgroundScope)
+            val ns = TrackNamespace.of("nests", "test-room")
+
+            val peer =
+                async {
+                    val ctrl = peerSide.peerOpenedBidiStreams().first()
+                    val cs = MoqCodec.decode(ctrl.incoming().first())!!.message as ClientSetup
+                    ctrl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first())))
+                    val ann = MoqCodec.decode(ctrl.incoming().first())!!.message as Announce
+                    ctrl.write(MoqCodec.encode(AnnounceOk(ann.namespace)))
+                }
+
+            speakerSession.setup(listOf(MoqVersion.DRAFT_17))
+            val state =
+                MutableStateFlow<NestsSpeakerState>(
+                    NestsSpeakerState.Connected(
+                        roomInfo = NestsRoomInfo(endpoint = "https://relay.example/moq"),
+                        negotiatedMoqVersion = speakerSession.selectedVersion!!,
+                    ),
+                )
+            val speaker =
+                DefaultNestsSpeaker(
+                    session = speakerSession,
+                    roomNamespace = ns,
+                    speakerTrackName = "test-pubkey".encodeToByteArray(),
+                    captureFactory = { ConstantCapture() },
+                    encoderFactory = { ConstantEncoder() },
+                    scope = backgroundScope,
+                    mutableState = state,
+                )
+
+            speaker.startBroadcasting()
+            peer.await()
+
+            val ex =
+                runCatching { speaker.startBroadcasting() }.exceptionOrNull()
+            assertTrue(ex is IllegalStateException, "second startBroadcasting should throw IllegalStateException, got $ex")
+
+            speaker.close()
+        }
+
+    // ----- Fakes -----
+
+    private class ConstantCapture : AudioCapture {
+        // Yields a single dummy frame then drains; per-instance so tests
+        // don't share state through a singleton.
+        private val ch = Channel<ShortArray>(capacity = 1)
+
+        init {
+            ch.trySend(ShortArray(960) { 0 })
+            ch.close()
+        }
+
+        override fun start() {}
+
+        override suspend fun readFrame(): ShortArray? = ch.receiveCatching().getOrNull()
+
+        override fun stop() {}
+    }
+
+    private class ConstantEncoder : OpusEncoder {
+        override fun encode(pcm: ShortArray): ByteArray = byteArrayOf(0xAA.toByte(), 0xBB.toByte())
+
+        override fun release() {}
+    }
+}
@@ -0,0 +1,210 @@
+/*
+ * 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
+
+import com.vitorpamplona.nestsclient.moq.MoqSession
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertContentEquals
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+class AudioRoomBroadcasterTest {
+    @Test
+    fun pcm_frames_flow_through_encoder_into_publisher_in_order() =
+        runTest {
+            val capture = ScriptedCapture(listOf(shortArrayOf(1), shortArrayOf(2), shortArrayOf(3)))
+            val encoder = ScriptedEncoder(prefix = byteArrayOf(0xAA.toByte()))
+            val publisher = RecordingPublisher("track")
+            val broadcaster = AudioRoomBroadcaster(capture, encoder, publisher, backgroundScope)
+
+            broadcaster.start()
+            // Capture exhausts after 3 frames (readFrame returns null) so the
+            // loop terminates on its own; broadcaster.stop() cleans up the
+            // resources but the wire results are observable now.
+            capture.awaitDrained()
+
+            assertEquals(
+                listOf(byteArrayOf(0xAA.toByte(), 1).toList(), byteArrayOf(0xAA.toByte(), 2).toList(), byteArrayOf(0xAA.toByte(), 3).toList()),
+                publisher.sent.map { it.toList() },
+            )
+
+            broadcaster.stop()
+            assertTrue(capture.startedFlag, "capture should have started")
+            assertTrue(capture.stopCount > 0, "capture should have stopped")
+            assertTrue(encoder.released, "encoder should have been released")
+            assertTrue(publisher.closed, "publisher should have been closed")
+        }
+
+    @Test
+    fun setMuted_true_keeps_capture_running_but_no_frames_published() =
+        runTest {
+            val capture = ScriptedCapture(listOf(shortArrayOf(1), shortArrayOf(2), shortArrayOf(3)))
+            val encoder = ScriptedEncoder(prefix = byteArrayOf(0xBB.toByte()))
+            val publisher = RecordingPublisher("track")
+            val broadcaster = AudioRoomBroadcaster(capture, encoder, publisher, backgroundScope)
+
+            broadcaster.setMuted(true)
+            broadcaster.start()
+            capture.awaitDrained()
+
+            // Mic was open + encoder ran; publisher saw nothing.
+            assertTrue(capture.startedFlag)
+            assertEquals(3, encoder.encodeCalls, "encoder still ran while muted (state preserved)")
+            assertTrue(publisher.sent.isEmpty(), "no frames should reach the wire while muted")
+
+            broadcaster.stop()
+        }
+
+    @Test
+    fun encoder_returning_empty_array_is_silently_skipped() =
+        runTest {
+            val capture = ScriptedCapture(listOf(shortArrayOf(1), shortArrayOf(2)))
+            val encoder = ScriptedEncoder(prefix = byteArrayOf(0xCC.toByte()), warmupSkips = 1)
+            val publisher = RecordingPublisher("track")
+            val broadcaster = AudioRoomBroadcaster(capture, encoder, publisher, backgroundScope)
+
+            broadcaster.start()
+            capture.awaitDrained()
+
+            // First frame was the warmup → encoder returned empty → skipped.
+            // Second frame should reach the publisher.
+            assertEquals(1, publisher.sent.size)
+            assertContentEquals(byteArrayOf(0xCC.toByte(), 2), publisher.sent.single())
+
+            broadcaster.stop()
+        }
+
+    @Test
+    fun encoder_failure_does_not_stop_the_loop() =
+        runTest {
+            val capture = ScriptedCapture(listOf(shortArrayOf(1), shortArrayOf(2), shortArrayOf(3)))
+            val encoder = ScriptedEncoder(prefix = byteArrayOf(0xDD.toByte()), failOnNthCall = 1)
+            val publisher = RecordingPublisher("track")
+            val broadcaster = AudioRoomBroadcaster(capture, encoder, publisher, backgroundScope)
+            val errors = mutableListOf<AudioException>()
+
+            broadcaster.start { errors.add(it) }
+            capture.awaitDrained()
+
+            assertEquals(1, errors.size, "exactly one encode error should have been reported")
+            assertEquals(2, publisher.sent.size, "two non-failed frames should have reached the publisher")
+            broadcaster.stop()
+        }
+
+    @Test
+    fun stop_is_idempotent() =
+        runTest {
+            val capture = ScriptedCapture(listOf(shortArrayOf(1)))
+            val encoder = ScriptedEncoder(prefix = byteArrayOf(0xEE.toByte()))
+            val publisher = RecordingPublisher("track")
+            val broadcaster = AudioRoomBroadcaster(capture, encoder, publisher, backgroundScope)
+
+            broadcaster.start()
+            capture.awaitDrained()
+            broadcaster.stop()
+            broadcaster.stop()
+            broadcaster.stop()
+
+            assertEquals(1, capture.stopCount)
+            assertTrue(encoder.released)
+            assertTrue(publisher.closed)
+        }
+
+    // ---------- Fakes ----------
+
+    private class ScriptedCapture(
+        frames: List<ShortArray>,
+    ) : AudioCapture {
+        private val queue = Channel<ShortArray>(capacity = frames.size + 1)
+        private val drainSignal = kotlinx.coroutines.CompletableDeferred<Unit>()
+        var startedFlag = false
+            private set
+        var stopCount = 0
+            private set
+
+        init {
+            frames.forEach { queue.trySend(it) }
+            queue.close()
+        }
+
+        override fun start() {
+            startedFlag = true
+        }
+
+        override suspend fun readFrame(): ShortArray? {
+            val next = queue.receiveCatching().getOrNull()
+            if (next == null && !drainSignal.isCompleted) drainSignal.complete(Unit)
+            return next
+        }
+
+        override fun stop() {
+            stopCount++
+        }
+
+        suspend fun awaitDrained() {
+            drainSignal.await()
+        }
+    }
+
+    private class ScriptedEncoder(
+        private val prefix: ByteArray,
+        private val warmupSkips: Int = 0,
+        private val failOnNthCall: Int = -1, // -1 = never fail
+    ) : OpusEncoder {
+        var encodeCalls: Int = 0
+            private set
+        var released: Boolean = false
+            private set
+
+        override fun encode(pcm: ShortArray): ByteArray {
+            val callIndex = encodeCalls
+            encodeCalls += 1
+            if (callIndex == failOnNthCall) error("scripted encoder failure at call $callIndex")
+            if (callIndex < warmupSkips) return ByteArray(0)
+            // Append the first PCM sample byte so tests can verify ordering.
+            return prefix + byteArrayOf(pcm.first().toByte())
+        }
+
+        override fun release() {
+            released = true
+        }
+    }
+
+    private class RecordingPublisher(
+        nameStr: String,
+    ) : MoqSession.TrackPublisher {
+        override val name: ByteArray = nameStr.encodeToByteArray()
+        val sent: MutableList<ByteArray> = mutableListOf()
+        var closed: Boolean = false
+            private set
+
+        override suspend fun send(payload: ByteArray): Boolean {
+            sent.add(payload)
+            return true
+        }
+
+        override suspend fun close() {
+            closed = true
+        }
+    }
+}