feat(nestsClient): MoQ ANNOUNCE family + Opus encoder + AudioRecord capture

Phase 4 (codec + audio capture slice). Adds the publisher-side MoQ control messages and the Opus encode + microphone capture pieces a speaker needs. The host-grants-speaker UI flow is deferred — that's multi-screen UX that should be designed before being implemented. commonMain (nestsclient.moq): - 5 new MoqMessageType entries with codec round-trip + tests: * Announce (0x06): publisher offers a track namespace. * AnnounceOk (0x07): subscriber acknowledges. * AnnounceError (0x08): subscriber rejects with error code + reason phrase. * Unannounce (0x09): publisher withdraws a previously-announced namespace. * SubscribeDone (0x0B): publisher tells the subscriber no more objects are coming for this subscription, with stream count and reason. commonMain (nestsclient.audio): - New `OpusEncoder` interface — symmetric to `OpusDecoder`, one instance per outgoing track since Opus state is per-stream. - New `AudioCapture` interface — `start()`, `readFrame()` returns a PCM frame or null when stopped, `stop()` releases the mic. androidMain (nestsclient.audio): - `MediaCodecOpusEncoder` — wraps `MediaCodec("audio/opus")` encoder variant (API 29+). 48 kHz mono in, 32 kbit/s VBR Opus out, 20 ms frames. Drains output queue per encode call. - `AudioRecordCapture` — wraps `AudioRecord` from `MediaRecorder.AudioSource.VOICE_COMMUNICATION` so the platform's echo-cancellation + noise-suppression filters apply when available. Reads exactly one PCM frame per readFrame() call, retries on underrun, throws AudioException(DeviceUnavailable) on permission/ resource failures. commonTest: - `AnnounceCodecTest` — 7 cases covering each message round-trip, concatenated decode in sequence, and a guard against accidentally reordering MoqMessageType enum codes. Permission already declared: - `RECORD_AUDIO` is already in amethyst/AndroidManifest.xml — no manifest change needed. What this does NOT include: - A publisher loop class (analogous to AudioRoomPlayer for publish direction) — can be added when the speak-button wiring lands. - The host-grants-speaker UI — needs design input. - STREAM_HEADER_SUBGROUP for stream-based object delivery — datagrams cover the listener happy-path; streams add reliability that nests may or may not require. https://claude.ai/code/session_013nVLALALKaHVgHm9u5Cg8D
2026-04-22 07:33:55 +00:00
parent 64b3367472
commit b62e3dd0ec
6 changed files with 535 additions and 0 deletions
@@ -0,0 +1,136 @@
+/*
+ * 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 android.media.AudioRecord
+import android.media.MediaRecorder
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+import android.media.AudioFormat as AndroidAudioFormat
+
+/**
+ * [AudioCapture] backed by Android's [AudioRecord] from the
+ * VOICE_COMMUNICATION input source — same source LiveKit, WebRTC, and most
+ * voice-chat libraries use, so it gets the platform's echo-cancellation and
+ * noise-suppression filters when available.
+ *
+ * **Permission:** the caller is responsible for holding `RECORD_AUDIO` before
+ * calling [start]; this class will throw [AudioException.Kind.DeviceUnavailable]
+ * if the OS denies the resource.
+ */
+class AudioRecordCapture(
+    private val source: Int = MediaRecorder.AudioSource.VOICE_COMMUNICATION,
+) : AudioCapture {
+    private var record: AudioRecord? = null
+    private var stopped = false
+
+    override fun start() {
+        check(!stopped) { "capture already stopped" }
+        if (record != null) return
+
+        val channelMask =
+            when (AudioFormat.CHANNELS) {
+                1 -> AndroidAudioFormat.CHANNEL_IN_MONO
+                2 -> AndroidAudioFormat.CHANNEL_IN_STEREO
+                else -> error("unsupported channel count ${AudioFormat.CHANNELS}")
+            }
+
+        val minBuffer =
+            AudioRecord.getMinBufferSize(
+                AudioFormat.SAMPLE_RATE_HZ,
+                channelMask,
+                AndroidAudioFormat.ENCODING_PCM_16BIT,
+            )
+        if (minBuffer <= 0) {
+            throw AudioException(
+                AudioException.Kind.DeviceUnavailable,
+                "AudioRecord.getMinBufferSize returned $minBuffer for ${AudioFormat.SAMPLE_RATE_HZ} Hz",
+            )
+        }
+        val bufferBytes =
+            maxOf(minBuffer, AudioFormat.FRAME_SIZE_SAMPLES * AudioFormat.BYTES_PER_SAMPLE * 4)
+
+        val rec =
+            try {
+                @Suppress("MissingPermission")
+                AudioRecord(source, AudioFormat.SAMPLE_RATE_HZ, channelMask, AndroidAudioFormat.ENCODING_PCM_16BIT, bufferBytes)
+            } catch (t: Throwable) {
+                throw AudioException(
+                    AudioException.Kind.DeviceUnavailable,
+                    "Failed to construct AudioRecord (RECORD_AUDIO permission?)",
+                    t,
+                )
+            }
+        if (rec.state != AudioRecord.STATE_INITIALIZED) {
+            runCatching { rec.release() }
+            throw AudioException(
+                AudioException.Kind.DeviceUnavailable,
+                "AudioRecord state=${rec.state} after construction (expected INITIALIZED)",
+            )
+        }
+        try {
+            rec.startRecording()
+        } catch (t: Throwable) {
+            runCatching { rec.release() }
+            throw AudioException(
+                AudioException.Kind.DeviceUnavailable,
+                "AudioRecord.startRecording() failed",
+                t,
+            )
+        }
+        record = rec
+    }
+
+    override suspend fun readFrame(): ShortArray? {
+        val rec = record ?: return null
+        val frame = ShortArray(AudioFormat.FRAME_SIZE_SAMPLES)
+        return withContext(Dispatchers.IO) {
+            var read = 0
+            while (read < frame.size) {
+                if (stopped) return@withContext null
+                val n = rec.read(frame, read, frame.size - read)
+                if (n < 0) {
+                    throw AudioException(
+                        AudioException.Kind.PlaybackFailed,
+                        "AudioRecord.read returned error code $n",
+                    )
+                }
+                if (n == 0) {
+                    // Underrun — wait briefly and retry. This avoids busy-waiting
+                    // on a slow producer.
+                    kotlinx.coroutines.delay(2)
+                    continue
+                }
+                read += n
+            }
+            frame
+        }
+    }
+
+    override fun stop() {
+        if (stopped) return
+        stopped = true
+        val rec = record ?: return
+        record = null
+        runCatching { rec.stop() }
+        runCatching { rec.release() }
+    }
+}
@@ -0,0 +1,139 @@
+/*
+ * 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 android.media.MediaCodec
+import android.media.MediaCodecInfo
+import android.media.MediaFormat
+import java.nio.ByteOrder
+
+/**
+ * [OpusEncoder] backed by Android's [MediaCodec] (`audio/opus`, encoder
+ * variant available on API 29+ — the decoder works back to API 21 but the
+ * encoder shipped later). One instance per outgoing track.
+ *
+ * Configuration:
+ *   - 48 kHz mono input PCM 16-bit (matches [AudioFormat]).
+ *   - Target bitrate ~32 kbit/s VBR — high-quality wideband speech.
+ *   - 20 ms frames (the encoder requires the input buffer to hold one frame
+ *     at a time for low latency).
+ */
+class MediaCodecOpusEncoder(
+    private val targetBitrate: Int = DEFAULT_BITRATE_BPS,
+) : OpusEncoder {
+    private val codec: MediaCodec =
+        try {
+            MediaCodec.createEncoderByType(MediaFormat.MIMETYPE_AUDIO_OPUS).apply {
+                configure(buildFormat(targetBitrate), null, null, MediaCodec.CONFIGURE_FLAG_ENCODE)
+                start()
+            }
+        } catch (t: Throwable) {
+            throw AudioException(
+                AudioException.Kind.DeviceUnavailable,
+                "Failed to allocate MediaCodec audio/opus encoder",
+                t,
+            )
+        }
+
+    private val bufferInfo = MediaCodec.BufferInfo()
+    private var presentationTimeUs: Long = 0L
+    private var released = false
+
+    override fun encode(pcm: ShortArray): ByteArray {
+        check(!released) { "encoder released" }
+        require(pcm.isNotEmpty()) { "PCM frame must not be empty" }
+
+        val inputIndex = codec.dequeueInputBuffer(DEQUEUE_TIMEOUT_US)
+        if (inputIndex < 0) return ByteArray(0)
+        val inputBuffer =
+            codec.getInputBuffer(inputIndex)
+                ?: throw AudioException(
+                    AudioException.Kind.DecoderError,
+                    "MediaCodec returned null input buffer at index $inputIndex",
+                )
+        inputBuffer.clear()
+        inputBuffer.order(ByteOrder.nativeOrder()).asShortBuffer().put(pcm)
+        val byteCount = pcm.size * AudioFormat.BYTES_PER_SAMPLE
+        codec.queueInputBuffer(inputIndex, 0, byteCount, presentationTimeUs, 0)
+        presentationTimeUs += FRAME_DURATION_US
+
+        // One PCM frame produces one Opus packet (sometimes after one warmup
+        // round). Drain the output queue once.
+        while (true) {
+            val outputIndex = codec.dequeueOutputBuffer(bufferInfo, DEQUEUE_TIMEOUT_US)
+            when {
+                outputIndex >= 0 -> {
+                    val outputBuffer = codec.getOutputBuffer(outputIndex) ?: continue
+                    val opus = ByteArray(bufferInfo.size)
+                    outputBuffer.position(bufferInfo.offset)
+                    outputBuffer.limit(bufferInfo.offset + bufferInfo.size)
+                    outputBuffer.get(opus)
+                    codec.releaseOutputBuffer(outputIndex, false)
+                    if (opus.isNotEmpty()) return opus
+                }
+
+                outputIndex == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED -> {
+                    continue
+                }
+
+                outputIndex == MediaCodec.INFO_TRY_AGAIN_LATER -> {
+                    return ByteArray(0)
+                }
+
+                else -> {
+                    return ByteArray(0)
+                }
+            }
+        }
+        @Suppress("UNREACHABLE_CODE")
+        return ByteArray(0)
+    }
+
+    override fun release() {
+        if (released) return
+        released = true
+        runCatching { codec.stop() }
+        runCatching { codec.release() }
+    }
+
+    companion object {
+        const val DEFAULT_BITRATE_BPS: Int = 32_000
+        private const val DEQUEUE_TIMEOUT_US = 10_000L
+        private const val FRAME_DURATION_US = 20_000L
+
+        private fun buildFormat(bitrate: Int): MediaFormat =
+            MediaFormat
+                .createAudioFormat(
+                    MediaFormat.MIMETYPE_AUDIO_OPUS,
+                    AudioFormat.SAMPLE_RATE_HZ,
+                    AudioFormat.CHANNELS,
+                ).apply {
+                    setInteger(MediaFormat.KEY_BIT_RATE, bitrate)
+                    setInteger(MediaFormat.KEY_PCM_ENCODING, android.media.AudioFormat.ENCODING_PCM_16BIT)
+                    // Encoder-side AAC/Opus profile selection: SignalingDelaySamples
+                    // is implicit; nothing else required for mono speech.
+                    setInteger(
+                        MediaFormat.KEY_AAC_PROFILE,
+                        MediaCodecInfo.CodecProfileLevel.AACObjectLC,
+                    )
+                }
+    }
+}
@@ -60,6 +60,42 @@ interface OpusDecoder {
    fun release()
 }

