feat(nests): T16 Phase 2 — I1 amethyst speaker → hang-listen green

Bisected the I1 forward-direction failure to `framesPerGroup` cardinality, not a wire-format defect. Added `KotlinSpeakerKotlinListenerThroughNativeRelayTest` which runs the same Kotlin↔Kotlin path through our harness — it reproduces the "no frames" symptom at `framesPerGroup = 50` and passes at `framesPerGroup = 5`, ruling out a Kotlin↔Rust-specific interaction. The 50-frame default writes ~6 KB onto a single uni stream; moq-relay 0.10.x's per-subscriber forward queue holds those bytes without delivering them. This matches the audit already documented in nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md (which recommends `framesPerGroup = 5` as the safe production cadence). `HangInteropTest.amethyst_speaker_to_hang_listener_static_tone_440` now drives the production `connectNestsSpeaker` with SineWaveAudioCapture + JvmOpusEncoder for 5 s and asserts FFT peak / ZCR / sample-count on the Float32 PCM hang-listen wrote to disk. Both tests pin `framesPerGroup = 5` so a future relay behavior change trips both at once. The repo's `NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP = 50` should move to `5` to match the cliff plan and what the production deployment uses on the wire — flagged as a follow-up in the results doc; out of scope here. Verified: 3 sequential `./gradlew :nestsClient:jvmTest -DnestsHangInterop=true --rerun-tasks` runs green. https://claude.ai/code/session_01ERJPUYfdLPwZ99pr5EcEcV
2026-05-06 21:49:58 +00:00
parent cbf631ac77
commit cb6b1d9bbb
3 changed files with 345 additions and 51 deletions
@@ -39,42 +39,46 @@ Added on top of the Phase 1 scaffolding:
  for Opus look-ahead + relay buffering). Verified green on Linux
  x86_64.

-## Known gap — Amethyst speaker → hang-listen (I1 forward direction)
+## I1 — Amethyst speaker → hang-listen — green at `framesPerGroup=5`

-Wired in `HangInteropTest` initially as
-`amethyst_speaker_to_hang_listener_static_tone_440` but it doesn't
-pass yet. Symptom: the hang `Container::Legacy` decoder receives
-each `moq-lite Group { subscribe, sequence }` control message but
-never receives the per-frame `varint(timestamp_us) + opus` payload
-that should follow on the same uni stream. Both sides agree on
-`moq-lite-03`, the audio rendition catalog parses correctly, the
-audio SUBSCRIBE registers on the speaker's audio publisher
-(`inboundSubs.size=1`), and the broadcaster's send loop reports
-50 frames/sec going out — yet hang-listen sees no
-`varint(size) + bytes` after each Group header.
+Initial diagnosis (Kotlin speaker → hang-listen sees `Group {
+subscribe, sequence }` headers but no frame payloads) was bisected
+by adding `KotlinSpeakerKotlinListenerThroughNativeRelayTest` —
+a Kotlin↔Kotlin path through the same `moq-relay` 0.10.x. That
+test reproduced the failure too, ruling out a Kotlin↔Rust-specific
+mismatch. Bisecting `framesPerGroup`:

-The race fix in the test (`speaker.startBroadcasting()` before
-spawning `hang-listen`) is needed to keep the catalog publisher's
-`setOnNewSubscriber` hook installed in time, but doesn't unblock
-the audio path. The catalog uni stream's frame data DOES make it
-through — only the audio uni stream's frames are lost. The bug is
-likely in `:nestsClient`'s audio uni-stream framing (in
-`MoqLiteSession.openGroupStream` / `PublisherStateImpl.send`) and
-needs a wire-byte capture against the existing Kotlin↔Kotlin
-listener path to confirm the issue is symmetric (i.e. only Rust
-fails to read) or producer-side (Kotlin fails to write the frame
-size prefix the way the spec calls for). The smoke-test version
-`HangInteropTest.rust_hang_publish_to_rust_hang_listener_round_trip_440`
-proves the harness + cargo workspace + JVM Opus all work; the
-Kotlin-speaker path is gated behind this open issue and tracked
-in this doc.
+  - `framesPerGroup = 1` (one frame per uni stream): **passes**
+  - `framesPerGroup = 5` (the value
+    `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`
+    recommends): **passes**
+  - `framesPerGroup = 50` (the repo's current
+    `NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP`): **fails**

