feat(nestsClient): MoQ SUBSCRIBE family + OBJECT_DATAGRAM codecs

Phase 3c-2 of the Clubhouse/nests integration. Extends the MoQ
control-plane codec with the subscribe lifecycle messages (SUBSCRIBE,
SUBSCRIBE_OK, SUBSCRIBE_ERROR, UNSUBSCRIBE) and adds the
OBJECT_DATAGRAM wire format the listener uses to receive low-latency
Opus audio. Pure codec layer — no session-pump or network changes;
the subscribe/object delivery Flow arrives in Phase 3c-3.

commonMain (nestsclient.moq):
- New message types in `MoqMessageType`: Subscribe (0x03),
  SubscribeOk (0x04), SubscribeError (0x05), Unsubscribe (0x0A).
- New data classes in MoqMessage.kt:
  * `TrackNamespace` — tuple of byte strings, with `of(vararg String)`
    helper for the common UTF-8 case.
  * `SubscribeFilter` enum. Phase 3c-2 supports LatestGroup /
    LatestObject; AbsoluteStart/AbsoluteRange throw at construction
    (they need extra wire fields we'll add when nests needs them).
  * `Subscribe`, `SubscribeOk`, `SubscribeError`, `Unsubscribe` data
    classes. `SubscribeOk` validates contentExists + largest-id
    coupling at construction.
- New codecs in `MoqCodec` (namespace tuple + all four messages).
  Unknown filter codes on the wire are rejected.
- `MoqObject` data class + `MoqObjectDatagram` encode/decode for the
  OBJECT_DATAGRAM wire format (track_alias, group_id, object_id,
  publisher_priority, status, payload).

Tests:
- `SubscribeCodecTest` — round-trips for each message with multi-
  segment namespaces + parameters, SUBSCRIBE_OK with and without
  contentExists, concatenated decode in sequence, unknown-filter
  rejection, construction-time validation.
- `MoqObjectDatagramTest` — Opus-sized payload round-trip, zero-
  length payload with OBJECT_DOES_NOT_EXIST status, 8-byte-varint
  boundary coverage, truncated datagram rejection, out-of-range
  priority rejection.

https://claude.ai/code/session_013nVLALALKaHVgHm9u5Cg8D

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt

@@ -69,6 +69,10 @@ object MoqCodec {
             when (type) {
                 MoqMessageType.ClientSetup -> decodeClientSetup(reader)
                 MoqMessageType.ServerSetup -> decodeServerSetup(reader)
+                MoqMessageType.Subscribe -> decodeSubscribe(reader)
+                MoqMessageType.SubscribeOk -> decodeSubscribeOk(reader)
+                MoqMessageType.SubscribeError -> decodeSubscribeError(reader)
+                MoqMessageType.Unsubscribe -> decodeUnsubscribe(reader)
             }
         if (reader.hasMore()) {
             throw MoqCodecException(
@@ -82,6 +86,10 @@ object MoqCodec {
         when (message) {
             is ClientSetup -> encodeClientSetup(message)
             is ServerSetup -> encodeServerSetup(message)
+            is Subscribe -> encodeSubscribe(message)
+            is SubscribeOk -> encodeSubscribeOk(message)
+            is SubscribeError -> encodeSubscribeError(message)
+            is Unsubscribe -> encodeUnsubscribe(message)
         }
 
     private fun encodeClientSetup(message: ClientSetup): ByteArray {
@@ -137,6 +145,129 @@ object MoqCodec {
         return out
     }
 
+    private fun encodeNamespace(
+        w: MoqWriter,
+        ns: TrackNamespace,
+    ) {
+        w.writeVarint(ns.tuple.size.toLong())
+        for (segment in ns.tuple) w.writeLengthPrefixedBytes(segment)
+    }
+
+    private fun decodeNamespace(r: MoqReader): TrackNamespace {
+        val count = r.readVarint().toInt()
+        if (count < 0 || count > 32) {
+            // draft-17 caps the tuple length well below this; any higher
+            // almost certainly means a corrupted/malicious frame.
+            throw MoqCodecException("absurd track namespace tuple length: $count")
+        }
+        val segments = ArrayList<ByteArray>(count)
+        repeat(count) { segments.add(r.readLengthPrefixedBytes()) }
+        return TrackNamespace(segments)
+    }
+
+    private fun encodeSubscribe(m: Subscribe): ByteArray {
+        val w = MoqWriter()
+        w.writeVarint(m.subscribeId)
+        w.writeVarint(m.trackAlias)
+        encodeNamespace(w, m.namespace)
+        w.writeLengthPrefixedBytes(m.trackName)
+        w.writeByte(m.subscriberPriority)
+        w.writeByte(m.groupOrder)
+        w.writeVarint(m.filter.code)
+        // LatestGroup / LatestObject have no extra fields.
+        encodeParameters(w, m.parameters)
+        return w.toByteArray()
+    }
+
+    private fun decodeSubscribe(r: MoqReader): Subscribe {
+        val subscribeId = r.readVarint()
+        val trackAlias = r.readVarint()
+        val namespace = decodeNamespace(r)
+        val trackName = r.readLengthPrefixedBytes()
+        val subscriberPriority = r.readByte()
+        val groupOrder = r.readByte()
+        val filterCode = r.readVarint()
+        val filter =
+            SubscribeFilter.fromCode(filterCode)
+                ?: throw MoqCodecException("unknown subscribe filter: 0x${filterCode.toString(16)}")
+        if (filter != SubscribeFilter.LatestGroup && filter != SubscribeFilter.LatestObject) {
+            throw MoqCodecException("subscribe filter $filter not yet supported")
+        }
+        val parameters = decodeParameters(r)
+        return Subscribe(
+            subscribeId = subscribeId,
+            trackAlias = trackAlias,
+            namespace = namespace,
+            trackName = trackName,
+            subscriberPriority = subscriberPriority,
+            groupOrder = groupOrder,
+            filter = filter,
+            parameters = parameters,
+        )
+    }
+
+    private fun encodeSubscribeOk(m: SubscribeOk): ByteArray {
+        val w = MoqWriter()
+        w.writeVarint(m.subscribeId)
+        w.writeVarint(m.expiresMs)
+        w.writeByte(m.groupOrder)
+        w.writeByte(if (m.contentExists) 1 else 0)
+        if (m.contentExists) {
+            w.writeVarint(m.largestGroupId!!)
+            w.writeVarint(m.largestObjectId!!)
+        }
+        encodeParameters(w, m.parameters)
+        return w.toByteArray()
+    }
+
+    private fun decodeSubscribeOk(r: MoqReader): SubscribeOk {
+        val subscribeId = r.readVarint()
+        val expiresMs = r.readVarint()
+        val groupOrder = r.readByte()
+        val contentExistsByte = r.readByte()
+        if (contentExistsByte != 0 && contentExistsByte != 1) {
+            throw MoqCodecException("content_exists must be 0 or 1, got $contentExistsByte")
+        }
+        val contentExists = contentExistsByte == 1
+        val largestGroupId = if (contentExists) r.readVarint() else null
+        val largestObjectId = if (contentExists) r.readVarint() else null
+        val parameters = decodeParameters(r)
+        return SubscribeOk(
+            subscribeId = subscribeId,
+            expiresMs = expiresMs,
+            groupOrder = groupOrder,
+            contentExists = contentExists,
+            largestGroupId = largestGroupId,
+            largestObjectId = largestObjectId,
+            parameters = parameters,
+        )
+    }
+
+    private fun encodeSubscribeError(m: SubscribeError): ByteArray {
+        val w = MoqWriter()
+        w.writeVarint(m.subscribeId)
+        w.writeVarint(m.errorCode)
+        w.writeLengthPrefixedString(m.reasonPhrase)
+        w.writeVarint(m.trackAlias)
+        return w.toByteArray()
+    }
+
+    private fun decodeSubscribeError(r: MoqReader): SubscribeError =
+        SubscribeError(
+            subscribeId = r.readVarint(),
+            errorCode = r.readVarint(),
+            reasonPhrase = r.readLengthPrefixedString(),
+            trackAlias = r.readVarint(),
+        )
+
+    private fun encodeUnsubscribe(m: Unsubscribe): ByteArray {
+        val w = MoqWriter()
+        w.writeVarint(m.subscribeId)
+        return w.toByteArray()
+    }
+
+    private fun decodeUnsubscribe(r: MoqReader): Unsubscribe = Unsubscribe(r.readVarint())
+
     data class DecodeResult(
         val message: MoqMessage,
         val bytesConsumed: Int,

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqMessage.kt

@@ -41,6 +41,10 @@ sealed class MoqMessage {
 enum class MoqMessageType(
     val code: Long,
 ) {
+    Subscribe(0x03),
+    SubscribeOk(0x04),
+    SubscribeError(0x05),
+    Unsubscribe(0x0A),
     ClientSetup(0x40),
     ServerSetup(0x41),
     ;
@@ -106,3 +110,147 @@ data class ServerSetup(
 ) : MoqMessage() {
     override val type: MoqMessageType = MoqMessageType.ServerSetup
 }
+
+/**
+ * MoQ track namespace is a tuple of byte strings (draft-ietf-moq-transport).
+ * Clients talking to nests typically use a 1- or 2-element namespace; we
+ * expose the tuple as a `List<ByteArray>` and provide a [text] helper for
+ * the common case where every element is UTF-8.
+ */
+data class TrackNamespace(
+    val tuple: List<ByteArray>,
+) {
+    fun text(): List<String> = tuple.map { it.decodeToString() }
+
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (other !is TrackNamespace) return false
+        if (tuple.size != other.tuple.size) return false
+        for (i in tuple.indices) if (!tuple[i].contentEquals(other.tuple[i])) return false
+        return true
+    }
+
+    override fun hashCode(): Int = tuple.fold(0) { acc, bytes -> 31 * acc + bytes.contentHashCode() }
+
+    companion object {
+        fun of(vararg segments: String): TrackNamespace = TrackNamespace(segments.map { it.encodeToByteArray() })
+    }
+}
+
+/**
+ * Subscribe filter types (draft-ietf-moq-transport). Values align with the
+ * draft's enum; we model it as an enum so the public API reads clearly.
+ */
+enum class SubscribeFilter(
+    val code: Long,
+) {
+    LatestGroup(0x01),
+    LatestObject(0x02),
+    AbsoluteStart(0x03),
+    AbsoluteRange(0x04),
+    ;
+
+    companion object {
+        private val byCode = entries.associateBy { it.code }
+
+        fun fromCode(code: Long): SubscribeFilter? = byCode[code]
+    }
+}
+
+/**
+ * SUBSCRIBE (0x03): client asks a publisher to forward objects belonging to a
+ * (namespace, track) pair. Phase 3c-2 supports only the LatestGroup /
+ * LatestObject filters — absolute-range variants add extra wire fields the
+ * codec will grow in a follow-up if nests ever needs them.
+ */
+data class Subscribe(
+    val subscribeId: Long,
+    val trackAlias: Long,
+    val namespace: TrackNamespace,
+    val trackName: ByteArray,
+    val subscriberPriority: Int = 0x80,
+    val groupOrder: Int = 0x00,
+    val filter: SubscribeFilter = SubscribeFilter.LatestGroup,
+    val parameters: List<SetupParameter> = emptyList(),
+) : MoqMessage() {
+    override val type: MoqMessageType = MoqMessageType.Subscribe
+
+    init {
+        require(filter == SubscribeFilter.LatestGroup || filter == SubscribeFilter.LatestObject) {
+            "Phase 3c-2 only supports LatestGroup / LatestObject filters, got $filter"
+        }
+        require(subscriberPriority in 0..255) { "subscriber_priority must fit in a byte" }
+        require(groupOrder in 0..255) { "group_order must fit in a byte" }
+    }
+
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (other !is Subscribe) return false
+        return subscribeId == other.subscribeId &&
+            trackAlias == other.trackAlias &&
+            namespace == other.namespace &&
+            trackName.contentEquals(other.trackName) &&
+            subscriberPriority == other.subscriberPriority &&
+            groupOrder == other.groupOrder &&
+            filter == other.filter &&
+            parameters == other.parameters
+    }
+
+    override fun hashCode(): Int {
+        var result = subscribeId.hashCode()
+        result = 31 * result + trackAlias.hashCode()
+        result = 31 * result + namespace.hashCode()
+        result = 31 * result + trackName.contentHashCode()
+        result = 31 * result + subscriberPriority
+        result = 31 * result + groupOrder
+        result = 31 * result + filter.hashCode()
+        result = 31 * result + parameters.hashCode()
+        return result
+    }
+}
+
+/**
+ * SUBSCRIBE_OK (0x04): publisher confirms a subscribe request. When
+ * [contentExists] is true the publisher also reports the largest group/object
+ * it has delivered so far.
+ */
+data class SubscribeOk(
+    val subscribeId: Long,
+    val expiresMs: Long,
+    val groupOrder: Int,
+    val contentExists: Boolean,
+    val largestGroupId: Long? = null,
+    val largestObjectId: Long? = null,
+    val parameters: List<SetupParameter> = emptyList(),
+) : MoqMessage() {
+    override val type: MoqMessageType = MoqMessageType.SubscribeOk
+
+    init {
+        require(groupOrder in 0..255) { "group_order must fit in a byte" }
+        if (contentExists) {
+            requireNotNull(largestGroupId) { "largestGroupId required when contentExists=true" }
+            requireNotNull(largestObjectId) { "largestObjectId required when contentExists=true" }
+        } else {
+            require(largestGroupId == null && largestObjectId == null) {
+                "largestGroupId/largestObjectId must be null when contentExists=false"
+            }
+        }
+    }
+}
+
+/** SUBSCRIBE_ERROR (0x05): publisher rejects a subscribe request. */
+data class SubscribeError(
+    val subscribeId: Long,
+    val errorCode: Long,
+    val reasonPhrase: String,
+    val trackAlias: Long,
+) : MoqMessage() {
+    override val type: MoqMessageType = MoqMessageType.SubscribeError
+}
+
+/** UNSUBSCRIBE (0x0A): subscriber asks the publisher to stop a subscription. */
+data class Unsubscribe(
+    val subscribeId: Long,
+) : MoqMessage() {
+    override val type: MoqMessageType = MoqMessageType.Unsubscribe
+}

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqObject.kt

@@ -0,0 +1,129 @@
+/*
+ * 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
+
+/**
+ * A single MoQ object — a protocol-level unit of publisher-produced data
+ * (one Opus frame for nests audio) identified by its (track, group, object)
+ * coordinates and carrying application bytes.
+ *
+ * MoQ objects are delivered in three ways per draft-ietf-moq-transport:
+ *
+ *   1. OBJECT_DATAGRAM — one object per QUIC datagram. Lowest latency, no
+ *      retransmits. Used by nests for real-time audio.
+ *   2. STREAM_HEADER_SUBGROUP — multiple objects per uni stream, reliable.
+ *   3. FETCH_HEADER — historical objects over a bidi stream.
+ *
+ * Phase 3c-2 covers only (1). Stream-delivered objects arrive in Phase 3c-3.
+ */
+data class MoqObject(
+    val trackAlias: Long,
+    val groupId: Long,
+    val objectId: Long,
+    val publisherPriority: Int,
+    val payload: ByteArray,
+    val status: Long = STATUS_NORMAL,
+) {
+    init {
+        require(publisherPriority in 0..255) { "publisher_priority must fit in a byte" }
+    }
+
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (other !is MoqObject) return false
+        return trackAlias == other.trackAlias &&
+            groupId == other.groupId &&
+            objectId == other.objectId &&
+            publisherPriority == other.publisherPriority &&
+            status == other.status &&
+            payload.contentEquals(other.payload)
+    }
+
+    override fun hashCode(): Int {
+        var result = trackAlias.hashCode()
+        result = 31 * result + groupId.hashCode()
+        result = 31 * result + objectId.hashCode()
+        result = 31 * result + publisherPriority
+        result = 31 * result + status.hashCode()
+        result = 31 * result + payload.contentHashCode()
+        return result
+    }
+
+    companion object {
+        /** Normal object with a payload. */
+        const val STATUS_NORMAL: Long = 0x00L
+
+        /**
+         * Indicates the object intentionally carries no payload (e.g. a
+         * keyframe boundary marker). draft-ietf-moq-transport defines extra
+         * status codes; we expose the raw varint so callers can interpret.
+         */
+        const val STATUS_OBJECT_DOES_NOT_EXIST: Long = 0x01L
+    }
+}
+
+/**
+ * Encoder / decoder for the OBJECT_DATAGRAM wire format.
+ *
+ * Per draft-ietf-moq-transport, the datagram layout is:
+ *
+ *   track_alias (varint)
+ *   group_id (varint)
+ *   object_id (varint)
+ *   publisher_priority (uint8)
+ *   object_status (varint, present only when payload is empty per status semantics)
+ *   payload (remaining bytes of the datagram)
+ *
+ * We deliberately serialise [MoqObject.status] even when the payload is
+ * non-empty, using status=0 for the normal case. That matches what several
+ * reference MoQ implementations emit and makes the codec symmetric; real
+ * servers accept a status byte followed by payload bytes without issue.
+ */
+object MoqObjectDatagram {
+    fun encode(obj: MoqObject): ByteArray {
+        val w = MoqWriter(32 + obj.payload.size)
+        w.writeVarint(obj.trackAlias)
+        w.writeVarint(obj.groupId)
+        w.writeVarint(obj.objectId)
+        w.writeByte(obj.publisherPriority)
+        w.writeVarint(obj.status)
+        w.writeBytes(obj.payload)
+        return w.toByteArray()
+    }
+
+    fun decode(datagram: ByteArray): MoqObject {
+        val r = MoqReader(datagram)
+        val trackAlias = r.readVarint()
+        val groupId = r.readVarint()
+        val objectId = r.readVarint()
+        val publisherPriority = r.readByte()
+        val status = r.readVarint()
+        val payload = if (r.remaining > 0) r.readBytes(r.remaining) else ByteArray(0)
+        return MoqObject(
+            trackAlias = trackAlias,
+            groupId = groupId,
+            objectId = objectId,
+            publisherPriority = publisherPriority,
+            status = status,
+            payload = payload,
+        )
+    }
+}