test(audio-rooms) + fix: round-trip test + audit pass

Adds an end-to-end MoQ round-trip test and lands the highest-severity findings from a 3-agent audit (protocol / ViewModel / Android lifecycle) of the M5–M7 + Activity work. Round-trip test (`:nestsClient` MoqRoundTripTest): - Two MoqSession.client() instances (publisher + subscriber) talk through a hand-rolled in-test relay coroutine that mirrors a real MoQ relay's wire behavior (forwards CLIENT_SETUP / SERVER_SETUP / ANNOUNCE / SUBSCRIBE / SUBSCRIBE_OK / OBJECT_DATAGRAM). - 100-Opus-frame test exercises the full publisher → wire → subscriber path, asserting payload bytes + monotonic group/object ids round-trip correctly. Catches any drift between our publisher and our own subscriber's wire format. - Second test verifies SUBSCRIBE_ERROR(TRACK_DOES_NOT_EXIST) flows back as MoqProtocolException when the publisher hasn't openTrack'd. MoQ protocol fixes (CRITICAL audit findings): - SubscribeDoneStatus constants were inverted: had UNSUBSCRIBED=0x01 (peer reads as INTERNAL_ERROR) and TRACK_ENDED=0x00 (peer reads as UNSUBSCRIBED). Swapped to draft-stable values: UNSUBSCRIBED=0x00, TRACK_ENDED=0x03. - Pending CompletableDeferreds for in-flight SUBSCRIBE / ANNOUNCE on session close were `cancel()`-ed, which propagates as CancellationException — caller's entire scope cancels instead of catching a domain MoqProtocolException. Switched all sites to `completeExceptionally(MoqProtocolException("session closed"))` including unsubscribe-while-pending-OK. ViewModel fixes: - openSubscription race: re-check `activeSubscriptions[pubkey] === slot && !closed` AFTER the suspending `subscribeSpeaker` returns; if the user removed the speaker mid-flight, fire-and-forget UNSUBSCRIBE rather than attaching a leaked SubscribeHandle + AudioRoomPlayer to a discarded slot. - Server-initiated `Closed` listener state now triggers `teardown(targetState=Closed)` and resets the auto-retry counter, so a transport-died-mid-handshake doesn't leave stale subscriptions in the VM map until the Activity finishes. - Cleanup-scope split: `disconnect()` (user-driven) routes the close through `viewModelScope` (still alive); `onCleared()` routes it through a process-lived `cleanupScope` (default GlobalScope, tests pass backgroundScope) so MoQ control frames (UNSUBSCRIBE, UNANNOUNCE, SUBSCRIBE_DONE) actually land before the QUIC transport drops, instead of being eaten by the cancelled viewModelScope. Android lifecycle fixes: - AudioRoomBridge.clear() wired into AccountViewModel.onCleared next to the existing CallSessionBridge.clear() — no more cross-account AccountViewModel leak after logout/switch. - registerReceiver flag now gated on Build.VERSION.SDK_INT TIRAMISU+ — RECEIVER_NOT_EXPORTED on pre-33 devices collides with RECEIVER_VISIBLE_TO_INSTANT_APPS. PendingIntents stay package-scoped via setPackage(packageName). - onUserLeaveHint no longer enters PIP unconditionally: gated on ui.connection == Connected AND PackageManager FEATURE_PICTURE_IN_PICTURE present, so PIP-from-lobby / PIP-on-incompatible-device doesn't leave a frozen full-screen card in Recents. - Replaced the process-wide singleton AudioRoomPipActions toggle Boolean with a per-Activity MutableSharedFlow<Unit> so a stale emission from a torn-down Activity can't leak into a new one. ViewModel test was extended with `cleanupScope = backgroundScope` plumbing so the post-disconnect `listener.close()` assertion remains observable in the test scheduler. Verified: spotlessApply clean; :commons:jvmTest (9 tests), :nestsClient:jvmTest (80 tests including 2 new round-trip), and :amethyst:compilePlayDebugKotlin all green. Audit findings deferred to follow-up commits (none ship-blocking): - Wire layout pinning vs draft-17 vs draft-11 (currently emits a draft-11-style OBJECT_DATAGRAM; we advertise draft-17). Resolve via the M4 manual interop pass against nostrnests. - UNANNOUNCE racing inbound SUBSCRIBE; control-pump cancel half-write (audit MoQ #5, #6) — small ordering tweaks. - Two retry coroutines stacking under fast Failed bursts (audit VM #4). - AudioRoomForegroundService startForeground always-first contract (audit Android #8).
2026-04-26 09:05:00 +00:00
parent 3816e471b7
commit f0b27654ba
7 changed files with 354 additions and 56 deletions
@@ -1574,6 +1574,8 @@ class AccountViewModel(
        callManager.dispose()
        com.vitorpamplona.amethyst.service.call.CallSessionBridge
            .clear()
+        com.vitorpamplona.amethyst.ui.screen.loggedIn.audiorooms.room.AudioRoomBridge
+            .clear()
        feedStates.destroy()
        super.onCleared()
    }
@@ -28,6 +28,7 @@ import android.content.Context
 import android.content.Intent
 import android.content.IntentFilter
 import android.graphics.drawable.Icon
+import android.os.Build
 import android.os.Bundle
 import android.util.Rational
 import androidx.activity.compose.setContent
@@ -35,6 +36,9 @@ import androidx.appcompat.app.AppCompatActivity
 import androidx.compose.runtime.mutableStateOf
 import com.vitorpamplona.amethyst.R
 import com.vitorpamplona.amethyst.ui.theme.AmethystTheme
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.asSharedFlow

 /**
 * Standalone activity that owns the lifetime of an audio-room session. The
@@ -54,6 +58,7 @@ import com.vitorpamplona.amethyst.ui.theme.AmethystTheme
 class AudioRoomActivity : AppCompatActivity() {
    private val isInPipMode = mutableStateOf(false)
    private val isMuted = mutableStateOf(false)
+    private val isConnected = mutableStateOf(false)

    private val pipReceiver =
        object : BroadcastReceiver() {
@@ -63,10 +68,11 @@ class AudioRoomActivity : AppCompatActivity() {
            ) {
                when (intent.action) {
                    ACTION_PIP_TOGGLE_MUTE -> {
-                        // The composable owns the VM; the easiest signal is a
-                        // shared static the content layer will pick up via
-                        // its mute state observer.
-                        AudioRoomPipActions.toggleMuteRequested.value = !AudioRoomPipActions.toggleMuteRequested.value
+                        // Per-Activity SharedFlow — observed by the composable
+                        // and forwarded to the VM. Replaces an earlier
+                        // process-wide singleton that could leak edges across
+                        // activity instances.
+                        toggleMuteSignal.tryEmit(Unit)
                    }

                    ACTION_PIP_LEAVE -> {
@@ -76,6 +82,12 @@ class AudioRoomActivity : AppCompatActivity() {
            }
        }

+    private val _toggleMuteSignal =
+        MutableSharedFlow<Unit>(replay = 0, extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
+
+    /** Emitted when the PIP-overlay mute action is tapped. */
+    val toggleMuteSignal: MutableSharedFlow<Unit> get() = _toggleMuteSignal
+
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)

