test/refactor(audio-rooms): nostrnests wire-shape fixes + interop expansion (phase 4)

Wire-shape corrections discovered while scoping the interop test suite against the real moq-rs relay: 1. WebTransport CONNECT path is now /<moqNamespace> (matches the relay's claims.root prefix check). Previously hardcoded "/anon". 2. JWT travels in the ?jwt=<token> query parameter, not the Authorization header — moq-rs only reads the query param. The bearer-token path on QuicWebTransportFactory is now unused for nests; left in place for non-nests WebTransport servers. 3. Harness `moqEndpoint` is the relay base URL only; the connect helpers append /<namespace>?jwt=<token> themselves. Interop test additions (all -DnestsInterop=true gated, default-skipped): - NostrNestsAuthFailureInteropTest — locks in the moq-auth sidecar's rejection paths (missing/wrong-scheme Authorization, NIP-98 signed for the wrong URL, malformed namespace per the strict regex, publish=true grant for any caller — sidecar does NOT gate by NIP-53 hostlist). - NostrNestsAuthEndpointsInteropTest — /health, /.well-known/jwks.json shape (must contain ES256/P-256), 404 on unknown route. - NostrNestsMultiPeerInteropTest — multi-listener fan-out, multi- speaker isolation, subscribe-before-announce. Code is wired through production connectNestsSpeaker / connectNestsListener; will pass once the moq-lite gap (below) is resolved. Major finding documented in nestsClient/plans/2026-04-26-moq-lite-gap.md: nostrnests's stack uses moq-lite (kixelated's variant), NOT IETF draft-ietf-moq-transport which `:nestsClient` currently implements. The two are wire-incompatible — single-string broadcast/track names vs. byte tuples, different ANNOUNCE/SUBSCRIBE framing. The wire-shape fixes here make the WebTransport CONNECT itself succeed, but the post-CONNECT MoQ framing layer still needs a moq-lite codec before round-trip / multi-peer tests can pass against real nests. Pursued as a separate phase.
2026-04-26 15:53:02 +00:00
parent 0ac8c0f791
commit 1887bd1fa7
7 changed files with 910 additions and 10 deletions
@@ -0,0 +1,117 @@
+/*
+ * 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.interop
+
+import kotlinx.coroutines.runBlocking
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import org.junit.AfterClass
+import org.junit.BeforeClass
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * Phase-4 ancillary endpoint smoke tests for the moq-auth sidecar. Cheap
+ * sanity checks for the routes we don't drive through the production
+ * client, so a regression in the sidecar's deployment (a port flip, a
+ * missing route, a busted CORS config) surfaces immediately rather than
+ * silently when the next /auth flow fires.
+ *
+ * Coverage:
+ *   - `GET /health` returns `{"status":"ok"}` (used by Docker health
+ *     probes; we read it for a test-time port-readiness check too)
+ *   - `GET /.well-known/jwks.json` returns a JWK set with at least one
+ *     ES256 verification key — moq-relay polls this every 30 s
+ *     (`MOQ_AUTH_REFRESH_INTERVAL`) so a malformed JWKS would brick
+ *     authentication
+ *   - Unknown route returns 404 (locks down the surface)
+ *
+ * Skipped by default — set `-DnestsInterop=true` to enable.
+ */
+class NostrNestsAuthEndpointsInteropTest {
+    @Test
+    fun health_endpoint_returns_status_ok() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            http.newCall(Request.Builder().url("${harness.authBaseUrl}/health").build()).execute().use { resp ->
+                val body = resp.body.string()
+                assertEquals(200, resp.code, "/health should be 200, got ${resp.code}: $body")
+                assertTrue("\"status\"" in body && "\"ok\"" in body, "expected {\"status\":\"ok\"} JSON, got: $body")
+            }
+        }
+
+    @Test
+    fun jwks_endpoint_returns_at_least_one_es256_signing_key() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            http.newCall(Request.Builder().url("${harness.authBaseUrl}/.well-known/jwks.json").build()).execute().use { resp ->
+                val body = resp.body.string()
+                assertEquals(200, resp.code, "/.well-known/jwks.json should be 200, got ${resp.code}: $body")
+                // Don't pull a JWK parser into tests — substring sniffing is
+                // enough to assert the shape moq-relay actually consumes.
+                assertTrue("\"keys\"" in body, "JWKS body must contain a `keys` array, got: $body")
+                assertTrue(
+                    "\"kty\"" in body,
+                    "JWKS entry must have a `kty` (key type) field per RFC 7517, got: $body",
+                )
+                assertTrue(
+                    "\"alg\":\"ES256\"" in body || "\"crv\":\"P-256\"" in body,
+                    "expected ES256 / P-256 signing key (matches moq-rs claim verification), got: $body",
+                )
+            }
+        }
+
+    @Test
+    fun unknown_route_returns_404() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            http.newCall(Request.Builder().url("${harness.authBaseUrl}/does-not-exist").build()).execute().use { resp ->
+                assertEquals(404, resp.code, "unknown route should be 404")
+            }
+        }
+
+    companion object {
+        private val http = OkHttpClient()
+        private var harnessOrNull: NostrNestsHarness? = null
+
+        @BeforeClass
+        @JvmStatic
+        fun setUpHarness() {
+            if (NostrNestsHarness.isEnabled()) {
+                harnessOrNull = NostrNestsHarness.start()
+            }
+        }
+
+        @AfterClass
+        @JvmStatic
+        fun tearDownHarness() {
+            harnessOrNull?.close()
+            harnessOrNull = null
+        }
+    }
+}
@@ -0,0 +1,231 @@
+/*
+ * 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.interop
+
+import com.vitorpamplona.nestsclient.NestsAuth
+import com.vitorpamplona.nestsclient.NestsException
+import com.vitorpamplona.nestsclient.NestsRoomConfig
+import com.vitorpamplona.nestsclient.OkHttpNestsClient
+import com.vitorpamplona.nestsclient.nestsAuthUrl
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
+import kotlinx.coroutines.runBlocking
+import okhttp3.MediaType.Companion.toMediaType
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.RequestBody.Companion.toRequestBody
+import org.junit.AfterClass
+import org.junit.BeforeClass
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+import kotlin.test.fail
+
+/**
+ * Phase-4 negative-auth interop tests. Locks in the wire-level rejection
+ * paths exposed by the moq-auth sidecar so a regression in our request
+ * shape (header format, NIP-98 signing, body schema, namespace regex)
+ * surfaces as a fast unit-of-work failure rather than a confused
+ * production stack trace.
+ *
+ * Mirrors the audit findings in moq-auth `index.ts` — rejection responses
+ * are JSON `{"error": "..."}` with HTTP 400 / 401 / 429 depending on the
+ * failure mode. We assert HTTP status only (the human-readable error text
+ * isn't part of the contract), but include the message in the failure
+ * message for debuggability.
+ *
+ * Skipped by default — set `-DnestsInterop=true` to enable.
+ */
+class NostrNestsAuthFailureInteropTest {
+    @Test
+    fun missing_authorization_header_is_rejected_401() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            val url = nestsAuthUrl(harness.authBaseUrl)
+            val body = """{"namespace":"${validNamespace()}","publish":false}"""
+            assertStatus(
+                expected = 401,
+                description = "missing Authorization header",
+                request =
+                    Request
+                        .Builder()
+                        .url(url)
+                        .post(body.toRequestBody(JSON))
+                        .build(),
+            )
+        }
+
+    @Test
+    fun authorization_header_with_wrong_scheme_is_rejected_401() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            val url = nestsAuthUrl(harness.authBaseUrl)
+            val body = """{"namespace":"${validNamespace()}","publish":false}"""
+            assertStatus(
+                expected = 401,
+                description = "Authorization header with non-Nostr scheme",
+                request =
+                    Request
+                        .Builder()
+                        .url(url)
+                        .header("Authorization", "Bearer not-a-nostr-event")
+                        .post(body.toRequestBody(JSON))
+                        .build(),
+            )
+        }
+
+    @Test
+    fun nip98_event_signed_for_a_different_url_is_rejected_401() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            val signer = NostrSignerInternal(KeyPair())
+            val realUrl = nestsAuthUrl(harness.authBaseUrl)
+            val body = """{"namespace":"${validNamespace()}","publish":false}"""
+            val bodyBytes = body.encodeToByteArray()
+
+            // NIP-98 binds the signed event to a `u` tag that the server
+            // re-checks against the request URL. Sign for a *different*
+            // URL so the host/path mismatch fires.
+            val authHeader =
+                NestsAuth.header(
+                    signer = signer,
+                    url = "https://other.example.test/auth",
+                    method = "POST",
+                    payload = bodyBytes,
+                )
+
+            assertStatus(
+                expected = 401,
+                description = "NIP-98 event signed for the wrong URL",
+                request =
+                    Request
+                        .Builder()
+                        .url(realUrl)
+                        .header("Authorization", authHeader)
+                        .post(body.toRequestBody(JSON))
+                        .build(),
+            )
+        }
+
+    @Test
+    fun malformed_namespace_is_rejected_400() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            // Per moq-auth's regex /^nests\/\d+:[0-9a-f]{64}:[a-zA-Z0-9._-]+$/
+            // — uppercase pubkey hex doesn't match the `[0-9a-f]` class.
+            val signer = NostrSignerInternal(KeyPair())
+            val client = OkHttpNestsClient(http = http)
+            val room =
+                NestsRoomConfig(
+                    authBaseUrl = harness.authBaseUrl,
+                    endpoint = harness.moqEndpoint,
+                    hostPubkey = "A".repeat(64),
+                    roomId = "regex-fuzz",
+                )
+            try {
+                client.mintToken(room = room, publish = false, signer = signer)
+                fail("expected NestsException for invalid namespace, none thrown")
+            } catch (e: NestsException) {
+                assertEquals(
+                    400,
+                    e.status,
+                    "expected 400 for namespace regex mismatch — got status=${e.status} message='${e.message}'",
+                )
+            }
+        }
+
+    @Test
+    fun publish_true_grants_a_token_for_any_caller() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            // The sidecar does NOT gate publish-vs-audience by NIP-53
+            // hostlist; any user requesting `publish: true` gets a JWT
+            // with `put: [<their-pubkey>]`. Lock that in so a future
+            // hostlist gate doesn't sneak in unnoticed.
+            val signer = NostrSignerInternal(KeyPair())
+            val client = OkHttpNestsClient(http = http)
+            val room =
+                NestsRoomConfig(
+                    authBaseUrl = harness.authBaseUrl,
+                    endpoint = harness.moqEndpoint,
+                    hostPubkey = signer.pubKey,
+                    roomId = "publish-grant-${System.currentTimeMillis()}",
+                )
+            val token = client.mintToken(room = room, publish = true, signer = signer)
+            assertTrue(token.count { it == '.' } == 2, "expected JWT with 3 segments, got: $token")
+        }
+
+    companion object {
+        private val JSON = "application/json".toMediaType()
+        private val http = OkHttpClient()
+
+        private var harnessOrNull: NostrNestsHarness? = null
+
+        private fun validNamespace(): String = "nests/30312:" + "0".repeat(64) + ":sample-room"
+
+        /**
+         * Issue [request] and assert the response status equals [expected].
+         * Reads the body once for the failure message — okhttp's `Response`
+         * body is single-shot, so we have to materialise the string before
+         * the `.use {}` block exits.
+         */
+        private fun assertStatus(
+            expected: Int,
+            description: String,
+            request: Request,
+        ) {
+            http.newCall(request).execute().use { response ->
+                val code = response.code
+                val bodyText = response.body.string()
+                assertEquals(
+                    expected,
+                    code,
+                    "$description — server returned $code with body: $bodyText",
+                )
+            }
+        }
+
+        @BeforeClass
+        @JvmStatic
+        fun setUpHarness() {
+            if (NostrNestsHarness.isEnabled()) {
+                harnessOrNull = NostrNestsHarness.start()
+            }
+        }
+
+        @AfterClass
+        @JvmStatic
+        fun tearDownHarness() {
+            harnessOrNull?.close()
+            harnessOrNull = null
+        }
+    }
+}
@@ -106,10 +106,12 @@ class NostrNestsHarness private constructor(
            return NostrNestsHarness(
                workDir = workDir,
                authBaseUrl = "http://127.0.0.1:$AUTH_HOST_PORT",
-                // The moq-relay terminates TLS — we use a permissive
-                // certificate validator on the QuicWebTransportFactory in
-                // tests. Path is `/anon` per the relay's default WT route.
-                moqEndpoint = "https://127.0.0.1:$MOQ_HOST_PORT/anon",
+                // moq-rs terminates TLS with a self-signed cert in dev —
+                // production tests pair this with PermissiveCertificateValidator.
+                // The path under this base is the namespace literal (per
+                // moq-rs `claims.root` matching); the `:nestsClient` connect
+                // helpers append `/<namespace>?jwt=<token>` themselves.
+                moqEndpoint = "https://127.0.0.1:$MOQ_HOST_PORT/",
            )
        }