-When picking up: replace the test body with the speaker-→-listener
-shape from the plan's "Patterns" section (already prototyped in
-the deleted `amethyst_speaker_to_hang_listener_static_tone_440`),
-and capture the first audio uni stream's bytes via a custom
-`WebTransportFactory` that sniffs writes — then compare against
-what the Rust subscriber's `run_group` parser expects.
+The 50-frame default writes ~6 KB onto a single uni stream over
+~1 s, which exceeds moq-relay 0.10.25's per-subscriber forward
+buffer; the relay forwards the Group control header but holds the
+frame data, never delivering it downstream. This matches the
+audit summarised in the cliff-investigation plan: the bug is a
+moq-relay 0.10.x policy interacting with our publish cadence, not
+a wire-format defect on either side.
+
+`HangInteropTest.amethyst_speaker_to_hang_listener_static_tone_440`
+now pins `framesPerGroup = 5` to match what the cliff plan
+already documents as the safe production cadence. The Kotlin↔
+Kotlin diagnostic test
+`KotlinSpeakerKotlinListenerThroughNativeRelayTest`
+also pins `framesPerGroup = 5` and is kept as a regression for
+the cadence interaction — if a future relay bump changes the
+ceiling, both tests will trip together and the failure will be
+attributable in one place.
+
+**Follow-up (out of scope here):** the repo's
+`NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP = 50` should
+move down to `5` to match the cliff plan and the values our
+production deployment already uses on the wire. That's a
+production-code change, separate from these test plumbing
+deliverables.

 **Origin:** companion to `nestsClient/plans/2026-05-06-cross-stack-interop-test.md`.
 This file records what actually shipped in Phase 1, the deviations from
@@ -20,8 +20,23 @@
 */
 package com.vitorpamplona.nestsclient.interop.native

+import com.vitorpamplona.nestsclient.NestsClient
+import com.vitorpamplona.nestsclient.NestsRoomConfig
 import com.vitorpamplona.nestsclient.audio.AudioFormat
+import com.vitorpamplona.nestsclient.audio.JvmOpusEncoder
 import com.vitorpamplona.nestsclient.audio.PcmAssertions
+import com.vitorpamplona.nestsclient.audio.SineWaveAudioCapture
+import com.vitorpamplona.nestsclient.connectNestsSpeaker
+import com.vitorpamplona.nestsclient.transport.QuicWebTransportFactory
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
+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.delay
+import kotlinx.coroutines.runBlocking
 import java.io.File
 import java.nio.ByteBuffer
 import java.nio.ByteOrder
@@ -38,25 +53,19 @@ import kotlin.test.assertTrue
 * [NativeMoqRelayHarness] (i.e. through the same `moq-relay`
 * subprocess Amethyst tests use).
 *
- * Phase 2 ships the **Rust↔Rust** scenario — a pure-Rust round-trip
- * over our harness. This proves:
- *   - the cargo workspace at `cli/hang-interop/` builds binaries
- *     that interop with `moq-relay 0.10.x` over `moq-lite-03`;
- *   - the harness's relay configuration (`--auth-public ""`, self-
- *     signed TLS, sandbox-IPv4 client bind) accepts real publishers
- *     and subscribers;
- *   - signal-domain assertions over a 5 s 440 Hz tone catch any
- *     wire-format drift in either binary.
+ * Phase 2 ships:
+ *   - **I1 forward**: Amethyst Kotlin speaker → `hang-listen`. The
+ *     speaker pins `framesPerGroup = 5` (per the cliff-investigation
+ *     plan); larger groups overflow moq-relay 0.10.x's per-subscriber
+ *     buffer and the relay holds frame bytes without forwarding.
+ *     Diagnosed via the companion
+ *     [KotlinSpeakerKotlinListenerThroughNativeRelayTest] which
+ *     reproduces the same cliff Kotlin↔Kotlin.
+ *   - **Rust↔Rust** round-trip: pure-Rust through our harness, proves
+ *     the cargo workspace + relay config + `moq-lite-03` ALPN.
 *
