feat(nestsclient): version-aware Lite-03/04 codec + ALPN negotiation — moq-lite Lite-04

Lite-03 audit L1: completes the moq-lite version surface so the session speaks the wire-format version the server picks via WebTransport ALPN negotiation. Pre-fix the codec hardcoded Lite-03; advertising `moq-lite-04` was a footgun because a Lite-04-preferring relay would desync on the very first Announce exchange. Verified diff between Lite-03 and Lite-04 by WebFetching kixelated's Rust reference (`rs/moq-lite/src/lite/{announce,subscribe,probe}.rs`, `rs/moq-lite/src/version.rs`, `rs/moq-lite/src/model/origin.rs`): exactly three fields differ; everything else is byte-identical. - `Announce.hops`: Lite-03 wire = single varint count (the spec explicitly fills with `Origin::UNKNOWN` placeholders on decode); Lite-04 wire = `varint(count) + count × varint(originId)` (the `OriginList`). MAX_HOPS = 32. - `AnnouncePlease.excludeHop`: Lite-03 absent; Lite-04 single varint after `prefix`. Sentinel `0` = no exclusion. - `Probe.rtt`: Lite-03 absent; Lite-04 single varint after `bitrate`. Sentinel `0` = unknown (decoded as null). Outgoing `Some(0)` is clamped to `Some(1)` to avoid colliding with the sentinel — mirrors kixelated's `encode_msg` clamp. New surface: - `MoqLiteVersion` enum (LITE_03, LITE_04) with `fromAlpn` lookup. - `MoqLiteCodec.{encode,decode}{AnnouncePlease,Announce,Probe}` take `version: MoqLiteVersion = LITE_03` parameter, branch on it. Default LITE_03 keeps existing call sites compiling. - `MoqLiteSession.client(transport, scope, version)` carries the version through every codec invocation. - `MoqLiteAnnouncePlease.excludeHop: Long = 0L` (default). - `MoqLiteAnnounce.hops: List<Long>` (was `Long`) — list of origin IDs, bounded to MAX_HOPS=32. Existing call sites migrate `hops = 0L` → `hops = emptyList()`, `hops = 7L` → `hops = List(7) { 0L }`. - `MoqLiteProbe.rtt: Long? = null` (default). Negotiation: - `WebTransportSession.negotiatedSubProtocol: String?` exposes the server's `wt-protocol` selection. `QuicWebTransportFactory` parses the response HEADERS, extracts the SF-string from `wt-protocol`, and threads it into `QuicWebTransportSession`. `FakeWebTransport.pair(negotiatedSubProtocol = …)` lets tests drive the same path. - Default advertised list now `[moq-lite-04, moq-lite-03]` (was `[moq-lite-03]`). Lite-04 sits first to match kixelated's preference; servers that don't support it fall back to Lite-03 cleanly. - `connectNestsListener` / `connectNestsSpeaker` resolve the version via `resolveMoqLiteVersion(webTransport.negotiatedSubProtocol)` and pass it to `MoqLiteSession.client(...)`. Falls back to Lite-03 when the server doesn't echo `wt-protocol` (older deployments) or echoes something unrecognised (forward-compat). - New `MOQ_LITE_04_VERSION = 0x6D71_6C04L` synthetic version code; `versionCode(version)` mapping function. Regression tests (all green): - `MoqLiteCodecTest.announcePlease_lite04_round_trips_excludeHop` - `MoqLiteCodecTest.announcePlease_lite03_omits_excludeHop_on_wire` - `MoqLiteCodecTest.announce_lite04_round_trips_full_origin_list` - `MoqLiteCodecTest.announce_lite03_drops_origin_ids_keeps_count` - `MoqLiteCodecTest.announce_decoder_rejects_oversize_hop_count` (MAX_HOPS bounds check) - `MoqLiteCodecTest.probe_lite04_round_trips_rtt` - `MoqLiteCodecTest.probe_lite04_clamps_some_zero_to_one_to_avoid_unknown_sentinel` - `MoqLiteCodecTest.probe_lite04_decodes_zero_rtt_as_null` - `MoqLiteCodecTest.probe_lite03_wire_omits_rtt` - `MoqLiteSessionTest.lite04_announce_round_trips_full_origin_list_through_session` (end-to-end Lite-04 session) Audit doc: nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md (L1). https://claude.ai/code/session_012TGfo99Ugz7fcCPv85a8Tq
2026-05-09 15:05:54 +00:00
parent d24953d98c
commit 335a337283
10 changed files with 627 additions and 93 deletions
@@ -24,6 +24,7 @@ import com.vitorpamplona.nestsclient.audio.AudioCapture
 import com.vitorpamplona.nestsclient.audio.OpusEncoder
 import com.vitorpamplona.nestsclient.moq.SubscribeHandle
 import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession
+import com.vitorpamplona.nestsclient.moq.lite.MoqLiteVersion
 import com.vitorpamplona.nestsclient.transport.WebTransportException
 import com.vitorpamplona.nestsclient.transport.WebTransportFactory
 import com.vitorpamplona.nestsclient.transport.WebTransportSession
@@ -106,19 +107,22 @@ suspend fun connectNestsListener(

    state.value = NestsListenerState.Connecting(NestsListenerState.Connecting.ConnectStep.MoqHandshake)

-    // moq-lite Lite-03 has NO setup message — the WebTransport handshake
-    // itself is the handshake, version is selected by the ALPN
-    // `moq-lite-03`.
+    // moq-lite has NO setup message — the WebTransport handshake itself
+    // is the handshake. Version (Lite-03 vs Lite-04) is selected by the
+    // ALPN exchange (`wt-available-protocols` / `wt-protocol`); fall
+    // back to Lite-03 if the server didn't echo `wt-protocol` (older
+    // single-protocol deployments).
+    val moqVersion = resolveMoqLiteVersion(webTransport.negotiatedSubProtocol)
    val moq =
        try {
-            MoqLiteSession.client(webTransport, scope)
+            MoqLiteSession.client(webTransport, scope, moqVersion)
        } catch (t: Throwable) {
            runCatching { webTransport.close(0, "moq-lite session init failed") }
            state.value = NestsListenerState.Failed("moq-lite session init failed: ${t.message}", t)
            return failedListener(state)
        }

-    state.value = NestsListenerState.Connected(room, MOQ_LITE_03_VERSION)
+    state.value = NestsListenerState.Connected(room, versionCode(moqVersion))
    return MoqLiteNestsListener(
        session = moq,
        mutableState = state,
@@ -134,6 +138,35 @@ suspend fun connectNestsListener(
 */
 const val MOQ_LITE_03_VERSION: Long = 0x6D71_6C03L

+/**
+ * Synthetic version code for moq-lite Lite-04 sessions — same shape
+ * as [MOQ_LITE_03_VERSION] with the ALPN suffix `-04` in the low
+ * bytes.
+ */
+const val MOQ_LITE_04_VERSION: Long = 0x6D71_6C04L
+
+/**
+ * Resolve the [MoqLiteVersion] to use for a session given the ALPN
+ * the WebTransport server selected (or `null` if the server didn't
+ * echo `wt-protocol`, e.g. older single-protocol deployments).
+ *
+ * Falls back to [MoqLiteVersion.LITE_03] when the negotiated value
+ * is unknown or absent — every nostrnests deployment supports
+ * Lite-03, so this stays safe under any of:
+ *   - server is older than draft-13 (no `wt-protocol` echo)
+ *   - server picked something we don't recognise (e.g. a future
+ *     `moq-lite-05`)
+ *   - client offered only one protocol
+ */
+internal fun resolveMoqLiteVersion(negotiatedSubProtocol: String?): MoqLiteVersion = MoqLiteVersion.fromAlpn(negotiatedSubProtocol) ?: MoqLiteVersion.LITE_03
+
+/** Map a [MoqLiteVersion] to its synthetic [MOQ_LITE_03_VERSION]-family code. */
+internal fun versionCode(version: MoqLiteVersion): Long =
+    when (version) {
+        MoqLiteVersion.LITE_03 -> MOQ_LITE_03_VERSION
+        MoqLiteVersion.LITE_04 -> MOQ_LITE_04_VERSION
+    }
+
 /**
 * Build a no-op [NestsListener] in a Failed state for callers that want a
 * uniform return type even on early-failure paths.
@@ -239,18 +272,19 @@ suspend fun connectNestsSpeaker(

    state.value = NestsSpeakerState.Connecting(NestsSpeakerState.Connecting.ConnectStep.MoqHandshake)

-    // moq-lite Lite-03 has NO setup message. Same logic as the listener
-    // path — version is selected by the `moq-lite-03` ALPN.
+    // moq-lite has NO setup message. Same logic as the listener path —
+    // version is selected by the ALPN exchange.
+    val moqVersion = resolveMoqLiteVersion(webTransport.negotiatedSubProtocol)
    val moq =
        try {
-            MoqLiteSession.client(webTransport, scope)
+            MoqLiteSession.client(webTransport, scope, moqVersion)
        } catch (t: Throwable) {
            runCatching { webTransport.close(0, "moq-lite session init failed") }
            state.value = NestsSpeakerState.Failed("moq-lite session init failed: ${t.message}", t)
            return failedSpeaker(state)
        }

-    state.value = NestsSpeakerState.Connected(room, MOQ_LITE_03_VERSION)
+    state.value = NestsSpeakerState.Connected(room, versionCode(moqVersion))
    return MoqLiteNestsSpeaker(
        session = moq,
        speakerPubkeyHex = speakerPubkeyHex,
@@ -47,12 +47,65 @@ object MoqLiteAlpn {
    const val LITE_03: String = "moq-lite-03"

    /**
-     * `moq-lite-04` ALPN string. Wire-incompatible with [MoqLiteCodec]
-     * today — see the object kdoc for the codec diff. Defined here so
-     * a future patch that lands version-aware Announce / Probe codecs
-     * can drop it into the [QuicWebTransportFactory] sub-protocol list
-     * without re-deriving the constant.
+     * `moq-lite-04` ALPN string. Wire-compatible with [MoqLiteCodec]
+     * when the codec is invoked with [MoqLiteVersion.LITE_04] — see
+     * [MoqLiteVersion] for the version-conditional codec branches.
+     * Lite-04 reshapes `Announce.hops` (varint count → `OriginList`),
+     * adds `AnnounceInterest.exclude_hop`, and adds `Probe.rtt`.
+     * Subscribe / SubscribeOk / Drop / Group / SubscribeResponse are
+     * unchanged from Lite-03. The factory advertises Lite-04 ahead
+     * of Lite-03 to match kixelated's preference order.
     */
    const val LITE_04: String = "moq-lite-04"
    const val LEGACY: String = "moql"
 }
+
+/**
+ * Version discriminator for the [MoqLiteCodec] / [MoqLiteSession]
+ * version-aware code paths. Selected at WebTransport CONNECT time
+ * via the `wt-available-protocols` / `wt-protocol` ALPN exchange
+ * (see `QuicWebTransportFactory` and
+ * [com.vitorpamplona.nestsclient.transport.WebTransportSession.negotiatedSubProtocol]).
+ *
+ * The wire delta between [LITE_03] and [LITE_04] is exactly three
+ * fields:
+ *   - `Announce.hops`: Lite-03 = single varint count; Lite-04 =
+ *     `varint(count) + count × varint(originId)` (the `OriginList`).
+ *   - `AnnounceInterest.exclude_hop` (= our `AnnouncePlease.excludeHop`):
+ *     Lite-03 = absent; Lite-04 = a single varint after `prefix`
+ *     (sentinel `0` = no exclusion).
+ *   - `Probe.rtt`: Lite-03 = absent; Lite-04 = a single varint after
+ *     `bitrate` (sentinel `0` = unknown; outgoing `Some(0)` is
+ *     clamped to `Some(1)` to avoid colliding with the sentinel).
+ *
+ * Everything else (`Subscribe`, `SubscribeOk`, `SubscribeDrop`,
+ * `SubscribeResponse`, `GroupHeader`, `ControlType`, `DataType`,
+ * `AnnounceStatus`) is byte-for-byte identical between the two
+ * versions.
+ *
+ * Source: `kixelated/moq` `rs/moq-lite/src/lite/{announce,probe,
+ * subscribe}.rs` Lite-04 branches (verified against `main` @
+ * 2026-05-09).
+ */
+enum class MoqLiteVersion(
+    val alpn: String,
+) {
+    LITE_03(MoqLiteAlpn.LITE_03),
+    LITE_04(MoqLiteAlpn.LITE_04),
+    ;
+
+    companion object {
+        /**
+         * Resolve a negotiated `wt-protocol` value back to the
+         * version enum, or `null` if the string isn't a recognised
+         * moq-lite version. Caller decides whether to fall back to
+         * a default ([LITE_03]) or fail loudly.
+         */
+        fun fromAlpn(alpn: String?): MoqLiteVersion? =
+            when (alpn) {
+                MoqLiteAlpn.LITE_03 -> LITE_03
+                MoqLiteAlpn.LITE_04 -> LITE_04
+                else -> null
+            }
+    }
+}
@@ -49,37 +49,86 @@ import com.vitorpamplona.nestsclient.moq.MoqWriter
 object MoqLiteCodec {
    // ---------------- AnnouncePlease ----------------

-    fun encodeAnnouncePlease(msg: MoqLiteAnnouncePlease): ByteArray {
+    fun encodeAnnouncePlease(
+        msg: MoqLiteAnnouncePlease,
+        version: MoqLiteVersion = MoqLiteVersion.LITE_03,
+    ): ByteArray {
        val body = MoqWriter()
        body.writeLengthPrefixedString(MoqLitePath.normalize(msg.prefix))
+        // Lite-04 appends `excludeHop` as a single varint after
+        // prefix. Sentinel `0` = no exclusion. Lite-03 omits the
+        // field entirely; ignore [msg.excludeHop] in that case.
+        if (version == MoqLiteVersion.LITE_04) {
+            body.writeVarint(msg.excludeHop)
+        }
        return wrapSizePrefixed(body)
    }

-    fun decodeAnnouncePlease(payload: ByteArray): MoqLiteAnnouncePlease {
+    fun decodeAnnouncePlease(
+        payload: ByteArray,
+        version: MoqLiteVersion = MoqLiteVersion.LITE_03,
+    ): MoqLiteAnnouncePlease {
        val r = MoqReader(payload)
        val prefix = MoqLitePath.normalize(r.readLengthPrefixedString())
+        val excludeHop =
+            if (version == MoqLiteVersion.LITE_04) r.readVarint() else 0L
        ensureFullyConsumed(r, "AnnouncePlease")
-        return MoqLiteAnnouncePlease(prefix = prefix)
+        return MoqLiteAnnouncePlease(prefix = prefix, excludeHop = excludeHop)
    }

    // ---------------- Announce ----------------

-    fun encodeAnnounce(msg: MoqLiteAnnounce): ByteArray {
+    fun encodeAnnounce(
+        msg: MoqLiteAnnounce,
+        version: MoqLiteVersion = MoqLiteVersion.LITE_03,
+    ): ByteArray {
        val body = MoqWriter()
        body.writeByte(msg.status.code)
        body.writeLengthPrefixedString(MoqLitePath.normalize(msg.suffix))
-        body.writeVarint(msg.hops)
+        // Lite-03: just the count. Lite-04: count + each origin id.
+        // Mirrors kixelated's `encode_hops` switch.
+        body.writeVarint(msg.hops.size.toLong())
+        if (version == MoqLiteVersion.LITE_04) {
+            for (originId in msg.hops) {
+                body.writeVarint(originId)
+            }
+        }
        return wrapSizePrefixed(body)
    }

-    fun decodeAnnounce(payload: ByteArray): MoqLiteAnnounce {
+    fun decodeAnnounce(
+        payload: ByteArray,
+        version: MoqLiteVersion = MoqLiteVersion.LITE_03,
+    ): MoqLiteAnnounce {
        val r = MoqReader(payload)
        val statusByte = r.readByte()
        val status =
            MoqLiteAnnounceStatus.fromCode(statusByte)
                ?: throw MoqCodecException("unknown moq-lite Announce status byte: $statusByte")
        val suffix = MoqLitePath.normalize(r.readLengthPrefixedString())
-        val hops = r.readVarint()
+        val count = r.readVarint()
+        if (count < 0L || count > MoqLiteAnnounce.MAX_HOPS) {
+            throw MoqCodecException("Announce hops count out of range [0, ${MoqLiteAnnounce.MAX_HOPS}]: $count")
+        }
+        val hops: List<Long> =
+            when (version) {
+                MoqLiteVersion.LITE_03 -> {
+                    // Wire only carries a count; fill with UNKNOWN
+                    // placeholders (origin id 0). Mirrors kixelated's
+                    // `Origin::UNKNOWN` fill in announce.rs.
+                    List(count.toInt()) { 0L }
+                }
+
+                MoqLiteVersion.LITE_04 -> {
+                    val list = ArrayList<Long>(count.toInt())
+                    repeat(count.toInt()) {
+                        val id = r.readVarint()
+                        require(id >= 0) { "decoded origin id must be non-negative" }
+                        list.add(id)
+                    }
+                    list
+                }
+            }
        ensureFullyConsumed(r, "Announce")
        return MoqLiteAnnounce(status = status, suffix = suffix, hops = hops)
    }
@@ -244,24 +293,52 @@ object MoqLiteCodec {

    // ---------------- Probe ----------------

-    fun decodeProbe(payload: ByteArray): MoqLiteProbe {
+    fun decodeProbe(
+        payload: ByteArray,
+        version: MoqLiteVersion = MoqLiteVersion.LITE_03,
+    ): MoqLiteProbe {
        val r = MoqReader(payload)
        val bitrate = r.readVarint()
+        // Lite-04 appends `rtt` as a single varint after bitrate.
+        // Sentinel `0` = unknown (decoded as null).
+        val rtt: Long? =
+            if (version == MoqLiteVersion.LITE_04) {
+                val raw = r.readVarint()
+                if (raw == 0L) null else raw
+            } else {
+                null
+            }
        ensureFullyConsumed(r, "Probe")
-        return MoqLiteProbe(bitrate = bitrate)
+        return MoqLiteProbe(bitrate = bitrate, rtt = rtt)
    }

    /**
-     * Encode a single Lite-03 Probe message body
-     * (`lite/probe.rs` — `bitrate: u62` only; `rtt` is Lite-04+).
-     * The publisher writes these size-prefixed onto a Probe bidi the
+     * Encode a Probe message body. Lite-03 writes only `bitrate`;
+     * Lite-04 appends `rtt` (sentinel `0` = unknown; outgoing
+     * `Some(0)` is clamped to `Some(1)` to avoid colliding with the
+     * sentinel — matches kixelated's `encode_msg` clamp). The
+     * publisher writes these size-prefixed onto a Probe bidi the
     * subscriber opened, advertising the publisher's expected
-     * bandwidth. Wrapping (size prefix) is the caller's responsibility,
-     * matching [encodeAnnouncePlease] / [encodeAnnounce].
+     * bandwidth. Wrapping (size prefix) is the caller's
+     * responsibility, matching [encodeAnnouncePlease] / [encodeAnnounce].
     */
-    fun encodeProbe(probe: MoqLiteProbe): ByteArray {
+    fun encodeProbe(
+        probe: MoqLiteProbe,
+        version: MoqLiteVersion = MoqLiteVersion.LITE_03,
+    ): ByteArray {
        val body = MoqWriter()
        body.writeVarint(probe.bitrate)
+        if (version == MoqLiteVersion.LITE_04) {
+            // null → 0 (unknown sentinel). Some(0) → 1 (clamp).
+            // Otherwise pass through.
+            val wire =
+                when (val r = probe.rtt) {
+                    null -> 0L
+                    0L -> 1L
+                    else -> r
+                }
+            body.writeVarint(wire)
+        }
        return wrapSizePrefixed(body)
    }

@@ -30,16 +30,32 @@ package com.vitorpamplona.nestsclient.moq.lite

 /**
 * "I'm interested in broadcasts under this prefix" — the first message
- * the subscriber writes on an Announce bidi.
+ * the subscriber writes on an Announce bidi. Mirrors kixelated's
+ * `AnnounceInterest` struct (`rs/moq-lite/src/lite/announce.rs`).
 *
 * Wire layout (size-prefixed):
- *   prefix  string  (varint length + UTF-8)
+ *   prefix       string  (varint length + UTF-8)
+ *   excludeHop   u62 varint  (Lite-04 ONLY; sentinel `0` = no exclusion)
 *
- * Empty prefix means "everything".
+ * Empty prefix means "everything". `excludeHop != 0` (Lite-04 only)
+ * asks the publisher to skip announces whose hop ID list contains
+ * this value — used in clustered moq-rs deployments to break
+ * forwarding loops.
 */
 data class MoqLiteAnnouncePlease(
    val prefix: String,
-)
+    /**
+     * Lite-04 only. `0` (the default) means "no exclusion." A
+     * non-zero value asks the publisher to skip announces whose
+     * `hops` list contains this origin ID. Encoded as a single
+     * varint after [prefix]; absent on the wire under Lite-03.
+     */
+    val excludeHop: Long = 0L,
+) {
+    init {
+        require(excludeHop >= 0) { "excludeHop must be non-negative: $excludeHop" }
+    }
+}

 /**
 * Per-broadcast announce update streamed by the publisher (or relay)
@@ -52,13 +68,37 @@ data class MoqLiteAnnouncePlease(
 *   suffix  string   (broadcast path with the requested `prefix`
 *                     stripped — `MoqLitePath.join(prefix, suffix)`
 *                     reconstitutes the absolute path)
- *   hops    u62      (relay routing depth, Lite-03 only)
+ *   hops    OriginList  (Lite-03: a single varint count;
+ *                        Lite-04: `varint(count) + count × varint(id)`)
+ *
+ * [hops] is modeled as `List<Long>` — origin IDs the announce has
+ * traversed. Empty list = no hops. Origin id `0` is reserved as
+ * `Origin::UNKNOWN` (used by Lite-03 decode to fill the count of
+ * placeholder entries when the wire only carries a count). Per
+ * kixelated, [hops] is bounded to 32 entries (`MAX_HOPS`).
 */
 data class MoqLiteAnnounce(
    val status: MoqLiteAnnounceStatus,
    val suffix: String,
-    val hops: Long,
-)
+    val hops: List<Long>,
+) {
+    init {
+        require(hops.size <= MAX_HOPS) { "hops list must not exceed MAX_HOPS=$MAX_HOPS, got ${hops.size}" }
+        // Origin id is a u62 varint; reject negatives so encoding
+        // can't silently produce a malformed varint. Per kixelated,
+        // id == 0 means UNKNOWN; otherwise must fit in 62 bits.
+        require(hops.all { it >= 0 }) { "hops origin ids must be non-negative" }
+    }
+
+    companion object {
+        /**
+         * Maximum number of origin entries in [hops]. Mirrors
+         * `kixelated/moq`'s `Origin::MAX_HOPS = 32`; the decoder
+         * rejects oversize lists with a codec exception.
+         */
+        const val MAX_HOPS: Int = 32
+    }
+}

 /**
 * "Subscribe me to (broadcast, track)" — the first message the
@@ -210,12 +250,28 @@ data class MoqLiteGroupHeader(
 /**
 * Probe message written by the *publisher* on a subscriber-initiated
 * Probe bidi (ControlType=4). `bitrate` is encoded as a u62 varint in
- * a size-prefixed body. Decode-only for now — the listener path
- * exposes the most recent reading, and we don't initiate probes from
- * the speaker side.
+ * a size-prefixed body.
+ *
+ * Wire layout (size-prefixed):
+ *   bitrate  u62 varint
+ *   rtt      u62 varint  (Lite-04 ONLY; sentinel `0` = unknown.
+ *                         Outgoing `Some(0)` is clamped to `Some(1)`
+ *                         to avoid colliding with the sentinel.)
 *
 * Source: `rs/moq-lite/src/lite/probe.rs`.
 */
 data class MoqLiteProbe(
    val bitrate: Long,
-)
+    /**
+     * Round-trip time hint, in the unit the application and peer
+     * have agreed to (kixelated's Rust type is plain `Option<u64>`
+     * — no unit attached at the type level). Lite-04 only; absent
+     * on the wire under Lite-03. `null` = unknown.
+     */
+    val rtt: Long? = null,
+) {
+    init {
+        require(bitrate >= 0) { "bitrate must be non-negative: $bitrate" }
+        if (rtt != null) require(rtt >= 0) { "rtt must be non-negative when present: $rtt" }
+    }
+}
@@ -77,6 +77,23 @@ class MoqLiteSession internal constructor(
     */
    internal val transport: WebTransportSession,
    private val scope: CoroutineScope,
+    /**
+     * Wire-format version this session speaks. Selected at
+     * WebTransport CONNECT time via the
+     * `wt-available-protocols` / `wt-protocol` exchange (see
+     * [com.vitorpamplona.nestsclient.transport.WebTransportSession.negotiatedSubProtocol]).
+     * Defaults to [MoqLiteVersion.LITE_03] for backward compat with
+     * call sites that don't pass an explicit version (most tests +
+     * the legacy single-version factory). Production
+     * `connectNestsListener` / `connectNestsSpeaker` pass the
+     * resolved version from the negotiated ALPN.
+     *
+     * The version selects between the version-conditional codec
+     * branches in [MoqLiteCodec] for `Announce.hops`,
+     * `AnnouncePlease.excludeHop`, and `Probe.rtt`. All other
+     * messages are wire-identical between the two versions.
+     */
+    val version: MoqLiteVersion = MoqLiteVersion.LITE_03,
 ) {
    private val state = Mutex()
    private val subscriptionsBySubscribeId: MutableMap<Long, ListenerSubscription> = HashMap()
@@ -132,7 +149,7 @@ class MoqLiteSession internal constructor(
        ensureOpen()
        val bidi = transport.openBidiStream()
        bidi.write(Varint.encode(MoqLiteControlType.Announce.code))
-        bidi.write(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease(prefix)))
+        bidi.write(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease(prefix), version))

        // replay=64, DROP_OLDEST: announces emitted before the
        // caller's `collect` attaches MUST NOT be dropped — that's
@@ -164,7 +181,7 @@ class MoqLiteSession internal constructor(
                        buffer.push(chunk)
                        while (true) {
                            val payload = buffer.readSizePrefixed() ?: break
-                            val decoded = MoqLiteCodec.decodeAnnounce(payload)
+                            val decoded = MoqLiteCodec.decodeAnnounce(payload, version)
                            emitCount += 1
                            Log.d("NestRx") {
                                "session.announce(prefix='$prefix') bidi pump emit #$emitCount " +
@@ -733,7 +750,7 @@ class MoqLiteSession internal constructor(
                        buffer.push(chunk)
                        while (true) {
                            val payload = buffer.readSizePrefixed() ?: break
-                            updates.emit(MoqLiteCodec.decodeProbe(payload))
+                            updates.emit(MoqLiteCodec.decodeProbe(payload, version))
                        }
                    }
                } catch (ce: CancellationException) {
@@ -949,7 +966,7 @@ class MoqLiteSession internal constructor(
                    when (controlType) {
                        MoqLiteControlType.Announce -> {
                            val pleasePayload = buffer.readSizePrefixed() ?: return@collect
-                            val please = MoqLiteCodec.decodeAnnouncePlease(pleasePayload)
+                            val please = MoqLiteCodec.decodeAnnouncePlease(pleasePayload, version)
                            // Per moq-lite Lite-03 (`rs/moq-lite/src/lite/announce.rs`),
                            // a publisher MUST only emit Active for broadcasts whose
                            // path starts with the requested prefix. If our suffix
@@ -976,8 +993,9 @@ class MoqLiteSession internal constructor(
                                    MoqLiteAnnounce(
                                        status = MoqLiteAnnounceStatus.Active,
                                        suffix = emittedSuffix,
-                                        hops = 0L,
+                                        hops = emptyList(),
                                    ),
+                                    version,
                                ),
                            )
                            announcePublisher.registerAnnounceBidi(bidi, emittedSuffix)
@@ -1134,7 +1152,7 @@ class MoqLiteSession internal constructor(
                            // which left the subscriber's ABR
                            // estimator with no signal.
                            runCatching {
-                                bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = NESTS_AUDIO_BITRATE_HINT_BPS)))
+                                bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = NESTS_AUDIO_BITRATE_HINT_BPS), version))
                                bidi.finish()
                            }
                            dispatched = true
@@ -1511,8 +1529,9 @@ class MoqLiteSession internal constructor(
                            MoqLiteAnnounce(
                                status = MoqLiteAnnounceStatus.Ended,
                                suffix = entry.emittedSuffix,
-                                hops = 0L,
+                                hops = emptyList(),
                            ),
+                            version,
                        ),
                    )
                }
@@ -1607,6 +1626,7 @@ class MoqLiteSession internal constructor(
        fun client(
            transport: WebTransportSession,
            pumpScope: CoroutineScope,
-        ): MoqLiteSession = MoqLiteSession(transport, pumpScope)
+            version: MoqLiteVersion = MoqLiteVersion.LITE_03,
+        ): MoqLiteSession = MoqLiteSession(transport, pumpScope, version)
    }
 }
@@ -47,6 +47,15 @@ class FakeWebTransport private constructor(
    private val inboundBidiStreams: Channel<FakeBidiStream>,
    private val inboundUniStreams: Channel<WebTransportReadStream>,
    private val outboundUniStreams: Channel<WebTransportReadStream>,
+    /**
+     * Test-controllable negotiated sub-protocol. Defaults to `null`
+     * — production fakes that don't care about ALPN negotiation
+     * keep the same behavior they had before this field existed.
+     * Tests that exercise the version-aware [MoqLiteSession] can
+     * pass a non-null value via [pair] to simulate a relay that
+     * picked Lite-04 (or any other ALPN string).
+     */
+    override val negotiatedSubProtocol: String? = null,
 ) : WebTransportSession {
    private val stateLock = Mutex()
    private var open = true
@@ -170,8 +179,14 @@ class FakeWebTransport private constructor(
         * Create two linked fakes that act as "client" and "server" endpoints of
         * the same virtual WebTransport session. Datagrams and streams opened on
         * one side appear on the other.
+         *
+         * @param negotiatedSubProtocol surfaced by both sides via
+         *   [WebTransportSession.negotiatedSubProtocol]. Defaults to
+         *   `null`. Tests that drive a version-aware
+         *   [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession]
+         *   pass e.g. `"moq-lite-04"` here.
         */
-        fun pair(): Pair<FakeWebTransport, FakeWebTransport> {
+        fun pair(negotiatedSubProtocol: String? = null): Pair<FakeWebTransport, FakeWebTransport> {
            val aToBDatagrams = Channel<ByteArray>(Channel.BUFFERED)
            val bToADatagrams = Channel<ByteArray>(Channel.BUFFERED)
            val aToBBidi = Channel<FakeBidiStream>(Channel.BUFFERED)
@@ -187,6 +202,7 @@ class FakeWebTransport private constructor(
                    inboundBidiStreams = bToABidi,
                    inboundUniStreams = bToAUni,
                    outboundUniStreams = aToBUni,
+                    negotiatedSubProtocol = negotiatedSubProtocol,
                )
            val b =
                FakeWebTransport(
@@ -196,6 +212,7 @@ class FakeWebTransport private constructor(
                    inboundBidiStreams = aToBBidi,
                    inboundUniStreams = aToBUni,
                    outboundUniStreams = bToAUni,
+                    negotiatedSubProtocol = negotiatedSubProtocol,
                )
            return a to b
        }
@@ -42,6 +42,21 @@ interface WebTransportSession {
    /** True once the Extended CONNECT exchange has returned 2xx and before [close] is called. */
    val isOpen: Boolean

+    /**
+     * The sub-protocol the server selected via the
+     * `wt-protocol` response header (RFC 8941 SF-string;
+     * draft-ietf-webtrans-http3 §3.3). `null` if the server didn't
+     * echo a sub-protocol — typically because the client offered
+     * none, or because the server is older than draft-13 and
+     * supports only a single protocol implicitly.
+     *
+     * Used by moq-lite negotiation: a client that advertised both
+     * `moq-lite-04` and `moq-lite-03` reads this to select the
+     * matching [com.vitorpamplona.nestsclient.moq.lite.MoqLiteVersion]
+     * for its [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession].
+     */
+    val negotiatedSubProtocol: String?
+
    /**
     * Open a new bidirectional WebTransport stream. The returned [WebTransportBidiStream]
     * is writable + readable and closes when either peer half-closes.