test(nests): T16 Phase 3 — udp-loss-shim + I9 + I5 hot-swap

- **udp-loss-shim** body: tokio UDP loopback that drops a
  configurable fraction of datagrams. Single-tenant (one client
  at a time) — moq-lite is connection-multiplexed by source port,
  so 1:1 forwarding is enough.
- **I9** (`packet_loss_1pct_does_not_kill_audio`): routes the
  Kotlin speaker through the shim with `--loss-rate 0.01`;
  asserts the decoded PCM has ≥ 80% expected sample count and
  the 440 Hz tone survives. moq-lite groups are reliable streams
  so retransmits absorb the loss.
- **I5** (`speaker_hot_swap_does_not_crash`): drives the
  reconnecting-speaker with `tokenRefreshAfterMs = 2_500` to
  force a hot-swap mid-broadcast. The reference hang-listen is
  single-shot subscribe (it doesn't re-subscribe on broadcast
  re-announce), so it captures only the pre-swap chunk; the
  test asserts the speaker survives without corrupting active
  uni streams (≥ 1 s of audio + FFT peak at 440 Hz on the
  captured chunk). The "no audible gap" property the spec
  calls for is an Amethyst-listener concern (handles re-announce
  transparently); a Phase 3 follow-up would exercise that path
  end-to-end through `connectNestsListener`.

I7 (publisher reconnect on the Rust side, ref→A) is the
mirror image and would need hang-publish to take a
"--reconnect-after-ms" flag, plus the Amethyst listener path
through the harness — also Phase 3 follow-up.

https://claude.ai/code/session_01ERJPUYfdLPwZ99pr5EcEcV

cli/hang-interop/Cargo.lock

@@ -555,7 +555,7 @@ checksum = "4e7f34442dbe69c60fe8eaf58a8cafff81a1f278816d8ab4db255b3bef4ac3c4"
 dependencies = [
  "getrandom 0.3.4",
  "libm",
- "rand",
+ "rand 0.9.4",
  "siphasher",
 ]
 