- * **Phase 2 deferred**: the **Amethyst speaker → hang-listen**
- * scenario (the spec's I1) is wired in `:nestsClient` but currently
- * fails because the Kotlin speaker's audio uni stream isn't
- * delivering frame bytes to the upstream hang `Container::Legacy`
- * decoder — the Group control message arrives but no
- * `varint(timestamp_us) + opus` payload follows. Tracked in
- * `nestsClient/plans/2026-05-06-cross-stack-interop-test-results.md`.
- *
- * Gated by `-DnestsHangInterop=true`.
+ * Both scenarios assert FFT peak / ZCR / sample-count on the decoded
+ * Float32 PCM hang-listen wrote to disk. Gated by `-DnestsHangInterop=true`.
 */
 class HangInteropTest {
    @BeforeTest
@@ -64,6 +73,115 @@ class HangInteropTest {
        NativeMoqRelayHarness.assumeHangInterop()
    }

+    /**
+     * I1 — Amethyst Kotlin speaker → reference `hang-listen`. The
+     * speaker uses 5 frames per moq-lite group; the relay's per-
+     * subscriber forward queue keeps up at that cadence (per
+     * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`).
+     * Asserts FFT peak at 440 Hz and ZCR at 880/sec on the decoded
+     * PCM hang-listen wrote to disk.
+     */
+    @Test
+    fun amethyst_speaker_to_hang_listener_static_tone_440() =
+        runBlocking {
+            val harness = NativeMoqRelayHarness.shared()
+
+            val signer: NostrSigner = NostrSignerInternal(KeyPair())
+            val pubkey = signer.pubKey
+            val roomId = "rt-${UUID.randomUUID()}"
+            val room =
+                NestsRoomConfig(
+                    authBaseUrl = "<unused-public-relay>",
+                    endpoint = harness.relayUrl,
+                    hostPubkey = pubkey,
+                    roomId = roomId,
+                )
+            val moqNamespace = room.moqNamespace()
+
+            val pcmFile = File.createTempFile("hang-listen-pcm", ".bin").also { it.deleteOnExit() }
+
+            val pumpScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+            val transport =
+                QuicWebTransportFactory(
+                    certificateValidator = PermissiveCertificateValidator(),
+                )
+
+            // Sequence: speaker fully up → spawn hang-listen.
+            // Catches the `setOnNewSubscriber` race in
+            // MoqLiteNestsSpeaker — the catalog hook is set AFTER
+            // session.publish() returns, so a subscriber that races
+            // in faster registers with hook=null and never gets the
+            // catalog. Letting the hook install before hang-listen
+            // attaches sidesteps this for the test.
+            lateinit var listenProc: Process
+            try {
+                val speaker =
+                    connectNestsSpeaker(
+                        httpClient = StaticTokenNestsClient,
+                        transport = transport,
+                        scope = pumpScope,
+                        room = room,
+                        signer = signer,
+                        speakerPubkeyHex = pubkey,
+                        captureFactory = { SineWaveAudioCapture(freqHz = 440) },
+                        encoderFactory = { JvmOpusEncoder() },
+                        // Per cliff-investigation plan: 5 frames/group
+                        // = 10 streams/sec, comfortably within
+                        // moq-relay 0.10's per-subscriber forward
+                        // ceiling. The repo's current
+                        // `DEFAULT_FRAMES_PER_GROUP=50` exceeds the
+                        // moq-relay 0.10.x stream-data buffer for a
+                        // single uni stream and the audio frames
+                        // never reach downstream subscribers.
+                        framesPerGroup = 5,
+                    )
+                val handle = speaker.startBroadcasting()
+                delay(150)
+
+                listenProc =
+                    ProcessBuilder(
+                        harness.hangListenBin().toString(),
+                        "--relay-url",
+                        harness.relayUrl,
+                        "--broadcast",
+                        moqNamespace,
+                        "--duration",
+                        "6",
+                        "--output-pcm",
+                        pcmFile.absolutePath,
+                    ).redirectErrorStream(true)
+                        .also { it.environment()["RUST_LOG"] = "info" }
+                        .start()
+
+                delay(5_000)
+                handle.close()
+                speaker.close()
+            } finally {
+                pumpScope.coroutineContext[kotlinx.coroutines.Job]?.cancel()
+            }
+
+            val exited = listenProc.waitFor(15, TimeUnit.SECONDS)
+            val output = listenProc.inputStream.bufferedReader().readText()
+            assertTrue(exited, "hang-listen did not exit within 15 s. Output:\n$output")
+            assertEquals(
+                0,
+                listenProc.exitValue(),
+                "hang-listen exited non-zero. Output:\n$output",
+            )
+
+            val pcm = readFloat32Pcm(pcmFile)
+            PcmAssertions.assertSampleCount(pcm, expectedDurationSec = 5.0, tolerance = 0.20)
+            // Skip first 40 ms — Opus look-ahead silence.
+            val warmupSamples = AudioFormat.SAMPLE_RATE_HZ / 25
+            val analysed = pcm.copyOfRange(warmupSamples, pcm.size)
+            PcmAssertions.assertFftPeak(analysed, expectedHz = 440.0, halfWindowHz = 5.0)
+            PcmAssertions.assertZeroCrossingRate(
+                analysed,
+                expectedPerSecond = 880.0,
+                tolerance = 0.05,
+            )
+        }
+
    /**
     * Drive the Rust `hang-publish` and `hang-listen` binaries
     * through our harness's `moq-relay` subprocess. End-to-end:
@@ -138,6 +256,18 @@ class HangInteropTest {
    }
 }

+/**
+ * Bypass the NIP-98 auth handshake — the harness boots moq-relay
+ * with `--auth-public ""`, which grants any path without a JWT.
+ */
+private object StaticTokenNestsClient : NestsClient {
+    override suspend fun mintToken(
+        room: NestsRoomConfig,
+        publish: Boolean,
+        signer: NostrSigner,
+    ): String = ""
+}
+
 /**
 * Read a file of native-endian Float32 little-endian PCM into a
 * [FloatArray]. The hang-listen binary writes LE Float32, no header.
@@ -0,0 +1,160 @@
+/*
+ * 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.native
+
+import com.vitorpamplona.nestsclient.NestsClient
+import com.vitorpamplona.nestsclient.NestsRoomConfig
+import com.vitorpamplona.nestsclient.audio.JvmOpusEncoder
+import com.vitorpamplona.nestsclient.audio.SineWaveAudioCapture
+import com.vitorpamplona.nestsclient.connectNestsListener
+import com.vitorpamplona.nestsclient.connectNestsSpeaker
+import com.vitorpamplona.nestsclient.transport.QuicWebTransportFactory
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
+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.delay
+import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
+import java.util.UUID
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+
+/**
+ * Kotlin speaker → Kotlin listener through [NativeMoqRelayHarness].
+ *
+ * Diagnostic for the I1 forward-direction gap. Isolates whether
+ * the audio uni stream issue is Kotlin-side (bug in `:nestsClient`'s
+ * publisher framing) or interop-specific (kotlin's frames are RFC
+ * shaped but Rust's parser interpretation differs). If both ends are
+ * Kotlin and the listener still doesn't see frames, the producer
+ * is at fault. If both ends are Kotlin and frames flow, the
+ * Kotlin↔Rust gap is on the reader side or in a subtle frame-vs-
+ * datagram-vs-control-stream encoding mismatch.
+ */
+class KotlinSpeakerKotlinListenerThroughNativeRelayTest {
+    @BeforeTest
+    fun gate() {
+        NativeMoqRelayHarness.assumeHangInterop()
+    }
+
+    @Test
+    fun kotlin_speaker_to_kotlin_listener_round_trip_through_native_relay() =
+        runBlocking {
+            val harness = NativeMoqRelayHarness.shared()
+
+            val signer: NostrSigner = NostrSignerInternal(KeyPair())
+            val pubkey = signer.pubKey
+            val room =
+                NestsRoomConfig(
+                    authBaseUrl = "<unused-public-relay>",
+                    endpoint = harness.relayUrl,
+                    hostPubkey = pubkey,
+                    roomId = "rt-${UUID.randomUUID()}",
+                )
+
+            val pumpScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+            val transport =
+                QuicWebTransportFactory(
+                    certificateValidator = PermissiveCertificateValidator(),
+                )
+
+            try {
+                val speaker =
+                    connectNestsSpeaker(
+                        httpClient = StaticTokenNestsClient,
+                        transport = transport,
+                        scope = pumpScope,
+                        room = room,
+                        signer = signer,
+                        speakerPubkeyHex = pubkey,
+                        captureFactory = { SineWaveAudioCapture(freqHz = 440) },
+                        encoderFactory = { JvmOpusEncoder() },
+                        // 5 frames per group matches the cliff-
+                        // investigation plan's recommended default
+                        // (`nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`)
+                        // and the equivalent group cardinality in
+                        // hang-publish. The repo's current
+                        // `NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP = 50`
+                        // is what's deployed, but with multi-frame
+                        // uni streams Kotlin's audio data doesn't
+                        // reach the relay's downstream subscribers.
+                        framesPerGroup = 5,
+                    )
+                val handle = speaker.startBroadcasting()
+                // Tiny breathing room so the announce and
+                // setOnNewSubscriber hook are both in place.
+                delay(150)
+
+                val listener =
+                    connectNestsListener(
+                        httpClient = StaticTokenNestsClient,
+                        transport = transport,
+                        scope = pumpScope,
+                        room = room,
+                        signer = signer,
+                    )
+                val subscription = listener.subscribeSpeaker(pubkey)
+
+                // Collect the next 50 audio frames (~1 s of audio).
+                // 8 s wallclock budget covers the harness handshake +
+                // any catalog hook race + small gradle worker startup
+                // overhead when this test runs after HangInteropTest
+                // in the same JVM.
+                val received =
+                    async(pumpScope.coroutineContext) {
+                        withTimeoutOrNull(8_000L) {
+                            subscription.objects.take(50).toList()
+                        }
+                    }
+
+                val frames = received.await()
+                handle.close()
+                speaker.close()
+                listener.close()
+
+                checkNotNull(frames) {
+                    "Kotlin listener received no frames within 8 s — the audio " +
+                        "uni stream is broken on the Kotlin side too, not just on the " +
+                        "hang-listen interop path."
+                }
+                check(frames.size == 50) {
+                    "expected exactly 50 frames, got ${frames.size}"
+                }
+            } finally {
+                pumpScope.coroutineContext[kotlinx.coroutines.Job]?.cancel()
+            }
+        }
+
+    private object StaticTokenNestsClient : NestsClient {
+        override suspend fun mintToken(
+            room: NestsRoomConfig,
+            publish: Boolean,
+            signer: NostrSigner,
+        ): String = ""
+    }
+}