feat(nests): drive Opus decoder + AudioTrack channel count from catalog

Web speakers (kixelated/hang reference) emit stereo Opus when the user's AudioContext picks a stereo input — and our catalog now lets us know that ahead of decoder construction. Previously the decoder + AudioTrack were both hardcoded to mono via AudioFormat.CHANNELS, so a stereo web publisher's frames either threw on the Opus TOC byte mismatch, decoded with downmix artifacts, or output left-channel-only depending on the device's MediaCodec implementation. Wire the catalog-discovered numberOfChannels through: - MediaCodecOpusDecoder accepts `channelCount: Int = 1`. Drives MediaFormat.createAudioFormat's channel count, the OpusHead CSD-0 channel byte, and the per-frame output buffer size (stereo frame = 2× shorts vs mono). Validates 1..2 — RFC 7845 mapping family 0 covers both mono and stereo without an explicit channel mapping table; multichannel needs family 1 which we don't support. - AudioTrackPlayer accepts `channelCount: Int = 1`. Drives the AudioTrack channel mask (CHANNEL_OUT_MONO vs CHANNEL_OUT_STEREO) and the 250 ms wall-clock buffer-size target (stereo doubles bytes per sample). - NestViewModel.openSubscription now starts the catalog fetch BEFORE waiting for it, then awaits `_speakerCatalogs[pubkey]` for CATALOG_AWAIT_TIMEOUT_MS (500 ms) before constructing the decoder + player. With the speaker-side emit-on-subscribe hook in place (3417e44), the catalog frame is on the wire within one round-trip of the SUBSCRIBE_OK, so the wait typically resolves in tens of ms. Falls back to mono if the catalog never arrives — covers legacy publishers that don't emit a catalog and the failure-mode where the relay never forwards the catalog group. - decoderFactory + playerFactory signatures change from `() -> X` to `(channelCount: Int) -> X`. Two call sites updated: NestViewModelFactory (production) and NestViewModelTest's newViewModel helper. Speaker side stays mono-only — our encoder is fixed at AudioFormat.CHANNELS=1 and the catalog we publish declares numberOfChannels=1. This change is exclusively about handling incoming audio from publishers that don't share that constraint. https://claude.ai/code/session_014JfZJHSTvyYYWJbC9VbB47
2026-05-06 17:26:47 +00:00
parent 87b43d4d78
commit 1c47fd2403
5 changed files with 161 additions and 31 deletions
@@ -53,8 +53,8 @@ internal class NestViewModelFactory(
            // `httpClient` slot rather than as the first positional arg.
            httpClient = OkHttpNestsClient(httpClient = Amethyst.instance.roleBasedHttpClientBuilder::okHttpClientForVideo),
            transport = QuicWebTransportFactory(),
-            decoderFactory = { MediaCodecOpusDecoder() },
-            playerFactory = { AudioTrackPlayer() },
+            decoderFactory = { channelCount -> MediaCodecOpusDecoder(channelCount = channelCount) },
+            playerFactory = { channelCount -> AudioTrackPlayer(channelCount = channelCount) },
            signer = signer,
            room = room,
            captureFactory = { AudioRecordCapture() },
@@ -35,6 +35,7 @@ import com.vitorpamplona.nestsclient.NestsSpeaker
 import com.vitorpamplona.nestsclient.NestsSpeakerState
 import com.vitorpamplona.nestsclient.audio.AudioCapture
 import com.vitorpamplona.nestsclient.audio.AudioException
+import com.vitorpamplona.nestsclient.audio.AudioFormat
 import com.vitorpamplona.nestsclient.audio.AudioPlayer
 import com.vitorpamplona.nestsclient.audio.NestPlayer
 import com.vitorpamplona.nestsclient.audio.OpusDecoder
@@ -56,9 +57,13 @@ import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.withTimeoutOrNull
 import kotlin.coroutines.cancellation.CancellationException

 /**
@@ -81,7 +86,11 @@ import kotlin.coroutines.cancellation.CancellationException
 * Audio-pipeline construction is injected via [decoderFactory] /
 * [playerFactory] so commonMain doesn't have to know which platform's
 * MediaCodec / AudioTrack is in play. M1 wires Android-only — desktop
- * passes nothing here yet.
+ * passes nothing here yet. Both factories take a `channelCount` (1 for
+ * mono, 2 for stereo) discovered from the publisher's `catalog.json`
+ * audio rendition; subscriptions await a brief catalog-arrival window
+ * before constructing the decoder + player so a stereo web publisher
+ * doesn't get its frames decoded as mono with downmix artifacts.
 *
 * **Threading contract:** all public methods (`connect`, `disconnect`,
 * `updateSpeakers`, `setMuted`, `setMicMuted`, `startBroadcast`,
@@ -98,8 +107,8 @@ import kotlin.coroutines.cancellation.CancellationException
 class NestViewModel(
    private val httpClient: NestsClient,
    private val transport: WebTransportFactory,
-    private val decoderFactory: () -> OpusDecoder,
-    private val playerFactory: () -> AudioPlayer,
+    private val decoderFactory: (channelCount: Int) -> OpusDecoder,
+    private val playerFactory: (channelCount: Int) -> AudioPlayer,
    private val signer: NostrSigner,
    private val room: NestsRoomConfig,
    // Speaker-side audio capture/encode actuals. Optional — desktop and
@@ -1038,6 +1047,55 @@ class NestViewModel(
        }
    }

+    /**
+     * Wait briefly for [pubkey]'s catalog to land in [_speakerCatalogs]
+     * and pick the channel count for the decoder + AudioTrack. Returns
+     * [AudioFormat.CHANNELS] on timeout (the catalog never arrived
+     * within [timeoutMs]) or when the catalog declares an unsupported
+     * count (anything outside `1..2`). Caller is responsible for
+     * having started [fetchSpeakerCatalog] first; otherwise this
+     * always times out.
+     *
+     * Why a timeout: the catalog handshake is best-effort. If the
+     * publisher doesn't publish a catalog (legacy publishers) or the
+     * relay never forwards the catalog group, audio playback should
+     * still proceed in the default config rather than block forever.
+     * 500 ms is generous — with the speaker-side
+     * `setOnNewSubscriber` hook the catalog group is on the wire
+     * within one round-trip of the SUBSCRIBE_OK, so this typically
+     * returns within tens of ms.
+     */
+    private suspend fun awaitDecoderChannelCount(
+        pubkey: String,
+        timeoutMs: Long,
+    ): Int {
+        val catalog =
+            withTimeoutOrNull(timeoutMs) {
+                _speakerCatalogs
+                    .map { it[pubkey] }
+                    .filterNotNull()
+                    .first()
+            }
+        val declared = catalog?.primaryAudio()?.numberOfChannels
+        return when {
+            declared == null -> {
+                AudioFormat.CHANNELS
+            }
+
+            declared !in 1..2 -> {
+                Log.w("NestRx") {
+                    "publisher catalog for pubkey='${pubkey.take(8)}' declares numberOfChannels=$declared " +
+                        "(only 1 / 2 supported); falling back to mono"
+                }
+                AudioFormat.CHANNELS
+            }
+
+            else -> {
+                declared
+            }
+        }
+    }
+
    /**
     * Open the speaker's `catalog.json` track in the background, parse
     * the first frame, and stash it in [speakerCatalogs]. Best-effort —
@@ -1100,15 +1158,29 @@ class NestViewModel(
                viewModelScope.launch { runCatching { handle.unsubscribe() } }
                return
            }
+            // Start the catalog fetch BEFORE we wait for it — runs in
+            // parallel with the main subscribe so the round-trip overlaps
+            // with the audio-side handshake. Cancelled by closeSubscription.
+            fetchSpeakerCatalog(l, pubkey)
+            // Wait briefly for the catalog so we can configure the
+            // decoder + AudioTrack to match the publisher's actual
+            // channel layout. With the speaker-side emit-on-subscribe
+            // hook the catalog frame is on the wire within one round-trip
+            // of the SUBSCRIBE_OK, so this typically returns within tens
+            // of ms. Falls back to mono if no catalog arrives within
+            // [CATALOG_AWAIT_TIMEOUT_MS] — covers legacy publishers and
+            // the failure-mode where the relay never forwards the catalog
+            // group; audio still plays, just in the default config.
+            val channelCount = awaitDecoderChannelCount(pubkey, CATALOG_AWAIT_TIMEOUT_MS)
            // Allocate native resources (MediaCodec decoder + AudioTrack
            // player on Android). Both are heavy and leaky if dropped on
            // the floor — wrap them in a nested try so any cancellation
            // or throw between here and slot.attach releases them
            // (audit round-2 VM #7).
-            val decoder = decoderFactory()
+            val decoder = decoderFactory(channelCount)
            val player =
                try {
-                    playerFactory()
+                    playerFactory(channelCount)
                } catch (t: Throwable) {
                    runCatching { decoder.release() }
                    throw t
@@ -1221,12 +1293,10 @@ class NestViewModel(
                _uiState.update {
                    it.copy(connectingSpeakers = (it.connectingSpeakers + pubkey).toPersistentSet())
                }
-                // Parallel catalog fetch — best-effort, doesn't gate
-                // audio playback. Tracked in catalogJobs; cancelled
-                // by closeSubscription so a removed speaker doesn't
-                // leave the catalog collector running on the
-                // wrapper's still-live re-issuing handle.
-                fetchSpeakerCatalog(l, pubkey)
+                // Catalog fetch was started earlier (before the decoder
+                // wait) so it could overlap with the audio-side handshake;
+                // its job is already in [catalogJobs] and gets cancelled
+                // by [closeSubscription] alongside the audio path.
            } catch (t: Throwable) {
                // Either CancellationException (scope cancelled mid-construction)
                // or an unexpected throw — release the half-built pipeline and
@@ -1718,6 +1788,22 @@ sealed class BroadcastUiState {
 */
 const val SPEAKING_TIMEOUT_MS: Long = 250L

+/**
+ * How long [NestViewModel.openSubscription] waits for the publisher's
+ * `catalog.json` to land before constructing the decoder + AudioTrack.
+ * The catalog declares the audio rendition's `numberOfChannels`, which
+ * the decoder needs at construction time so a stereo web publisher
+ * doesn't get its frames decoded as mono with downmix artifacts.
+ *
+ * Sized to be generous enough that a typical publisher's catalog
+ * (which arrives within a single round-trip of the SUBSCRIBE_OK with
+ * the speaker-side emit-on-subscribe hook in place) lands well before
+ * the timeout, and short enough that a publisher that never emits a
+ * catalog (legacy publishers, relay-side bug) doesn't visibly stall
+ * the listener — audio playback proceeds in the default mono config.
+ */
+const val CATALOG_AWAIT_TIMEOUT_MS: Long = 500L
+
 /**
 * Per-subscription consecutive Opus decode-error count that triggers a
 * single conspicuous warning log. Single-frame decode errors are normal
@@ -455,8 +455,8 @@ class NestViewModelTest {
        NestViewModel(
            httpClient = NoopNestsClient,
            transport = NoopWebTransportFactory,
-            decoderFactory = { NoopOpusDecoder },
-            playerFactory = { NoopAudioPlayer() },
+            decoderFactory = { _ -> NoopOpusDecoder },
+            playerFactory = { _ -> NoopAudioPlayer() },
            signer = NoopSigner,
            room = ROOM_CONFIG,
            connector =
@@ -77,7 +77,22 @@ import android.media.AudioFormat as AndroidAudioFormat
 class AudioTrackPlayer(
    private val usage: Int = AudioAttributes.USAGE_MEDIA,
    private val contentType: Int = AudioAttributes.CONTENT_TYPE_SPEECH,
+    /**
+     * Number of channels in the PCM stream this player will receive. Must
+     * match the [com.vitorpamplona.nestsclient.audio.OpusDecoder]'s output
+     * configuration (mono Opus → 1, stereo Opus → 2 with L/R interleaving).
+     * Drives both the AudioTrack channel mask and the underlying buffer-
+     * size target. Default is [AudioFormat.CHANNELS] (mono) so existing
+     * call sites that don't pass a channel count keep the prior behaviour.
+     */
+    private val channelCount: Int = AudioFormat.CHANNELS,
 ) : AudioPlayer {
+    init {
+        require(channelCount in 1..2) {
+            "AudioTrackPlayer supports mono (1) or stereo (2) only, got $channelCount"
+        }
+    }
+
    private var track: AudioTrack? = null
    private var muted: Boolean = false
    private var volume: Float = 1f
@@ -100,10 +115,10 @@ class AudioTrackPlayer(
            .d("NestPlay") { "AudioTrackPlayer.start() — allocating AudioTrack" }

        val channelMask =
-            when (AudioFormat.CHANNELS) {
+            when (channelCount) {
                1 -> AndroidAudioFormat.CHANNEL_OUT_MONO
                2 -> AndroidAudioFormat.CHANNEL_OUT_STEREO
-                else -> error("unsupported channel count ${AudioFormat.CHANNELS}")
+                else -> error("unsupported channel count $channelCount")
            }

        val minBuffer =
@@ -122,9 +137,11 @@ class AudioTrackPlayer(
        // miss its 20 ms cadence by an order of magnitude before the device
        // underruns. Take the larger of `minBuffer * 16` and an explicit
        // 250 ms-equivalent so devices that report a small minBuffer still
-        // get the same wall-clock slack.
+        // get the same wall-clock slack. Stereo doubles the byte count
+        // per sample (interleaved L,R 16-bit shorts) — scaling
+        // [channelCount] in keeps the wall-clock target constant.
        val targetBytes250Ms =
-            (AudioFormat.SAMPLE_RATE_HZ / 4) * AudioFormat.BYTES_PER_SAMPLE * AudioFormat.CHANNELS
+            (AudioFormat.SAMPLE_RATE_HZ / 4) * AudioFormat.BYTES_PER_SAMPLE * channelCount
        val bufferBytes = maxOf(minBuffer * 16, targetBytes250Ms)

        val newTrack =
@@ -31,21 +31,42 @@ import java.nio.ByteOrder
 * across packets, so sharing a decoder across speakers would cause clicks.
 *
 * Configuration:
- *   - 48 kHz mono signed 16-bit PCM output (matches [AudioFormat]).
- *   - CSD-0: Opus identification header per RFC 7845 §5.1, 19 bytes.
+ *   - 48 kHz signed 16-bit PCM output (Opus's internal sample rate;
+ *     MediaCodec's Opus decoder always emits 48 kHz regardless of the
+ *     OpusHead `inputSampleRate` field).
+ *   - [channelCount] — 1 (mono) or 2 (stereo) interleaved L/R. Drives both
+ *     the MediaFormat channel-count and the OpusHead CSD-0 channel byte.
+ *     A web publisher (kixelated/hang) emits stereo when the user's
+ *     AudioContext picks a stereo input; without matching channel
+ *     configuration the decoder either downmixes with artifacts or
+ *     refuses the packet.
+ *   - CSD-0: Opus identification header per RFC 7845 §5.1, 19 bytes
+ *     (mapping family 0; covers both mono and stereo with implicit
+ *     L,R interleaving).
 *   - CSD-1 / CSD-2: pre-skip + seek pre-roll, both zero (we don't seek).
+ *
+ * Default is [AudioFormat.CHANNELS] (mono) so existing call sites that
+ * don't pass a channel count continue to behave exactly as before.
 */
-class MediaCodecOpusDecoder : OpusDecoder {
+class MediaCodecOpusDecoder(
+    private val channelCount: Int = AudioFormat.CHANNELS,
+) : OpusDecoder {
+    init {
+        require(channelCount in 1..2) {
+            "MediaCodecOpusDecoder supports mono (1) or stereo (2) only, got $channelCount"
+        }
+    }
+
    private val codec: MediaCodec =
        try {
            MediaCodec
                .createDecoderByType(MediaFormat.MIMETYPE_AUDIO_OPUS)
                .apply {
-                    configure(buildFormat(), null, null, 0)
+                    configure(buildFormat(channelCount), null, null, 0)
                    start()
                }.also {
                    com.vitorpamplona.quartz.utils.Log.d("NestPlay") {
-                        "MediaCodecOpusDecoder allocated codec='${it.name}'"
+                        "MediaCodecOpusDecoder allocated codec='${it.name}' channelCount=$channelCount"
                    }
                }
        } catch (t: Throwable) {
@@ -70,7 +91,11 @@ class MediaCodecOpusDecoder : OpusDecoder {
        // via ShortBuffer.get(dst, off, len) — the previous shape went
        // through ArrayList<Short>, which boxed every PCM sample
        // (~48 000 alloc/sec/speaker on the audio hot path).
-        val out = ShortArray(AudioFormat.FRAME_SIZE_SAMPLES)
+        //
+        // Stereo Opus emits L/R-interleaved samples, so a 20 ms / 960-
+        // sample frame produces `FRAME_SIZE_SAMPLES * channelCount`
+        // shorts — twice as many for stereo as for mono.
+        val out = ShortArray(AudioFormat.FRAME_SIZE_SAMPLES * channelCount)
        var outPos = 0

        // 1. Acquire an input slot. On rare back-pressure (output buffers
@@ -196,14 +221,14 @@ class MediaCodecOpusDecoder : OpusDecoder {

        private const val FRAME_DURATION_US = 20_000L // 20 ms

-        private fun buildFormat(): MediaFormat {
+        private fun buildFormat(channelCount: Int): MediaFormat {
            val format =
                MediaFormat.createAudioFormat(
                    MediaFormat.MIMETYPE_AUDIO_OPUS,
                    AudioFormat.SAMPLE_RATE_HZ,
-                    AudioFormat.CHANNELS,
+                    channelCount,
                )
-            format.setByteBuffer("csd-0", ByteBuffer.wrap(buildOpusIdHeader()))
+            format.setByteBuffer("csd-0", ByteBuffer.wrap(buildOpusIdHeader(channelCount)))
            // Pre-skip + seek pre-roll: both zero, encoded as little-endian
            // 64-bit nanoseconds per Android's MediaCodec contract.
            format.setByteBuffer("csd-1", ByteBuffer.wrap(zeroLongLe()))
@@ -211,12 +236,14 @@ class MediaCodecOpusDecoder : OpusDecoder {
            return format
        }

-        private fun buildOpusIdHeader(): ByteArray {
-            // RFC 7845 §5.1 — 19 bytes for mono, mapping family 0.
+        private fun buildOpusIdHeader(channelCount: Int): ByteArray {
+            // RFC 7845 §5.1 — 19 bytes, mapping family 0 (covers mono and
+            // stereo with implicit L,R interleaving; no per-channel
+            // mapping table required).
            val buf = ByteBuffer.allocate(19).order(ByteOrder.LITTLE_ENDIAN)
            buf.put("OpusHead".encodeToByteArray()) // 8 bytes magic
            buf.put(1.toByte()) // version
-            buf.put(AudioFormat.CHANNELS.toByte()) // channel count
+            buf.put(channelCount.toByte()) // channel count (1 or 2)
            buf.putShort(0) // pre-skip
            buf.putInt(AudioFormat.SAMPLE_RATE_HZ) // input sample rate
            buf.putShort(0) // output gain (Q7.8 dB)