@@ -0,0 +1,406 @@
+/*
+ * 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.interop
+
+import com.vitorpamplona.nestsclient.NestsListener
+import com.vitorpamplona.nestsclient.NestsListenerState
+import com.vitorpamplona.nestsclient.NestsRoomConfig
+import com.vitorpamplona.nestsclient.NestsSpeaker
+import com.vitorpamplona.nestsclient.NestsSpeakerState
+import com.vitorpamplona.nestsclient.OkHttpNestsClient
+import com.vitorpamplona.nestsclient.audio.AudioCapture
+import com.vitorpamplona.nestsclient.audio.OpusEncoder
+import com.vitorpamplona.nestsclient.connectNestsListener
+import com.vitorpamplona.nestsclient.connectNestsSpeaker
+import com.vitorpamplona.nestsclient.moq.MoqObject
+import com.vitorpamplona.nestsclient.moq.SubscribeHandle
+import com.vitorpamplona.nestsclient.transport.QuicWebTransportFactory
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
+import com.vitorpamplona.quic.tls.PermissiveCertificateValidator
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
+import org.junit.AfterClass
+import org.junit.BeforeClass
+import org.junit.Test
+import kotlin.test.assertContentEquals
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+/**
+ * Phase-4 multi-peer interop tests. Each test brings up a real
+ * nostrnests stack, exercises a multi-publisher / multi-subscriber
+ * scenario through the production [connectNestsSpeaker] +
+ * [connectNestsListener] entry points, and asserts the relay's fan-out
+ * + ordering guarantees end-to-end.
+ *
+ * Coverage:
+ *   - **Multi-listener fan-out** — 1 speaker, 2 listeners; each listener
+ *     receives the full frame stream.
+ *   - **Multi-speaker** — 2 speakers under the same room; one listener
+ *     subscribes to each by pubkey hex; both subscriptions deliver
+ *     their respective speaker's frames with no cross-talk.
+ *   - **Subscribe-before-announce** — listener subscribes before the
+ *     speaker has announced; per moq-rs the relay holds the subscribe
+ *     and resolves it once a publisher arrives.
+ *
+ * Skipped by default — set `-DnestsInterop=true` to enable.
+ */
+class NostrNestsMultiPeerInteropTest {
+    @Test
+    fun one_speaker_fans_out_to_two_listeners() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            val room = freshRoom(harness, hostPubkey = "0".repeat(64))
+            val supervisor = SupervisorJob()
+            val pumpScope = CoroutineScope(supervisor + Dispatchers.IO)
+
+            val speakerSigner = NostrSignerInternal(KeyPair())
+            val speakerRoom = room.copy(hostPubkey = speakerSigner.pubKey)
+            val capture = DriverCapture()
+
+            val listenerSignerA = NostrSignerInternal(KeyPair())
+            val listenerSignerB = NostrSignerInternal(KeyPair())
+
+            try {
+                val speaker =
+                    connectSpeaker(
+                        scope = pumpScope,
+                        room = speakerRoom,
+                        signer = speakerSigner,
+                        capture = capture,
+                    )
+                speaker.startBroadcasting()
+
+                val listenerA = connectListener(pumpScope, speakerRoom, listenerSignerA)
+                val listenerB = connectListener(pumpScope, speakerRoom, listenerSignerB)
+
+                val subA = listenerA.subscribeSpeaker(speakerSigner.pubKey)
+                val subB = listenerB.subscribeSpeaker(speakerSigner.pubKey)
+
+                val collectedA = collectFrames(pumpScope, subA, FRAMES_FANOUT)
+                val collectedB = collectFrames(pumpScope, subB, FRAMES_FANOUT)
+
+                delay(SUBSCRIBE_SETTLE_MS)
+                for (i in 0 until FRAMES_FANOUT) {
+                    capture.push(shortArrayOf(i.toShort()))
+                    delay(FRAME_SPACING_MS)
+                }
+
+                assertFrameSequence(collectedA.await(), FRAMES_FANOUT, "listener A")
+                assertFrameSequence(collectedB.await(), FRAMES_FANOUT, "listener B")
+
+                runCatching { subA.unsubscribe() }
+                runCatching { subB.unsubscribe() }
+                runCatching { listenerA.close() }
+                runCatching { listenerB.close() }
+                runCatching { speaker.close() }
+            } finally {
+                capture.stop()
+                supervisor.cancelAndJoin()
+            }
+            Unit
+        }
+
+    @Test
+    fun two_speakers_in_same_room_deliver_independently_to_one_listener() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            // Two speakers sharing the same room namespace must publish
+            // under different sub-paths (relay enforces `put: [pubkey]`).
+            // The room's hostPubkey gates `claims.root` — we use signer A's
+            // pubkey as the hostPubkey for the namespace, but each speaker
+            // mints their own JWT bound to their own pubkey via `put`.
+            val signerA = NostrSignerInternal(KeyPair())
+            val signerB = NostrSignerInternal(KeyPair())
+            val listenerSigner = NostrSignerInternal(KeyPair())
+            val room = freshRoom(harness, hostPubkey = signerA.pubKey)
+
+            val supervisor = SupervisorJob()
+            val pumpScope = CoroutineScope(supervisor + Dispatchers.IO)
+
+            val captureA = DriverCapture()
+            val captureB = DriverCapture()
+
+            try {
+                val speakerA =
+                    connectSpeaker(pumpScope, room, signerA, captureA, encoderPrefix = "A:")
+                val speakerB =
+                    connectSpeaker(pumpScope, room, signerB, captureB, encoderPrefix = "B:")
+                speakerA.startBroadcasting()
+                speakerB.startBroadcasting()
+
+                val listener = connectListener(pumpScope, room, listenerSigner)
+                val subA = listener.subscribeSpeaker(signerA.pubKey)
+                val subB = listener.subscribeSpeaker(signerB.pubKey)
+
+                val collectedA = collectFrames(pumpScope, subA, FRAMES_MULTI_SPEAKER)
+                val collectedB = collectFrames(pumpScope, subB, FRAMES_MULTI_SPEAKER)
+
+                delay(SUBSCRIBE_SETTLE_MS)
+                for (i in 0 until FRAMES_MULTI_SPEAKER) {
+                    captureA.push(shortArrayOf((i + 0).toShort()))
+                    captureB.push(shortArrayOf((i + 100).toShort()))
+                    delay(FRAME_SPACING_MS)
+                }
+
+                val resA = collectedA.await()
+                val resB = collectedB.await()
+                assertNotNull(resA, "listener never received speaker-A frames")
+                assertNotNull(resB, "listener never received speaker-B frames")
+                assertEquals(FRAMES_MULTI_SPEAKER, resA.size, "speaker A count")
+                assertEquals(FRAMES_MULTI_SPEAKER, resB.size, "speaker B count")
+
+                resA.forEachIndexed { idx, obj ->
+                    assertContentEquals(
+                        "A:".encodeToByteArray() + byteArrayOf(idx.toByte()),
+                        obj.payload,
+                        "speaker A frame $idx — must not contain B's payload (no cross-talk)",
+                    )
+                }
+                resB.forEachIndexed { idx, obj ->
+                    assertContentEquals(
+                        "B:".encodeToByteArray() + byteArrayOf((idx + 100).toByte()),
+                        obj.payload,
+                        "speaker B frame $idx — must not contain A's payload (no cross-talk)",
+                    )
+                }
+
+                runCatching { subA.unsubscribe() }
+                runCatching { subB.unsubscribe() }
+                runCatching { listener.close() }
+                runCatching { speakerA.close() }
+                runCatching { speakerB.close() }
+            } finally {
+                captureA.stop()
+                captureB.stop()
+                supervisor.cancelAndJoin()
+            }
+            Unit
+        }
+
+    @Test
+    fun listener_subscribed_before_announce_receives_late_frames() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            // The relay holds the SUBSCRIBE open until a publisher
+            // announces. Validates moq-rs's "subscribe before announce"
+            // contract end-to-end through our client.
+            val speakerSigner = NostrSignerInternal(KeyPair())
+            val listenerSigner = NostrSignerInternal(KeyPair())
+            val room = freshRoom(harness, hostPubkey = speakerSigner.pubKey)
+
+            val supervisor = SupervisorJob()
+            val pumpScope = CoroutineScope(supervisor + Dispatchers.IO)
+            val capture = DriverCapture()
+
+            try {
+                val listener = connectListener(pumpScope, room, listenerSigner)
+                val sub = listener.subscribeSpeaker(speakerSigner.pubKey)
+                val collected = collectFrames(pumpScope, sub, FRAMES_PRESUB)
+
+                // Wait briefly so the SUBSCRIBE is on the relay before
+                // the publisher arrives — the relay should hold it.
+                delay(SUBSCRIBE_SETTLE_MS)
+
+                val speaker = connectSpeaker(pumpScope, room, speakerSigner, capture)
+                speaker.startBroadcasting()
+
+                // Give the announce + publisher-side subscribe matchup
+                // time to plumb before pushing frames.
+                delay(SUBSCRIBE_SETTLE_MS)
+                for (i in 0 until FRAMES_PRESUB) {
+                    capture.push(shortArrayOf(i.toShort()))
+                    delay(FRAME_SPACING_MS)
+                }
+
+                assertFrameSequence(collected.await(), FRAMES_PRESUB, "pre-subscribed listener")
+
+                runCatching { sub.unsubscribe() }
+                runCatching { listener.close() }
+                runCatching { speaker.close() }
+            } finally {
+                capture.stop()
+                supervisor.cancelAndJoin()
+            }
+            Unit
+        }
+
+    // ---------- Helpers ----------
+
+    private suspend fun connectSpeaker(
+        scope: CoroutineScope,
+        room: NestsRoomConfig,
+        signer: NostrSignerInternal,
+        capture: AudioCapture,
+        encoderPrefix: String = "FRAME-",
+    ): NestsSpeaker {
+        val speaker =
+            connectNestsSpeaker(
+                httpClient = http,
+                transport = transport,
+                scope = scope,
+                room = room,
+                signer = signer,
+                speakerPubkeyHex = signer.pubKey,
+                captureFactory = { capture },
+                encoderFactory = { StubEncoder(encoderPrefix.encodeToByteArray()) },
+            )
+        assertTrue(
+            speaker.state.value is NestsSpeakerState.Connected,
+            "speaker did not reach Connected — was ${speaker.state.value}",
+        )
+        return speaker
+    }
+
+    private suspend fun connectListener(
+        scope: CoroutineScope,
+        room: NestsRoomConfig,
+        signer: NostrSignerInternal,
+    ): NestsListener {
+        val listener =
+            connectNestsListener(
+                httpClient = http,
+                transport = transport,
+                scope = scope,
+                room = room,
+                signer = signer,
+            )
+        assertTrue(
+            listener.state.value is NestsListenerState.Connected,
+            "listener did not reach Connected — was ${listener.state.value}",
+        )
+        return listener
+    }
+
+    private fun collectFrames(
+        scope: CoroutineScope,
+        sub: SubscribeHandle,
+        count: Int,
+    ) = scope.async {
+        withTimeoutOrNull(RECEIVE_TIMEOUT_MS) {
+            sub.objects.take(count).toList()
+        }
+    }
+
+    private fun assertFrameSequence(
+        objs: List<MoqObject>?,
+        expectedCount: Int,
+        who: String,
+    ) {
+        assertNotNull(objs, "$who did not receive $expectedCount frames within ${RECEIVE_TIMEOUT_MS}ms")
+        assertEquals(expectedCount, objs.size, "$who frame count")
+        objs.forEachIndexed { idx, obj ->
+            assertEquals(idx.toLong(), obj.objectId, "$who object id at index $idx")
+            assertContentEquals(
+                "FRAME-".encodeToByteArray() + byteArrayOf(idx.toByte()),
+                obj.payload,
+                "$who payload at index $idx",
+            )
+        }
+    }
+
+    private class DriverCapture : AudioCapture {
+        private val frames = Channel<ShortArray>(capacity = Channel.UNLIMITED)
+
+        @Volatile private var started: Boolean = false
+
+        override fun start() {
+            started = true
+        }
+
+        override suspend fun readFrame(): ShortArray? {
+            if (!started) return null
+            return frames.receiveCatching().getOrNull()
+        }
+
+        override fun stop() {
+            frames.close()
+        }
+
+        fun push(pcm: ShortArray) {
+            frames.trySend(pcm)
+        }
+    }
+
+    private class StubEncoder(
+        private val prefix: ByteArray,
+    ) : OpusEncoder {
+        override fun encode(pcm: ShortArray): ByteArray = prefix + byteArrayOf(pcm.first().toByte())
+
+        override fun release() = Unit
+    }
+
+    companion object {
+        private const val FRAMES_FANOUT = 6
+        private const val FRAMES_MULTI_SPEAKER = 6
+        private const val FRAMES_PRESUB = 4
+        private const val SUBSCRIBE_SETTLE_MS = 500L
+        private const val FRAME_SPACING_MS = 25L
+        private const val RECEIVE_TIMEOUT_MS = 15_000L
+
+        private val http = OkHttpNestsClient()
+        private val transport =
+            QuicWebTransportFactory(certificateValidator = PermissiveCertificateValidator())
+        private var harnessOrNull: NostrNestsHarness? = null
+
+        private fun freshRoom(
+            harness: NostrNestsHarness,
+            hostPubkey: String,
+        ) = NestsRoomConfig(
+            authBaseUrl = harness.authBaseUrl,
+            endpoint = harness.moqEndpoint,
+            hostPubkey = hostPubkey,
+            roomId = "mp-${System.nanoTime()}",
+        )
+
+        @BeforeClass
+        @JvmStatic
+        fun setUpHarness() {
+            if (NostrNestsHarness.isEnabled()) {
+                harnessOrNull = NostrNestsHarness.start()
+            }
+        }
+
+        @AfterClass
+        @JvmStatic
+        fun tearDownHarness() {
+            harnessOrNull?.close()
+            harnessOrNull = null
+        }
+    }
+}