@@ -88,14 +100,21 @@ class AudioRoomActivity : AppCompatActivity() {
            return
        }

-        registerReceiver(
-            pipReceiver,
+        val filter =
            IntentFilter().apply {
                addAction(ACTION_PIP_TOGGLE_MUTE)
                addAction(ACTION_PIP_LEAVE)
-            },
-            RECEIVER_NOT_EXPORTED,
-        )
+            }
+        // RECEIVER_NOT_EXPORTED requires API 33+. The PendingIntents that
+        // fire these actions all use setPackage(packageName) so the
+        // visibility risk on older devices is bounded; pre-33 we register
+        // without the flag.
+        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.TIRAMISU) {
+            registerReceiver(pipReceiver, filter, RECEIVER_NOT_EXPORTED)
+        } else {
+            @Suppress("UnspecifiedRegisterReceiverFlag")
+            registerReceiver(pipReceiver, filter)
+        }

        setContent {
            AmethystTheme {
@@ -111,6 +130,8 @@ class AudioRoomActivity : AppCompatActivity() {
                            updatePipParams()
                        }
                    },
+                    onConnectedChange = { isConnected.value = it },
+                    pipMuteSignal = toggleMuteSignal.asSharedFlow(),
                    onLeave = { finish() },
                )
            }
@@ -119,6 +140,11 @@ class AudioRoomActivity : AppCompatActivity() {

    override fun onUserLeaveHint() {
        super.onUserLeaveHint()
+        // PIP only makes sense once the audio session is live; entering
+        // PIP from the lobby/Connecting state would freeze a half-rendered
+        // card in Recents. Also gates on the system supporting PIP.
+        if (!isConnected.value) return
+        if (!packageManager.hasSystemFeature(android.content.pm.PackageManager.FEATURE_PICTURE_IN_PICTURE)) return
        runCatching { enterPictureInPictureMode(buildPipParams()) }
    }

@@ -198,16 +224,3 @@ class AudioRoomActivity : AppCompatActivity() {
        }
    }
 }
