feat(nestsclient): subscriber-driven Probe API (MoqLiteSession.probe) — moq-lite Lite-03

Lite-03 audit L3: completes the subscriber-side Probe surface to
mirror kixelated's `Subscriber::run_probe_stream`
(`rs/moq-lite/src/lite/subscriber.rs`).

`MoqLiteSession.probe()` opens a bidi, writes
`ControlType::Probe` (varint 4), and returns a `MoqLiteProbeHandle`
whose `updates` flow yields each size-prefixed `MoqLiteProbe`
message the publisher pushes. The handle's `close()` FINs the
bidi and cancels the pump.

`updates` is a `MutableSharedFlow(replay=8)` so a collector that
attaches after the publisher's first emit doesn't miss it
(matches the same shape the announce-watch uses).

No production consumer for the API today — Amethyst's nests
listener doesn't run ABR on a fixed-rate Opus encoder. The API
exists to round out the moq-lite Lite-03 surface: a diagnostic
tool can now read a publisher's bitrate without subscribing to
its data track.

The publisher-side handler (existing) was already correct: it
writes one `MoqLiteProbe` (32 kbps Opus voice hint) and FINs.

Regression test:
`MoqLiteSessionTest.subscriber_probe_writes_control_type_and_decodes_publisher_bitrate`.

Audit doc: nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md (L3).

https://claude.ai/code/session_012TGfo99Ugz7fcCPv85a8Tq

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt

@@ -54,6 +54,28 @@ class MoqLiteAnnouncesHandle internal constructor(
     suspend fun close() = close.invoke()
 }
 
+/**
+ * Active probe handle returned by [MoqLiteSession.probe]. Mirrors
+ * kixelated's `Subscriber::run_probe_stream`
+ * (`rs/moq-lite/src/lite/subscriber.rs`): subscriber opens a bidi,
+ * writes `ControlType::Probe`, then reads size-prefixed
+ * [MoqLiteProbe] messages indefinitely until the publisher FINs or
+ * the consumer calls [close].
+ *
+ * [updates] is a cold flow that emits every Probe message the
+ * publisher pushes — typically a single bitrate hint at session
+ * start, but a publisher running an ABR codec MAY emit multiple
+ * over time. For a fixed-rate Opus producer (which is what
+ * Amethyst's nests speaker is) the publisher emits one and FINs;
+ * the flow then completes naturally and consumers see end-of-flow.
+ */
+class MoqLiteProbeHandle internal constructor(
+    val updates: Flow<MoqLiteProbe>,
+    private val close: suspend () -> Unit,
+) {
+    suspend fun close() = close.invoke()
+}
+
 /** Thrown when subscribe is rejected (Drop) or the response stream dies. */
 class MoqLiteSubscribeException(
     message: String,

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt

@@ -691,6 +691,67 @@ class MoqLiteSession internal constructor(
         sub.frames.close()
     }
 
+    /**
+     * Open a Probe bidi against the relay / publisher (audit L3).
+     * Mirrors kixelated's `Subscriber::run_probe_stream`
+     * (`rs/moq-lite/src/lite/subscriber.rs`): subscriber writes
+     * `ControlType::Probe` (varint 4), then reads size-prefixed
+     * [MoqLiteProbe] messages until the publisher FINs.
+     *
+     * The publisher reciprocates by writing one or more [MoqLiteProbe]
+     * messages indicating its expected outbound bitrate. Amethyst's
+     * own publisher path is fixed-rate Opus and emits exactly one
+     * Probe before FIN, but third-party (e.g. browser-side
+     * `kixelated/moq` watcher → relay-served broadcast) publishers
+     * MAY emit multiple over time as their ABR estimates change.
+     *
+     * The returned handle's `updates` flow is cold — collecting it
+     * starts the read pump on this session's scope. On collector
+     * cancellation OR [MoqLiteProbeHandle.close], the bidi FINs and
+     * the pump exits.
+     *
+     * No application-level use of probe data exists in nests today
+     * (we don't run ABR), but the API completes the moq-lite Lite-03
+     * protocol surface. Useful for diagnostic tools that want to
+     * read a publisher's bitrate without subscribing to its data
+     * track.
+     */
+    suspend fun probe(): MoqLiteProbeHandle {
+        ensureOpen()
+        val bidi = transport.openBidiStream()
+        bidi.write(Varint.encode(MoqLiteControlType.Probe.code))
+        val updates =
+            MutableSharedFlow<MoqLiteProbe>(
+                replay = 8,
+                onBufferOverflow = BufferOverflow.DROP_OLDEST,
+            )
+        val pump =
+            scope.launch {
+                val buffer = MoqLiteFrameBuffer()
+                try {
+                    bidi.incoming().collect { chunk ->
+                        buffer.push(chunk)
+                        while (true) {
+                            val payload = buffer.readSizePrefixed() ?: break
+                            updates.emit(MoqLiteCodec.decodeProbe(payload))
+                        }
+                    }
+                } catch (ce: CancellationException) {
+                    throw ce
+                } catch (_: Throwable) {
+                    // Probe bidi died — best-effort silent. The
+                    // updates flow simply stops emitting.
+                }
+            }
+        return MoqLiteProbeHandle(
+            updates = updates,
+            close = {
+                runCatching { bidi.finish() }
+                pump.cancelAndJoin()
+            },
+        )
+    }
+
     // ====================================================================
     // Publisher side
     // ====================================================================

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt

@@ -940,6 +940,50 @@ class MoqLiteSessionTest {
             session.close()
         }
 
+    @Test
+    fun subscriber_probe_writes_control_type_and_decodes_publisher_bitrate() =
+        runBlocking {
+            // Lite-03 audit L3: the subscriber side opens a Probe bidi
+            // by writing ControlType=Probe and reads size-prefixed
+            // MoqLiteProbe messages from the publisher. Mirrors
+            // kixelated's `Subscriber::run_probe_stream`
+            // (`rs/moq-lite/src/lite/subscriber.rs`).
+            val (clientSide, serverSide) = FakeWebTransport.pair()
+            val session = MoqLiteSession.client(clientSide, pumpScope)
+
+            // Server side: read the subscriber's control byte, then
+            // write two Probe messages.
+            val serverPump =
+                async {
+                    val bidi = serverSide.peerOpenedBidiStreams().first()
+                    val buf = MoqLiteFrameBuffer()
+                    // Read the leading ControlType varint. The session
+                    // writes it as a single chunk via bidi.write.
+                    val ctChunk = withTimeout(2_000) { bidi.incoming().first() }
+                    buf.push(ctChunk)
+                    val ct = buf.readVarint()
+                    assertEquals(MoqLiteControlType.Probe.code, ct, "subscriber writes ControlType=Probe")
+
+                    // Push two probes (different bitrates) then FIN.
+                    bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = 32_000L)))
+                    bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = 64_000L)))
+                    bidi.finish()
+                }
+
+            val handle = session.probe()
+            // Drain both probes from the cold flow. SharedFlow with
+            // replay=8 means we can collect after the publisher has
+            // already pushed without losing emissions.
+            val received = withTimeout(2_000) { handle.updates.take(2).toList() }
+            assertEquals(2, received.size)
+            assertEquals(32_000L, received[0].bitrate)
+            assertEquals(64_000L, received[1].bitrate)
+
+            serverPump.await()
+            handle.close()
+            session.close()
+        }
+
     @Test
     fun unsubscribe_FINs_the_subscribe_bidi() =
         runBlocking {