@@ -1268,7 +1268,7 @@ dependencies = [
  "conducer",
  "futures",
  "num_enum",
- "rand",
+ "rand 0.9.4",
  "serde",
  "thiserror 2.0.18",
  "tokio",
@@ -1331,7 +1331,7 @@ dependencies = [
  "moq-lite",
  "parking_lot",
  "quinn",
- "rand",
+ "rand 0.9.4",
  "rcgen",
  "reqwest",
  "rustls",
@@ -1671,7 +1671,7 @@ dependencies = [
  "fastbloom",
  "getrandom 0.3.4",
  "lru-slab",
- "rand",
+ "rand 0.9.4",
  "ring",
  "rustc-hash",
  "rustls",
@@ -1713,14 +1713,35 @@ version = "5.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
 
+[[package]]
+name = "rand"
+version = "0.8.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a"
+dependencies = [
+ "libc",
+ "rand_chacha 0.3.1",
+ "rand_core 0.6.4",
+]
+
 [[package]]
 name = "rand"
 version = "0.9.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea"
 dependencies = [
- "rand_chacha",
+ "rand_chacha 0.9.0",
- "rand_core",
+ "rand_core 0.9.5",
+]
+
+[[package]]
+name = "rand_chacha"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88"
+dependencies = [
+ "ppv-lite86",
+ "rand_core 0.6.4",
 ]
 
 [[package]]
@@ -1730,7 +1751,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
 dependencies = [
  "ppv-lite86",
- "rand_core",
+ "rand_core 0.9.5",
+]
+
+[[package]]
+name = "rand_core"
+version = "0.6.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
+dependencies = [
+ "getrandom 0.2.17",
 ]
 
 [[package]]
@@ -2639,7 +2669,10 @@ version = "0.0.1"
 dependencies = [
  "anyhow",
  "clap",
+ "rand 0.8.6",
  "tokio",
+ "tracing",
+ "tracing-subscriber",
 ]
 
 [[package]]

cli/hang-interop/udp-loss-shim/Cargo.toml

@@ -5,7 +5,15 @@ edition.workspace = true
 publish.workspace = true
 license.workspace = true
 
-# Phase 1: stub. Phase 3 wires this for the I9 packet-loss scenario.
+# UDP loopback that drops a configurable fraction of datagrams.
+# Used by the I9 packet-loss interop scenario:
+#
+#   client (--server-bind 0.0.0.0:0) → udp-loss-shim --listen X
+#                                       → moq-relay --upstream Y
+#
+# The shim is a single-tenant relay (one client at a time) — moq-lite
+# is on QUIC which is connection-multiplexed by the client's source
+# port, so we forward 1:1.
 
 [[bin]]
 name = "udp-loss-shim"
@@ -15,3 +23,6 @@ path = "src/main.rs"
 anyhow.workspace = true
 clap.workspace = true
 tokio.workspace = true
+rand = "0.8"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }

cli/hang-interop/udp-loss-shim/src/main.rs

@@ -1,27 +1,152 @@
-//! udp-loss-shim — UDP loopback that drops a configurable fraction of
+//! udp-loss-shim — UDP loopback that drops a configurable fraction
-//! datagrams, used by the I9 packet-loss scenario. **Phase 1 stub.**
+//! of datagrams. Used by the I9 packet-loss cross-stack interop
+//! scenario. See
+//! `nestsClient/plans/2026-05-06-cross-stack-interop-test.md`.
+//!
+//! Topology:
+//!
+//!   client → `--listen <addr>` (this binary) → `--upstream <addr>` (moq-relay)
+//!
+//! The shim picks one client (the first peer that sends to its
+//! listen socket), forwards datagrams in both directions
+//! 1:1 modulo the loss roll, and exits when the parent test
+//! kills it. moq-lite is on QUIC which is connection-multiplexed
+//! by the client's source port, so single-tenant forwarding is
+//! enough for the test scenarios.
 
-use anyhow::Result;
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
 use clap::Parser;
+use tokio::net::UdpSocket;
+use tokio::sync::Mutex;
 
 #[derive(Parser, Debug)]
-#[command(name = "udp-loss-shim", about = "UDP packet-loss shim (Phase 1 stub)")]
+#[command(name = "udp-loss-shim", about = "UDP loopback with configurable packet loss")]
 struct Args {
+    /// Address to listen on (the client connects here).
     #[arg(long)]
     listen: String,
+
+    /// Upstream address to forward to (moq-relay's UDP port).
     #[arg(long)]
     upstream: String,
+
+    /// Fraction of datagrams to drop, 0.0–1.0. Applied independently
+    /// to each direction.
     #[arg(long, default_value_t = 0.0)]
     loss_rate: f32,
 }
 
 #[tokio::main]
 async fn main() -> Result<()> {
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .with_writer(std::io::stderr)
+        .try_init();
+
     let args = Args::parse();
-    eprintln!(
+    anyhow::ensure!(
-        "udp-loss-shim Phase-1 stub — listen={} upstream={} loss_rate={}",
+        (0.0..=1.0).contains(&args.loss_rate),
-        args.listen, args.upstream, args.loss_rate
+        "loss-rate must be in 0.0..=1.0, got {}",
+        args.loss_rate
     );
-    eprintln!("Phase 3 will implement actual UDP forwarding. Exiting cleanly.");
+
-    Ok(())
+    let listen_addr: SocketAddr = args.listen.parse().context("parse --listen")?;
+    let upstream_addr: SocketAddr = args.upstream.parse().context("parse --upstream")?;
+
+    // Listen socket — accepts datagrams from the client.
+    let listen_sock = Arc::new(
+        UdpSocket::bind(listen_addr)
+            .await
+            .with_context(|| format!("bind --listen {listen_addr}"))?,
+    );
+    // Upstream socket — talks to moq-relay. Bound to an ephemeral
+    // port; relay's outbound path back replies to whatever source
+    // port we picked.
+    let upstream_sock = Arc::new(
+        UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0)))
+            .await
+            .context("bind upstream socket")?,
+    );
+    upstream_sock
+        .connect(upstream_addr)
+        .await
+        .with_context(|| format!("connect upstream {upstream_addr}"))?;
+
+    tracing::info!(
+        listen = %listen_addr,
+        upstream = %upstream_addr,
+        loss_rate = args.loss_rate,
+        "udp-loss-shim ready"
+    );
+
+    // Track the client's source address. moq-lite's QUIC client
+    // doesn't use connection migration in our test setup, so the
+    // first peer that sends to us is the only client we care about.
+    let client_addr: Arc<Mutex<Option<SocketAddr>>> = Arc::new(Mutex::new(None));
+
+    // Direction 1: client → upstream (with loss).
+    let loss = args.loss_rate;
+    let listen_clone = listen_sock.clone();
+    let upstream_clone = upstream_sock.clone();
+    let client_clone = client_addr.clone();
+    tokio::spawn(async move {
+        let mut buf = [0u8; 65_535];
+        loop {
+            let (n, src) = match listen_clone.recv_from(&mut buf).await {
+                Ok(v) => v,
+                Err(e) => {
+                    tracing::warn!(%e, "listen recv error; exiting");
+                    return;
+                }
+            };
+            // Latch the client address on first packet.
+            {
+                let mut c = client_clone.lock().await;
+                if c.is_none() {
+                    tracing::info!(%src, "client latched");
+                    *c = Some(src);
+                }
+            }
+            if rand::random::<f32>() < loss {
+                tracing::trace!(bytes = n, %src, "drop client→upstream");
+                continue;
+            }
+            if let Err(e) = upstream_clone.send(&buf[..n]).await {
+                tracing::warn!(%e, "upstream send failed");
+            }
+        }
+    });
+
+    // Direction 2: upstream → client (with loss).
+    let loss = args.loss_rate;
+    let upstream_clone = upstream_sock.clone();
+    let listen_clone = listen_sock.clone();
+    let client_clone = client_addr.clone();
+    let mut buf = [0u8; 65_535];
+    loop {
+        let n = match upstream_clone.recv(&mut buf).await {
+            Ok(v) => v,
+            Err(e) => {
+                tracing::warn!(%e, "upstream recv error; exiting");
+                return Ok(());
+            }
+        };
+        if rand::random::<f32>() < loss {
+            tracing::trace!(bytes = n, "drop upstream→client");
+            continue;
+        }
+        let dst = match *client_clone.lock().await {
+            Some(addr) => addr,
+            None => {
+                tracing::trace!(bytes = n, "upstream sent before client latched; ignoring");
+                continue;
+            }
+        };
+        if let Err(e) = listen_clone.send_to(&buf[..n], dst).await {
+            tracing::warn!(%e, %dst, "listen send_to failed");
+        }
+    }
 }

nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/native/HangInteropTest.kt

@@ -28,6 +28,7 @@ import com.vitorpamplona.nestsclient.audio.PcmAssertions
 import com.vitorpamplona.nestsclient.audio.SineWaveAudioCapture
 import com.vitorpamplona.nestsclient.buildRelayConnectTarget
 import com.vitorpamplona.nestsclient.connectNestsSpeaker
+import com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker
 import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession
 import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSubscribeException
 import com.vitorpamplona.nestsclient.transport.QuicWebTransportFactory
@@ -384,6 +385,88 @@ class HangInteropTest {
             PcmAssertions.assertFftPeak(tailWindow, expectedHz = 440.0, halfWindowHz = 5.0)
         }
 
+    /**
+     * I9 — 1% packet loss via the `udp-loss-shim` between the
+     * Kotlin speaker's UDP socket and the relay. Asserts the
+     * decoded PCM still has ≥ 80 % of the expected sample count
+     * and the 440 Hz peak survives — moq-lite groups are
+     * reliable streams (`bestEffort=false`), so lost bytes get
+     * retransmitted and the listener still observes the whole
+     * tone with mild jitter.
+     */
+    @Test
+    fun packet_loss_1pct_does_not_kill_audio() =
+        runBlocking {
+            val out =
+                runSpeakerToHangListen(
+                    speakerSeconds = 5,
+                    captureFirstFrame = false,
+                    udpLossRate = 0.01f,
+                )
+            val pcm = readFloat32Pcm(out.pcmFile)
+            val expected = 5.0 * AudioFormat.SAMPLE_RATE_HZ
+            assertTrue(
+                pcm.size >= expected * 0.80,
+                "expected ≥ 80% of $expected samples under 1% packet loss, " +
+                    "got ${pcm.size} (${"%.1f".format(pcm.size / expected * 100)} %)",
+            )
+            val warmup = AudioFormat.SAMPLE_RATE_HZ / 25
+            val analysed = pcm.copyOfRange(warmup, pcm.size)
+            PcmAssertions.assertFftPeak(analysed, expectedHz = 440.0, halfWindowHz = 5.0)
+        }
+
+    /**
+     * I5 — speaker hot-swap mid-broadcast (proactive JWT refresh).
+     * Drives `connectReconnectingNestsSpeaker` with a 2.5 s
+     * `tokenRefreshAfterMs` so the speaker recycles its session
+     * mid-broadcast.
+     *
+     * **Limitation:** the reference `hang-listen` is single-shot
+     * — it reads the catalog once and subscribes once. When the
+     * speaker hot-swaps, the relay unannounces the old broadcast
+     * and announces a new one; hang-listen's audio subscription
+     * dies and it doesn't re-subscribe. So the listener captures
+     * only the pre-hot-swap chunk (~2.5 s of 5 s).
+     *
+     * The Amethyst production listener handles re-announce
+     * transparently (`ReconnectingNestsListener.kt`), so the
+     * "no audio gap" assertion the spec calls for is a property
+     * of the Amethyst path, not the hang-listen path. This test
+     * therefore asserts only:
+     *
+     *   - the speaker survives the hot-swap (no crash, got some
+     *     audio),
+     *   - the FFT peak on the captured pre-swap chunk is still
+     *     at 440 Hz (the swap doesn't corrupt active uni
+     *     streams).
+     *
+     * The "no audible gap" assertion belongs to a Phase 3
+     * follow-up that exercises this scenario through the
+     * Amethyst Kotlin LISTENER instead of `hang-listen`.
+     */
+    @Test
+    fun speaker_hot_swap_does_not_crash() =
+        runBlocking {
+            val out =
+                runSpeakerToHangListen(
+                    speakerSeconds = 5,
+                    captureFirstFrame = false,
+                    hotSwapAfterMs = 2_500L,
+                )
+            val pcm = readFloat32Pcm(out.pcmFile)
+            // Got SOMETHING (≥ 1 s of audio) — speaker didn't
+            // crash on the hot-swap.
+            assertTrue(
+                pcm.size >= AudioFormat.SAMPLE_RATE_HZ,
+                "expected ≥ 1 s of audio across the hot-swap, got ${pcm.size} samples",
+            )
+            // The captured chunk's tone is still 440 Hz —
+            // spectral integrity intact.
+            val warmup = AudioFormat.SAMPLE_RATE_HZ / 25
+            val analysed = pcm.copyOfRange(warmup, pcm.size)
+            PcmAssertions.assertFftPeak(analysed, expectedHz = 440.0, halfWindowHz = 5.0)
+        }
+
     /**
      * Rust↔Rust round-trip: pure-Rust through our harness.
      * Validates the cargo workspace + relay config + `moq-lite-03`
@@ -478,15 +561,67 @@ private suspend fun runSpeakerToHangListen(
     listenerLateJoinDelayMs: Long = 150L,
     muteWindowMs: ClosedRange<Long>? = null,
     captureFirstFrame: Boolean,
+    /**
+     * If non-null, route the Kotlin speaker's UDP through a
+     * `udp-loss-shim` subprocess that drops this fraction of
+     * datagrams (0.0..=1.0). The shim listens on a fresh
+     * ephemeral port and forwards to the harness's relay; the
+     * speaker's `endpoint` is rewritten to the shim port. The
+     * hang-listen subprocess still connects directly to the
+     * relay (no loss), so any frame deficit is attributable to
+     * the speaker→relay leg.
+     */
+    udpLossRate: Float? = null,
+    /**
+     * If non-null, drive the speaker through
+     * `connectReconnectingNestsSpeaker` with this
+     * `tokenRefreshAfterMs`, forcing a session recycle (hot-swap)
+     * mid-broadcast. Default uses the simple non-reconnecting
+     * speaker — the I1/I2/I3/I8/I10/I11 scenarios don't need
+     * the reconnect orchestrator.
+     */
+    hotSwapAfterMs: Long? = null,
 ): HangListenOutput {
     val harness = NativeMoqRelayHarness.shared()
 
     val signer: NostrSigner = NostrSignerInternal(KeyPair())
     val pubkey = signer.pubKey
+
+    // If a loss rate is requested, spin up a udp-loss-shim
+    // between the speaker and the relay. The speaker connects
+    // through the shim (lossy); hang-listen still connects to
+    // the relay directly so any frame deficit is attributable
+    // to the lossy leg.
+    val (relayHostForSpeaker, relayPortForSpeaker, lossShimProc) =
+        if (udpLossRate != null) {
+            val shimPort = java.net.ServerSocket(0).use { it.localPort }
+            val (relayHost, relayPort) = harness.loopbackHostPort()
+            val proc =
+                ProcessBuilder(
+                    harness.udpLossShimBin().toString(),
+                    "--listen",
+                    "127.0.0.1:$shimPort",
+                    "--upstream",
+                    "$relayHost:$relayPort",
+                    "--loss-rate",
+                    udpLossRate.toString(),
+                ).redirectErrorStream(true)
+                    .also { it.environment()["RUST_LOG"] = "info" }
+                    .start()
+            // Tiny breathing room for the shim's listen socket
+            // to bind before the speaker's QUIC handshake hits.
+            Thread.sleep(200)
+            Triple("127.0.0.1", shimPort, proc)
+        } else {
+            val (h, p) = harness.loopbackHostPort()
+            Triple(h, p, null)
+        }
+    val speakerEndpoint = "https://$relayHostForSpeaker:$relayPortForSpeaker"
+
     val room =
         NestsRoomConfig(
             authBaseUrl = "<unused-public-relay>",
-            endpoint = harness.relayUrl,
+            endpoint = speakerEndpoint,
             hostPubkey = pubkey,
             roomId = "rt-${UUID.randomUUID()}",
         )
@@ -518,17 +653,44 @@ private suspend fun runSpeakerToHangListen(
     lateinit var listenProc: Process
     try {
         val speaker =
-            connectNestsSpeaker(
+            if (hotSwapAfterMs != null) {
-                httpClient = StaticTokenNestsClient,
+                connectReconnectingNestsSpeaker(
-                transport = transport,
+                    httpClient = StaticTokenNestsClient,
-                scope = pumpScope,
+                    transport = transport,
-                room = room,
+                    scope = pumpScope,
-                signer = signer,
+                    room = room,
-                speakerPubkeyHex = pubkey,
+                    signer = signer,
-                captureFactory = { SineWaveAudioCapture(freqHz = 440) },
+                    speakerPubkeyHex = pubkey,
-                encoderFactory = { JvmOpusEncoder() },
+                    captureFactory = { SineWaveAudioCapture(freqHz = 440) },
-                framesPerGroup = 5,
+                    encoderFactory = { JvmOpusEncoder() },
-            )
+                    tokenRefreshAfterMs = hotSwapAfterMs,
+                    connector = {
+                        connectNestsSpeaker(
+                            httpClient = StaticTokenNestsClient,
+                            transport = transport,
+                            scope = pumpScope,
+                            room = room,
+                            signer = signer,
+                            speakerPubkeyHex = pubkey,
+                            captureFactory = { SineWaveAudioCapture(freqHz = 440) },
+                            encoderFactory = { JvmOpusEncoder() },
+                            framesPerGroup = 5,
+                        )
+                    },
+                )
+            } else {
+                connectNestsSpeaker(
+                    httpClient = StaticTokenNestsClient,
+                    transport = transport,
+                    scope = pumpScope,
+                    room = room,
+                    signer = signer,
+                    speakerPubkeyHex = pubkey,
+                    captureFactory = { SineWaveAudioCapture(freqHz = 440) },
+                    encoderFactory = { JvmOpusEncoder() },
+                    framesPerGroup = 5,
+                )
+            }
         val handle = speaker.startBroadcasting()
         delay(listenerLateJoinDelayMs)
 
@@ -575,6 +737,7 @@ private suspend fun runSpeakerToHangListen(
         speaker.close()
     } finally {
         pumpScope.coroutineContext[Job]?.cancel()
+        lossShimProc?.destroy()
     }
 
     val exited = listenProc.waitFor(15, TimeUnit.SECONDS)