-
-/**
- * Cross-talk between the PIP system buttons (which arrive as broadcast
- * intents in the Activity) and the composable that owns the VM. The
- * Activity flips `toggleMuteRequested`; a `LaunchedEffect` in the
- * composable observes it and forwards the toggle to the VM.
- *
- * Lives at file scope so the composable can see it without cross-package
- * visibility shenanigans.
- */
-internal object AudioRoomPipActions {
-    val toggleMuteRequested = mutableStateOf(false)
-}
@@ -44,6 +44,7 @@ import com.vitorpamplona.quartz.nip53LiveActivities.presence.MeetingRoomPresence
 import com.vitorpamplona.quartz.nip53LiveActivities.streaming.tags.ROLE
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch

@@ -64,6 +65,8 @@ internal fun AudioRoomActivityContent(
    accountViewModel: AccountViewModel,
    isInPipMode: Boolean,
    onMuteState: (Boolean) -> Unit,
+    onConnectedChange: (Boolean) -> Unit,
+    pipMuteSignal: SharedFlow<Unit>,
    onLeave: () -> Unit,
 ) {
    val parsedAddress = remember(addressValue) { Address.parse(addressValue) }
@@ -84,6 +87,8 @@ internal fun AudioRoomActivityContent(
            accountViewModel = accountViewModel,
            isInPipMode = isInPipMode,
            onMuteState = onMuteState,
+            onConnectedChange = onConnectedChange,
+            pipMuteSignal = pipMuteSignal,
            onLeave = onLeave,
        )
    }