+/**
+ * Encoder for one PCM frame at a time.
+ *
+ * Mirror of [OpusDecoder] for the publish direction. Like the decoder, Opus
+ * encoder state is per-stream — one instance per outgoing track.
+ */
+interface OpusEncoder {
+    /**
+     * Encode one PCM frame (typically [AudioFormat.FRAME_SIZE_SAMPLES] samples
+     * of signed 16-bit mono at [AudioFormat.SAMPLE_RATE_HZ]) into one Opus
+     * packet. Returns an empty array if the encoder is still warming up
+     * (some pipelines need a few frames before producing output).
+     */
+    fun encode(pcm: ShortArray): ByteArray
+
+    fun release()
+}
+
+/**
+ * Source for PCM audio capture. Implementations open the device's microphone
+ * and produce one frame at a time via [readFrame].
+ */
+interface AudioCapture {
+    /** Allocate the microphone resource and begin capturing. */
+    fun start()
+
+    /**
+     * Read one PCM frame ([AudioFormat.FRAME_SIZE_SAMPLES] samples). Suspends
+     * until enough samples are available. Returns null when [stop] is called.
+     */
+    suspend fun readFrame(): ShortArray?
+
+    /** Stop capture and release the microphone. After this, [readFrame] returns null. */
+    fun stop()
+}
+
 /**
 * Sink for PCM audio playback. Implementations buffer internally — [enqueue]
 * may suspend if the device's playback buffer is full.
@@ -73,6 +73,11 @@ object MoqCodec {
                MoqMessageType.SubscribeOk -> decodeSubscribeOk(reader)
                MoqMessageType.SubscribeError -> decodeSubscribeError(reader)
                MoqMessageType.Unsubscribe -> decodeUnsubscribe(reader)
+                MoqMessageType.SubscribeDone -> decodeSubscribeDone(reader)
+                MoqMessageType.Announce -> decodeAnnounce(reader)
+                MoqMessageType.AnnounceOk -> decodeAnnounceOk(reader)
+                MoqMessageType.AnnounceError -> decodeAnnounceError(reader)
+                MoqMessageType.Unannounce -> decodeUnannounce(reader)
            }
        if (reader.hasMore()) {
            throw MoqCodecException(
@@ -90,6 +95,11 @@ object MoqCodec {
            is SubscribeOk -> encodeSubscribeOk(message)
            is SubscribeError -> encodeSubscribeError(message)
            is Unsubscribe -> encodeUnsubscribe(message)
+            is SubscribeDone -> encodeSubscribeDone(message)
+            is Announce -> encodeAnnounce(message)
+            is AnnounceOk -> encodeAnnounceOk(message)
+            is AnnounceError -> encodeAnnounceError(message)
+            is Unannounce -> encodeUnannounce(message)
        }

    private fun encodeClientSetup(message: ClientSetup): ByteArray {
@@ -268,6 +278,63 @@ object MoqCodec {

    private fun decodeUnsubscribe(r: MoqReader): Unsubscribe = Unsubscribe(r.readVarint())

+    private fun encodeSubscribeDone(m: SubscribeDone): ByteArray {
+        val w = MoqWriter()
+        w.writeVarint(m.subscribeId)
+        w.writeVarint(m.statusCode)
+        w.writeVarint(m.streamCount)
+        w.writeLengthPrefixedString(m.reasonPhrase)
+        return w.toByteArray()
+    }
+
+    private fun decodeSubscribeDone(r: MoqReader): SubscribeDone =
+        SubscribeDone(
+            subscribeId = r.readVarint(),
+            statusCode = r.readVarint(),
+            streamCount = r.readVarint(),
+            reasonPhrase = r.readLengthPrefixedString(),
+        )
+
+    private fun encodeAnnounce(m: Announce): ByteArray {
+        val w = MoqWriter()
+        encodeNamespace(w, m.namespace)
+        encodeParameters(w, m.parameters)
+        return w.toByteArray()
+    }
+
+    private fun decodeAnnounce(r: MoqReader): Announce = Announce(namespace = decodeNamespace(r), parameters = decodeParameters(r))
+
+    private fun encodeAnnounceOk(m: AnnounceOk): ByteArray {
+        val w = MoqWriter()
+        encodeNamespace(w, m.namespace)
+        return w.toByteArray()
+    }
+
+    private fun decodeAnnounceOk(r: MoqReader): AnnounceOk = AnnounceOk(decodeNamespace(r))
+
+    private fun encodeAnnounceError(m: AnnounceError): ByteArray {
+        val w = MoqWriter()
+        encodeNamespace(w, m.namespace)
+        w.writeVarint(m.errorCode)
+        w.writeLengthPrefixedString(m.reasonPhrase)
+        return w.toByteArray()
+    }
+
+    private fun decodeAnnounceError(r: MoqReader): AnnounceError =
+        AnnounceError(
+            namespace = decodeNamespace(r),
+            errorCode = r.readVarint(),
+            reasonPhrase = r.readLengthPrefixedString(),
+        )
+
+    private fun encodeUnannounce(m: Unannounce): ByteArray {
+        val w = MoqWriter()
+        encodeNamespace(w, m.namespace)
+        return w.toByteArray()
+    }
+
+    private fun decodeUnannounce(r: MoqReader): Unannounce = Unannounce(decodeNamespace(r))
+
    data class DecodeResult(
        val message: MoqMessage,
        val bytesConsumed: Int,
@@ -44,7 +44,12 @@ enum class MoqMessageType(
    Subscribe(0x03),
    SubscribeOk(0x04),
    SubscribeError(0x05),
+    Announce(0x06),
+    AnnounceOk(0x07),
+    AnnounceError(0x08),
+    Unannounce(0x09),
    Unsubscribe(0x0A),
+    SubscribeDone(0x0B),
    ClientSetup(0x40),
    ServerSetup(0x41),
    ;
@@ -254,3 +259,53 @@ data class Unsubscribe(
 ) : MoqMessage() {
    override val type: MoqMessageType = MoqMessageType.Unsubscribe
 }
+
+/**
+ * SUBSCRIBE_DONE (0x0B): publisher tells the subscriber that no more objects
+ * are coming for this subscription, optionally indicating the last group/object
+ * boundary. Sent on subscription expiry, publisher-side track closure, or
+ * after an UNSUBSCRIBE was acknowledged.
+ */
+data class SubscribeDone(
+    val subscribeId: Long,
+    val statusCode: Long,
+    val streamCount: Long,
+    val reasonPhrase: String,
+) : MoqMessage() {
+    override val type: MoqMessageType = MoqMessageType.SubscribeDone
+}
+
+/**
+ * ANNOUNCE (0x06): publisher offers a track namespace. nests publishers send
+ * one ANNOUNCE per audio-room they host so subscribers know which namespace
+ * to subscribe under.
+ */
+data class Announce(
+    val namespace: TrackNamespace,
+    val parameters: List<SetupParameter> = emptyList(),
+) : MoqMessage() {
+    override val type: MoqMessageType = MoqMessageType.Announce
+}
+
+/** ANNOUNCE_OK (0x07): subscriber acknowledges an ANNOUNCE. */
+data class AnnounceOk(
+    val namespace: TrackNamespace,
+) : MoqMessage() {
+    override val type: MoqMessageType = MoqMessageType.AnnounceOk
+}
+
+/** ANNOUNCE_ERROR (0x08): subscriber rejects an ANNOUNCE. */
+data class AnnounceError(
+    val namespace: TrackNamespace,
+    val errorCode: Long,
+    val reasonPhrase: String,
+) : MoqMessage() {
+    override val type: MoqMessageType = MoqMessageType.AnnounceError
+}
+
+/** UNANNOUNCE (0x09): publisher withdraws a previously-announced namespace. */
+data class Unannounce(
+    val namespace: TrackNamespace,
+) : MoqMessage() {
+    override val type: MoqMessageType = MoqMessageType.Unannounce
+}
@@ -0,0 +1,102 @@
+/*
+ * 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.moq
+
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertIs
+
+class AnnounceCodecTest {
+    @Test
+    fun announce_round_trip_with_parameters() {
+        val msg =
+            Announce(
+                namespace = TrackNamespace.of("nests", "test-room"),
+                parameters = listOf(SetupParameter(0x10L, byteArrayOf(0x01, 0x02))),
+            )
+        val decoded = MoqCodec.decode(MoqCodec.encode(msg))!!.message
+        assertEquals(msg, assertIs<Announce>(decoded))
+    }
+
+    @Test
+    fun announce_ok_round_trip() {
+        val msg = AnnounceOk(namespace = TrackNamespace.of("ns"))
+        val decoded = MoqCodec.decode(MoqCodec.encode(msg))!!.message
+        assertEquals(msg, assertIs<AnnounceOk>(decoded))
+    }
+
+    @Test
+    fun announce_error_round_trip() {
+        val msg =
+            AnnounceError(
+                namespace = TrackNamespace.of("nests", "denied-room"),
+                errorCode = 403,
+                reasonPhrase = "namespace already taken",
+            )
+        val decoded = MoqCodec.decode(MoqCodec.encode(msg))!!.message
+        assertEquals(msg, assertIs<AnnounceError>(decoded))
+    }
+
+    @Test
+    fun unannounce_round_trip() {
+        val msg = Unannounce(namespace = TrackNamespace.of("nests", "test-room"))
+        val decoded = MoqCodec.decode(MoqCodec.encode(msg))!!.message
+        assertEquals(msg, assertIs<Unannounce>(decoded))
+    }
+
+    @Test
+    fun subscribe_done_round_trip() {
+        val msg =
+            SubscribeDone(
+                subscribeId = 42,
+                statusCode = 0,
+                streamCount = 17,
+                reasonPhrase = "subscription expired",
+            )
+        val decoded = MoqCodec.decode(MoqCodec.encode(msg))!!.message
+        assertEquals(msg, assertIs<SubscribeDone>(decoded))
+    }
+
+    @Test
+    fun announce_then_announce_ok_decode_in_sequence() {
+        val ns = TrackNamespace.of("nests", "abc")
+        val ann = Announce(ns)
+        val ok = AnnounceOk(ns)
+        val bytes = MoqCodec.encode(ann) + MoqCodec.encode(ok)
+
+        val first = MoqCodec.decode(bytes, offset = 0)!!
+        val second = MoqCodec.decode(bytes, offset = first.bytesConsumed)!!
+
+        assertIs<Announce>(first.message)
+        assertIs<AnnounceOk>(second.message)
+        assertEquals(bytes.size, first.bytesConsumed + second.bytesConsumed)
+    }
+
+    @Test
+    fun every_publisher_message_type_has_a_known_code() {
+        // Guards against accidentally reordering MoqMessageType enum entries.
+        assertEquals(0x06L, MoqMessageType.Announce.code)
+        assertEquals(0x07L, MoqMessageType.AnnounceOk.code)
+        assertEquals(0x08L, MoqMessageType.AnnounceError.code)
+        assertEquals(0x09L, MoqMessageType.Unannounce.code)
+        assertEquals(0x0BL, MoqMessageType.SubscribeDone.code)
+    }
+}