@@ -97,6 +102,8 @@ private fun AudioRoomActivityBody(
    accountViewModel: AccountViewModel,
    isInPipMode: Boolean,
    onMuteState: (Boolean) -> Unit,
+    onConnectedChange: (Boolean) -> Unit,
+    pipMuteSignal: SharedFlow<Unit>,
    onLeave: () -> Unit,
 ) {
    val participants = remember(event) { event.participants() }
@@ -136,18 +143,18 @@ private fun AudioRoomActivityBody(

    val ui by viewModel.uiState.collectAsState()

-    // Push mute state into the activity so it can rebuild the PIP RemoteAction.
+    // Push mute + connected state up so the Activity can rebuild PIP
+    // RemoteActions and gate `enterPictureInPictureMode` on Connected.
    LaunchedEffect(ui.isMuted) { onMuteState(ui.isMuted) }
+    val isConnectedNow = ui.connection is ConnectionUiState.Connected
+    LaunchedEffect(isConnectedNow) { onConnectedChange(isConnectedNow) }

-    // The PIP mute system action signals via a shared toggle flag whose
-    // value flips on every tap. Observe the changes (skipping the initial
-    // composition) and forward to the VM.
-    val toggleSignal by AudioRoomPipActions.toggleMuteRequested
-    var lastSeenToggle by remember { mutableStateOf(toggleSignal) }
-    LaunchedEffect(toggleSignal) {
-        if (toggleSignal != lastSeenToggle) {
-            lastSeenToggle = toggleSignal
-            viewModel.setMuted(!ui.isMuted)
+    // Forward PIP-overlay mute taps into the VM. Per-Activity SharedFlow
+    // so a stale emission from a previous Activity instance can't leak
+    // into a new one.
+    LaunchedEffect(viewModel) {
+        pipMuteSignal.collect {
+            viewModel.setMuted(!viewModel.uiState.value.isMuted)
        }
    }

@@ -44,6 +44,7 @@ import kotlinx.collections.immutable.ImmutableSet
 import kotlinx.collections.immutable.persistentSetOf
 import kotlinx.collections.immutable.toPersistentSet
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -96,6 +97,13 @@ class AudioRoomViewModel(
    // listener whose state they can drive directly.
    private val connector: NestsListenerConnector = DefaultNestsListenerConnector,
    private val speakerConnector: NestsSpeakerConnector = DefaultNestsSpeakerConnector,
+    // Scope used for fire-and-forget MoQ cleanup (UNANNOUNCE,
+    // SUBSCRIBE_DONE, MoQ session close) that needs to outlive the VM's
+    // own scope. Production passes [GlobalScope] so onCleared can finish
+    // sending control frames before the QUIC transport drops; tests pass
+    // their `backgroundScope` so assertions can observe the close.
+    @OptIn(kotlinx.coroutines.DelicateCoroutinesApi::class)
+    private val cleanupScope: CoroutineScope = GlobalScope,
 ) : ViewModel() {
    private val _uiState = MutableStateFlow(AudioRoomUiState())
    val uiState: StateFlow<AudioRoomUiState> = _uiState.asStateFlow()
@@ -278,16 +286,16 @@ class AudioRoomViewModel(
        autoRetryJob?.cancel()
        autoRetryJob = null
        autoRetryAttempts = 0
-        teardownBroadcast(BroadcastUiState.Idle)
-        teardown(targetState = ConnectionUiState.Idle)
+        teardownBroadcast(BroadcastUiState.Idle, finalCleanup = false)
+        teardown(targetState = ConnectionUiState.Idle, finalCleanup = false)
    }

    override fun onCleared() {
        closed = true
        autoRetryJob?.cancel()
        autoRetryJob = null
-        teardownBroadcast(BroadcastUiState.Idle)
-        teardown(targetState = ConnectionUiState.Closed)
+        teardownBroadcast(BroadcastUiState.Idle, finalCleanup = true)
+        teardown(targetState = ConnectionUiState.Closed, finalCleanup = true)
        super.onCleared()
    }

@@ -313,7 +321,10 @@ class AudioRoomViewModel(
            }
    }

-    private fun teardownBroadcast(targetState: BroadcastUiState) {
+    private fun teardownBroadcast(
+        targetState: BroadcastUiState,
+        finalCleanup: Boolean = false,
+    ) {
        speakerConnectJob?.cancel()
        speakerConnectJob = null
        speakerStateJob?.cancel()
@@ -323,7 +334,12 @@ class AudioRoomViewModel(
        broadcastHandle = null
        speaker = null
        if (handle != null || s != null) {
-            viewModelScope.launch {
+            // User-driven disconnect can use viewModelScope (still alive);
+            // onCleared must use cleanupScope because viewModelScope is
+            // already cancelled and a launch on it would no-op without
+            // emitting the UNANNOUNCE / SUBSCRIBE_DONE wire frames.
+            val scope = if (finalCleanup) cleanupScope else viewModelScope
+            scope.launch {
                handle?.runCatching { close() }
                s?.runCatching { close() }
            }
@@ -349,6 +365,14 @@ class AudioRoomViewModel(
                            scheduleAutoRetry()
                        }

+                        // Server-initiated Closed: reset the retry counter and
+                        // tear down stale local state so any later user-driven
+                        // reconnect starts fresh.
+                        NestsListenerState.Closed -> {
+                            autoRetryAttempts = 0
+                            teardown(targetState = ConnectionUiState.Closed)
+                        }
+
                        else -> { /* no extra side effect */ }
                    }
                }
@@ -467,6 +491,15 @@ class AudioRoomViewModel(
        if (closed || activeSubscriptions[pubkey] !== slot) return
        try {
            val handle = l.subscribeSpeaker(pubkey)
+            // Re-check after the suspending subscribeSpeaker — the user
+            // may have removed this speaker via updateSpeakers / disconnected
+            // while the SUBSCRIBE was in flight. If so, abandon the handle
+            // cleanly (fire-and-forget UNSUBSCRIBE) instead of attaching it
+            // to a slot the reconcile loop has already discarded.
+            if (closed || activeSubscriptions[pubkey] !== slot) {
+                viewModelScope.launch { runCatching { handle.unsubscribe() } }
+                return
+            }
            val decoder = decoderFactory()
            val player = playerFactory()
            val isMuted = _uiState.value.isMuted
@@ -491,7 +524,10 @@ class AudioRoomViewModel(
        }
    }

-    private fun teardown(targetState: ConnectionUiState) {
+    private fun teardown(
+        targetState: ConnectionUiState,
+        finalCleanup: Boolean = false,
+    ) {
        connectJob?.cancel()
        connectJob = null
        stateObserverJob?.cancel()
@@ -505,10 +541,14 @@ class AudioRoomViewModel(
        val l = listener
        listener = null
        if (l != null) {
-            // Closing the listener is suspending; fire-and-forget on the VM
-            // scope is fine — even if the scope is cancelled (onCleared), the
-            // underlying transport's own cleanup runs.
-            viewModelScope.launch { runCatching { l.close() } }
+            // Listener.close() sends MoQ control frames (UNSUBSCRIBE etc.)
+            // before the QUIC transport drops. User-driven disconnect uses
+            // the still-alive viewModelScope; onCleared has already had its
+            // viewModelScope cancelled, so it routes through cleanupScope
+            // (a process-lived scope) so the peer sees a clean teardown
+            // rather than a hard QUIC drop.
+            val scope = if (finalCleanup) cleanupScope else viewModelScope
+            scope.launch { runCatching { l.close() } }
        }
        _uiState.update {
            it.copy(
@@ -40,6 +40,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.resetMain
 import kotlinx.coroutines.test.runTest
@@ -211,7 +212,7 @@ class AudioRoomViewModelTest {
            )
        }

-    private fun newViewModel(connect: suspend (CoroutineScope) -> NestsListener): AudioRoomViewModel =
+    private fun TestScope.newViewModel(connect: suspend (CoroutineScope) -> NestsListener): AudioRoomViewModel =
        AudioRoomViewModel(
            httpClient = NoopNestsClient,
            transport = NoopWebTransportFactory,
@@ -224,6 +225,9 @@ class AudioRoomViewModelTest {
                NestsListenerConnector { _, _, scope, _, _, _ ->
                    connect(scope)
                },
+            // Wire to the test's backgroundScope so close calls run during
+            // the test rather than escaping to the real GlobalScope.
+            cleanupScope = backgroundScope,
        )

    private class FakeNestsListener : NestsListener {
@@ -336,7 +336,9 @@ class MoqSession private constructor(
    suspend fun unsubscribe(subscribeId: Long) {
        val alias =
            stateMutex.withLock {
-                pendingSubscribes.remove(subscribeId)?.cancel()
+                pendingSubscribes.remove(subscribeId)?.completeExceptionally(
+                    MoqProtocolException("unsubscribed before SUBSCRIBE_OK arrived"),
+                )
                aliasBySubscribeId.remove(subscribeId)?.also { sinks.remove(it)?.close() }
            } ?: return

@@ -550,18 +552,20 @@ class MoqSession private constructor(
        stateMutex.withLock {
            if (closed) return
            closed = true
-            // Fail any in-flight subscribe waiters cleanly.
+            // Fail any in-flight subscribe waiters with a domain exception
+            // (not Cancellation — that would propagate as scope cancellation
+            // in the awaiting coroutine instead of a recoverable error).
+            val sessionGone = MoqProtocolException("session closed")
            for ((_, deferred) in pendingSubscribes) {
-                deferred.cancel()
+                deferred.completeExceptionally(sessionGone)
            }
            pendingSubscribes.clear()
            for ((_, ch) in sinks) ch.close()
            sinks.clear()
            aliasBySubscribeId.clear()
-            // Fail in-flight announce waiters and surface an exception, since
-            // the caller is awaiting on .await().
+            // Fail in-flight announce waiters the same way.
            for ((_, deferred) in pendingAnnounces) {
-                deferred.cancel()
+                deferred.completeExceptionally(sessionGone)
            }
            pendingAnnounces.clear()
            announces.values.forEach { it.markSessionClosedLocked() }
@@ -845,13 +849,15 @@ object ErrorCode {
 }

 /**
- * MoQ-transport SUBSCRIBE_DONE status_code values. As above, draft-stable
- * subset; not exhaustive.
+ * MoQ-transport SUBSCRIBE_DONE status_code values. Per draft-ietf-moq-transport
+ * (draft-11+): 0x00 = UNSUBSCRIBED, 0x03 = TRACK_ENDED. The constants
+ * shipped initially had these inverted, which the audit caught — a peer
+ * decoded our "track closed" SUBSCRIBE_DONE as INTERNAL_ERROR.
 */
 object SubscribeDoneStatus {
-    /** The publisher has no more objects coming for this track. */
-    const val TRACK_ENDED: Long = 0x00L
-
    /** Subscriber explicitly UNSUBSCRIBEd; publisher is acknowledging. */
-    const val UNSUBSCRIBED: Long = 0x01L
+    const val UNSUBSCRIBED: Long = 0x00L
+
+    /** The publisher has no more objects coming for this track. */
+    const val TRACK_ENDED: Long = 0x03L
 }
@@ -0,0 +1,226 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.nestsclient.moq
+
+import com.vitorpamplona.nestsclient.transport.FakeWebTransport
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.async
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertContentEquals
+import kotlin.test.assertEquals
+
+/**
+ * End-to-end round-trip tests: an actual `MoqSession.client()` publisher
+ * announces, opens a track, and pushes OBJECT_DATAGRAMs which transit a
+ * minimal in-test relay and arrive at an actual `MoqSession.client()`
+ * subscriber via its `SubscribeHandle.objects` flow.
+ *
+ * The relay role here is a hand-rolled coroutine that mirrors the wire
+ * behavior of a real MoQ relay (e.g. nostrnests):
+ *   - reads CLIENT_SETUP from each side, writes SERVER_SETUP back
+ *   - matches SUBSCRIBE on the subscriber side to ANNOUNCE on the
+ *     publisher side, forwards SUBSCRIBE to the publisher
+ *   - matches SUBSCRIBE_OK from the publisher and forwards to the
+ *     subscriber
+ *   - forwards every OBJECT_DATAGRAM from publisher → subscriber
+ *
+ * It's intentionally not generic — just enough to drive one
+ * (announce, openTrack, send) ↔ (subscribe, collect) round trip per test.
+ *
+ * What this catches that the per-side tests don't: that what our publisher
+ * emits is byte-for-byte what our subscriber expects. If we ever drift on
+ * MoQ message layouts between publisher and listener, this fails.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class MoqRoundTripTest {
+    @Test
+    fun publisher_to_subscriber_round_trip_delivers_objects_in_order() =
+        runTest(UnconfinedTestDispatcher()) {
+            val (pubA, pubB) = FakeWebTransport.pair()
+            val (subA, subB) = FakeWebTransport.pair()
+
+            val ns = TrackNamespace.of("nests", "round-trip-room")
+            val trackName = "alice-pubkey".encodeToByteArray()
+            val publisherSession = MoqSession.client(pubA, backgroundScope)
+            val subscriberSession = MoqSession.client(subA, backgroundScope)
+
+            val relayJob = launch { runRelay(pubB, subB, ns, trackName) }
+
+            publisherSession.setup(listOf(MoqVersion.DRAFT_17))
+            subscriberSession.setup(listOf(MoqVersion.DRAFT_17))
+
+            val announceHandle = publisherSession.announce(ns)
+            val publisher = announceHandle.openTrack(trackName)
+
+            // Subscribe runs on a separate coroutine because it has to stay
+            // alive until SUBSCRIBE_OK comes back through the relay path,
+            // which requires the publisher's openTrack to be present (it is,
+            // by virtue of the line above).
+            val handleDeferred =
+                async {
+                    subscriberSession.subscribe(
+                        namespace = ns,
+                        trackName = trackName,
+                        // Big enough that the test's burst of 100 frames
+                        // never trips DROP_OLDEST regardless of how far the
+                        // consumer falls behind under UnconfinedTestDispatcher.
+                        objectBufferCapacity = 256,
+                    )
+                }
+            val handle = handleDeferred.await()
+
+            // Push N Opus-shaped payloads through the publisher and verify
+            // every one arrives on the subscriber's flow with intact
+            // group/object ids and payload bytes. Drives the publisher loop
+            // on `backgroundScope` so it runs concurrently with the
+            // consumer's `take().toList()` collect.
+            val frameCount = 100
+            backgroundScope.launch {
+                repeat(frameCount) { i -> publisher.send(opusPayload(i)) }
+            }
+
+            val received = handle.objects.take(frameCount).toList()
+
+            assertEquals(frameCount, received.size)
+            received.forEachIndexed { idx, obj ->
+                assertEquals(0L, obj.groupId, "frame $idx groupId")
+                assertEquals(idx.toLong(), obj.objectId, "frame $idx objectId monotonic")
+                assertContentEquals(opusPayload(idx), obj.payload, "frame $idx payload")
+            }
+
+            handle.unsubscribe()
+            publisher.close()
+            announceHandle.unannounce()
+            publisherSession.close()
+            subscriberSession.close()
+            relayJob.cancel()
+        }
+
+    @Test
+    fun round_trip_unknown_namespace_subscribe_fails_with_protocol_exception() =
+        runTest {
+            val (pubA, pubB) = FakeWebTransport.pair()
+            val (subA, subB) = FakeWebTransport.pair()
+            val ns = TrackNamespace.of("nests", "real-room")
+            val trackName = "alice".encodeToByteArray()
+            val pub = MoqSession.client(pubA, backgroundScope)
+            val sub = MoqSession.client(subA, backgroundScope)
+
+            val relayJob = launch { runRelay(pubB, subB, ns, trackName) }
+
+            pub.setup(listOf(MoqVersion.DRAFT_17))
+            sub.setup(listOf(MoqVersion.DRAFT_17))
+
+            // Publisher announces "real-room" but does NOT openTrack — the
+            // inbound SUBSCRIBE on the publisher side will return
+            // SUBSCRIBE_ERROR(TRACK_DOES_NOT_EXIST) because no publisher is
+            // registered for that name.
+            pub.announce(ns)
+
+            val ex =
+                runCatching {
+                    sub.subscribe(namespace = ns, trackName = trackName)
+                }.exceptionOrNull()
+
+            assertEquals(true, ex is MoqProtocolException, "expected MoqProtocolException, got $ex")
+
+            pub.close()
+            sub.close()
+            relayJob.cancel()
+        }
+
+    /**
+     * Minimal MoQ-relay simulator. Mirrors what a nests relay does on the
+     * wire so two of our [MoqSession]s can talk through it.
+     *
+     * Limitations vs a real relay (deliberate, this is a test fixture):
+     *   - One namespace, one track, one subscriber per test.
+     *   - SubscribeId / trackAlias are forwarded 1:1 across the relay (a
+     *     real relay remaps; we don't need to since neither side reuses ids).
+     *   - No SUBSCRIBE_DONE relay — the test calls `unsubscribe()` and
+     *     checks it doesn't blow up; the publisher's UNSUBSCRIBE goes to
+     *     the relay which silently drops it.
+     */
+    private suspend fun runRelay(
+        pubTransport: FakeWebTransport,
+        subTransport: FakeWebTransport,
+        @Suppress("UNUSED_PARAMETER") expectedNs: TrackNamespace,
+        @Suppress("UNUSED_PARAMETER") expectedTrack: ByteArray,
+    ) {
+        val pubControl = pubTransport.peerOpenedBidiStreams().first()
+        val subControl = subTransport.peerOpenedBidiStreams().first()
+
+        // SETUP both sides.
+        val pubCs = MoqCodec.decode(pubControl.incoming().first())!!.message as ClientSetup
+        pubControl.write(MoqCodec.encode(ServerSetup(pubCs.supportedVersions.first())))
+
+        val subCs = MoqCodec.decode(subControl.incoming().first())!!.message as ClientSetup
+        subControl.write(MoqCodec.encode(ServerSetup(subCs.supportedVersions.first())))
+
+        // Read ANNOUNCE from publisher, ack with ANNOUNCE_OK.
+        val announce = MoqCodec.decode(pubControl.incoming().first())!!.message as Announce
+        pubControl.write(MoqCodec.encode(AnnounceOk(announce.namespace)))
+
+        // Read SUBSCRIBE from subscriber, forward to publisher verbatim.
+        val sub = MoqCodec.decode(subControl.incoming().first())!!.message as Subscribe
+        pubControl.write(MoqCodec.encode(sub))
+
+        // Wait for the publisher's reply (SUBSCRIBE_OK or SUBSCRIBE_ERROR)
+        // and forward to the subscriber.
+        val pubReply = MoqCodec.decode(pubControl.incoming().first())!!.message
+        when (pubReply) {
+            is SubscribeOk -> {
+                subControl.write(MoqCodec.encode(pubReply))
+            }
+
+            is SubscribeError -> {
+                subControl.write(MoqCodec.encode(pubReply))
+                return
+            }
+
+            else -> {
+                error("unexpected publisher reply to SUBSCRIBE: ${pubReply.type}")
+            }
+        }
+
+        // Forward every OBJECT_DATAGRAM from publisher → subscriber until
+        // either side cancels the relay coroutine.
+        pubTransport.incomingDatagrams().collect { datagram ->
+            subTransport.sendDatagram(datagram)
+        }
+    }
+
+    private fun opusPayload(seqId: Int): ByteArray {
+        // Mimic an Opus packet shape (~80 bytes, varying content).
+        val size = 80
+        val out = ByteArray(size)
+        for (i in 0 until size) {
+            out[i] = ((seqId xor i).toByte())
+        }
+        return out
+    }
+}