chore(nests): mv cli/hang-interop → nestsClient/tests/hang-interop

Per maintainer request: the Rust sidecar workspace lives
under the module that owns it, parallel to the upcoming
nestsClient-browser-interop/ harness. The cargo workspace,
Gradle wiring, gitignore, CI YAML, plan docs, and harness
kdoc are all updated. cargo build --release still compiles;
HangInteropTest.amethyst_speaker_to_hang_listener_stereo_440_660
green at the new path.

Note: the live Phase 4 browser-harness agent (worktree
agent-a97a6be483ecee618) was branched from before this move
and references `cli/hang-interop` in its plan-doc imports.
It will need to rebase onto this commit before it pushes;
no code-level conflicts since the agent works exclusively
in nestsClient-browser-interop/ + a new
BrowserInteropTest.kt.

https://claude.ai/code/session_01ERJPUYfdLPwZ99pr5EcEcV

nestsClient/tests/hang-interop/Cargo.toml

@@ -0,0 +1,26 @@
+[workspace]
+resolver = "2"
+members = [
+    "hang-listen",
+    "hang-publish",
+    "udp-loss-shim",
+]
+
+# Phase 1 ships these as stub binaries that compile cleanly but do
+# nothing. Phase 2 fills in real implementations against `hang` /
+# `moq-lite` / `web-transport-quinn`. Versions tracked in REV.
+[workspace.package]
+version = "0.0.1"
+edition = "2024"
+publish = false
+license = "MIT OR Apache-2.0"
+
+[workspace.dependencies]
+anyhow = "1"
+clap = { version = "4", features = ["derive"] }
+tokio = { version = "1", features = ["full"] }
+
+[profile.release]
+opt-level = 3
+lto = "thin"
+strip = true

nestsClient/tests/hang-interop/REV

@@ -0,0 +1,22 @@
+# Pinned upstream revisions for the cross-stack interop harness.
+#
+# These versions are what NativeMoqRelayHarness installs (via
+# `cargo install --version <v> --root <cache>`) and what our sidecar
+# crates pin against in Cargo.toml. Bump deliberately — a silent
+# upstream rev change can mask a regression.
+#
+# See: nestsClient/plans/2026-05-06-cross-stack-interop-test.md
+
+# https://github.com/kixelated/moq commit at the time this harness was
+# written. Used as the source-of-truth for the exact API shapes the
+# sidecar crates are coded against. Phase 1 doesn't compile against
+# upstream yet (sidecars are stubs); Phase 2 will pin the published
+# crate versions below to track this rev.
+KIXELATED_MOQ_GIT_REV=9e2461ee4941968f7b8c410e472448639d2aa4a3
+
+# Published crate versions on crates.io that come from the rev above.
+MOQ_RELAY_VERSION=0.10.25
+MOQ_TOKEN_CLI_VERSION=0.5.23
+HANG_VERSION=0.15.8
+MOQ_LITE_VERSION=0.15.15
+MOQ_NATIVE_VERSION=0.13

nestsClient/tests/hang-interop/hang-listen/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "hang-listen"
+version.workspace = true
+edition.workspace = true
+publish.workspace = true
+license.workspace = true
+
+# Real subscribe/decode body: connects to a moq-lite-03 relay, reads
+# the hang catalog, picks the first Opus / Container::Legacy audio
+# rendition, and writes Float32 PCM to --output-pcm. Used by the
+# cross-stack interop tests in nestsClient/src/jvmTest/.../interop/native/.
+
+[[bin]]
+name = "hang-listen"
+path = "src/main.rs"
+
+[dependencies]
+anyhow.workspace = true
+clap.workspace = true
+tokio.workspace = true
+hang = "0.15"
+moq-lite = "0.15"
+moq-mux = "0.3"
+moq-native = { version = "0.13", default-features = false, features = ["quinn", "aws-lc-rs"] }
+opus = "0.3"
+rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+url = "2"

nestsClient/tests/hang-interop/hang-listen/src/main.rs

@@ -0,0 +1,301 @@
+//! hang-listen — reference moq-lite / hang audio listener.
+//!
+//! Used by the Amethyst cross-stack interop test harness to verify
+//! that an Amethyst Kotlin speaker is intelligible to the canonical
+//! `kixelated/moq` listener stack. See
+//! `nestsClient/plans/2026-05-06-cross-stack-interop-test.md`.
+//!
+//! Wire path:
+//!   1. Connect via `web-transport-quinn` over QUIC.
+//!   2. Open a `moq-lite-03` session (via moq_native::ClientConfig).
+//!   3. Subscribe to the hang Catalog track at `<broadcast>/catalog.json`.
+//!   4. Pick the first audio rendition with `codec="opus"` and
+//!      `container.kind="legacy"`.
+//!   5. Subscribe to that rendition's track via
+//!      `moq_mux::container::Consumer<Hang::Legacy>`.
+//!   6. For each frame: decode Opus → Float32 PCM, write to stdout
+//!      or `--output-pcm` as raw little-endian f32s.
+
+use std::io::Write;
+use std::path::PathBuf;
+use std::time::Duration;
+
+use anyhow::{Context, anyhow};
+use clap::Parser;
+use hang::catalog::{AudioCodec, Container};
+
+const SAMPLE_RATE_HZ: u32 = 48_000;
+/// 120 ms at 48 kHz — Opus's worst-case frame size; pre-allocate
+/// once and let `Decoder::decode` write what it actually decoded.
+const MAX_PCM_PER_PACKET: usize = (SAMPLE_RATE_HZ as usize) / 1000 * 120;
+
+#[derive(Parser, Debug)]
+#[command(
+    name = "hang-listen",
+    about = "Reference moq-lite / hang audio listener for cross-stack interop"
+)]
+struct Args {
+    /// HTTPS URL of the relay, e.g. `https://127.0.0.1:34721`.
+    #[arg(long)]
+    relay_url: String,
+
+    /// Optional JWT for the `?jwt=` query string. The Amethyst test
+    /// harness configures the relay with `--auth-public ""`, in which
+    /// case this can be omitted.
+    #[arg(long)]
+    jwt: Option<String>,
+
+    /// Broadcast namespace. The full path is `<broadcast>/<track>`.
+    #[arg(long)]
+    broadcast: String,
+
+    /// Maximum runtime in seconds.
+    #[arg(long, default_value_t = 5)]
+    duration: u64,
+
+    /// Output Float32 little-endian PCM here. Use `-` for stdout.
+    /// If absent, the binary discards PCM (used as a smoke test).
+    #[arg(long)]
+    output_pcm: Option<String>,
+
+    /// Dump the first audio frame's raw bytes (the post-Hang::Legacy
+    /// payload — already stripped of the moq-lite frame size prefix
+    /// but NOT the hang VarInt timestamp prefix) to this path. Used
+    /// by I11 to assert the publisher isn't shipping
+    /// `OpusHead\\1\\1...` Codec-Specific-Data as the first audio
+    /// frame (the T8 regression in the audit branch).
+    #[arg(long)]
+    dump_first_frame: Option<String>,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    // rustls 0.23 requires an explicit crypto-provider install.
+    // Mirror moq-relay's main.rs choice (aws-lc-rs).
+    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+
+    // Init logger early so config / handshake errors surface.
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .with_writer(std::io::stderr)
+        .try_init();
+
+    let args = Args::parse();
+
+    let result = tokio::time::timeout(
+        Duration::from_secs(args.duration + 5),
+        run(args),
+    )
+    .await
+    .context("hang-listen wallclock timeout")?;
+
+    result
+}
+
+async fn run(args: Args) -> anyhow::Result<()> {
+    let url = build_url(&args.relay_url, &args.broadcast, args.jwt.as_deref())?;
+
+    // moq-lite-03 ALPN, IPv4 client bind (sandbox friendly), and TLS
+    // verification disabled so the test harness's --tls-generate
+    // cert chain works without a custom truststore.
+    let cfg = moq_native::ClientConfig::parse_from([
+        "hang-listen",
+        "--client-bind",
+        "127.0.0.1:0",
+        "--client-version",
+        "moq-lite-03",
+        "--tls-disable-verify=true",
+    ]);
+    let client = cfg.init().context("init moq client")?;
+
+    // Set up an Origin so the session can publish incoming
+    // broadcasts to us, then drive both the session and the
+    // subscribe loop in parallel.
+    let origin = moq_lite::Origin::produce();
+    let consumer = origin.consume();
+
+    let session_url = url.clone();
+    let session = tokio::spawn(async move {
+        // Use reconnect() with a tight timeout so the test exits
+        // quickly when the relay drops us. closed() returns when the
+        // backoff loop finally gives up.
+        let reconnect = client.with_consume(origin).reconnect(session_url);
+        if let Err(err) = reconnect.closed().await {
+            tracing::warn!(%err, "reconnect loop exited");
+        }
+    });
+
+    let listen_result = listen(
+        consumer,
+        args.output_pcm.as_deref(),
+        args.dump_first_frame.as_deref(),
+        args.duration,
+    )
+    .await;
+
+    // The session task will exit on its own when the URL closes; we
+    // don't need to abort it for a clean shutdown.
+    drop(session);
+
+    listen_result
+}
+
+async fn listen(
+    mut origin: moq_lite::OriginConsumer,
+    output_pcm: Option<&str>,
+    output_dump_first_frame: Option<&str>,
+    duration_sec: u64,
+) -> anyhow::Result<()> {
+    // Open the PCM sink up front so we fail fast on a bad path.
+    let mut pcm: Box<dyn Write + Send> = match output_pcm {
+        Some("-") => Box::new(std::io::stdout()),
+        Some(path) => Box::new(
+            std::fs::File::create(PathBuf::from(path))
+                .with_context(|| format!("create output-pcm file '{path}'"))?,
+        ),
+        None => Box::new(std::io::sink()),
+    };
+
+    // Wait for the broadcast to be announced. The relay forwards
+    // any matching broadcast across the configured namespace.
+    let (path, broadcast) = origin
+        .announced()
+        .await
+        .ok_or_else(|| anyhow!("origin closed before any broadcast announced"))?;
+    let broadcast = broadcast.ok_or_else(|| anyhow!("broadcast unannounced: {path}"))?;
+    tracing::info!(%path, "broadcast announced");
+
+    // Subscribe to the catalog and read the first published version.
+    let catalog_track = broadcast
+        .subscribe_track(&hang::Catalog::default_track())
+        .context("subscribe catalog")?;
+    let mut catalog = hang::CatalogConsumer::new(catalog_track);
+
+    let info = catalog
+        .next()
+        .await
+        .context("read catalog")?
+        .ok_or_else(|| anyhow!("catalog ended before first publish"))?;
+
+    // Pick the first Opus / Container::Legacy audio rendition.
+    let (track_name, audio_cfg) = info
+        .audio
+        .renditions
+        .iter()
+        .find(|(_, cfg)| matches!(cfg.codec, AudioCodec::Opus) && cfg.container == Container::Legacy)
+        .ok_or_else(|| {
+            anyhow!(
+                "no audio rendition with codec=opus container.kind=legacy in catalog: {:?}",
+                info.audio.renditions.keys().collect::<Vec<_>>()
+            )
+        })?;
+
+    // Audio renditions advertise `numberOfChannels` (1 for mono, 2 for
+    // stereo). nests speakers send mono; stereo is exercised by I4.
+    let channels = match audio_cfg.channel_count {
+        1 => opus::Channels::Mono,
+        2 => opus::Channels::Stereo,
+        n => anyhow::bail!("unsupported channel count: {n}"),
+    };
+    tracing::info!(
+        track = %track_name,
+        sample_rate = audio_cfg.sample_rate,
+        channels = audio_cfg.channel_count,
+        "subscribing to audio rendition"
+    );
+
+    let track = moq_lite::Track {
+        name: track_name.clone(),
+        priority: 1,
+    };
+    let track_consumer = broadcast.subscribe_track(&track).context("subscribe audio")?;
+    let mut frames = moq_mux::hang::Consumer::new(track_consumer, moq_mux::hang::Legacy)
+        // Zero latency = aggressive skip. We prefer a more forgiving
+        // budget so jitter doesn't drop frames in tests.
+        .with_latency(Duration::from_millis(500));
+
+    let mut decoder =
+        opus::Decoder::new(audio_cfg.sample_rate, channels).context("init opus decoder")?;
+    let mut pcm_buf = vec![0i16; MAX_PCM_PER_PACKET * audio_cfg.channel_count as usize];
+
+    let dump_first_frame_path = output_dump_first_frame.map(PathBuf::from);
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(duration_sec);
+    let mut total_samples: u64 = 0;
+    let mut frame_count: u64 = 0;
+    loop {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        if remaining.is_zero() {
+            break;
+        }
+        let frame = match tokio::time::timeout(remaining, frames.read()).await {
+            Ok(Ok(Some(f))) => f,
+            Ok(Ok(None)) => {
+                tracing::info!("track ended");
+                break;
+            }
+            Ok(Err(e)) => {
+                // A "cancelled" tail-error after we've already
+                // collected frames is just the publisher closing
+                // its side of the broadcast — treat it as a
+                // normal end-of-stream rather than failing the
+                // whole run. Test scripts assert against the PCM
+                // file size + content, not the exit code's
+                // distinction between graceful-end and
+                // publisher-cancel.
+                if frame_count > 0 {
+                    tracing::info!(error = %e, "track cancelled after {frame_count} frames; treating as EOF");
+                    break;
+                }
+                return Err(anyhow::Error::new(e).context("read audio frame"));
+            }
+            Err(_) => {
+                tracing::info!("duration elapsed");
+                break;
+            }
+        };
+
+        // First-frame capture for I11. payload is the post-
+        // Container::Legacy-strip codec payload (i.e. the raw
+        // Opus packet, no timestamp prefix). If the publisher
+        // accidentally ships `OpusHead\1\1...` Codec-Specific-Data
+        // as the first audio frame, this is where it shows up.
+        if frame_count == 0 {
+            if let Some(path) = dump_first_frame_path.as_ref() {
+                std::fs::write(path, frame.payload.as_ref())
+                    .with_context(|| format!("write dump-first-frame to '{}'", path.display()))?;
+            }
+        }
+
+        // payload is the raw Opus packet — the timestamp varint has
+        // already been stripped by `Hang::Legacy` decoding.
+        let n = decoder
+            .decode(&frame.payload, &mut pcm_buf, false)
+            .with_context(|| format!("decode opus packet ({} bytes)", frame.payload.len()))?;
+
+        // n is the number of samples per channel; total interleaved
+        // samples in pcm_buf is n * channels.
+        let interleaved = n * audio_cfg.channel_count as usize;
+        for s in &pcm_buf[..interleaved] {
+            // i16 → f32 in [-1, 1].
+            let f = (*s as f32) / 32_768.0;
+            pcm.write_all(&f.to_le_bytes())
+                .context("write pcm sample")?;
+        }
+        total_samples += interleaved as u64;
+        frame_count += 1;
+    }
+
+    pcm.flush().ok();
+    tracing::info!(frames = frame_count, samples = total_samples, "hang-listen done");
+    Ok(())
+}
+
+fn build_url(relay_url: &str, broadcast: &str, jwt: Option<&str>) -> anyhow::Result<url::Url> {
+    let trimmed = relay_url.trim_end_matches('/');
+    let raw = if let Some(jwt) = jwt {
+        format!("{trimmed}/{broadcast}?jwt={jwt}")
+    } else {
+        format!("{trimmed}/{broadcast}")
+    };
+    url::Url::parse(&raw).with_context(|| format!("malformed relay/broadcast url: {raw}"))
+}

nestsClient/tests/hang-interop/hang-publish/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "hang-publish"
+version.workspace = true
+edition.workspace = true
+publish.workspace = true
+license.workspace = true
+
+# Reference moq-lite / hang publisher: opens a broadcast, declares one
+# Opus / Container::Legacy audio rendition, encodes a sine wave, and
+# pumps frames in 5-frame groups for `--duration` seconds. Used by
+# the cross-stack interop tests for the Rust → Amethyst direction.
+
+[[bin]]
+name = "hang-publish"
+path = "src/main.rs"
+
+[dependencies]
+anyhow.workspace = true
+clap.workspace = true
+tokio.workspace = true
+hang = "0.15"
+moq-lite = "0.15"
+moq-native = { version = "0.13", default-features = false, features = ["quinn", "aws-lc-rs"] }
+opus = "0.3"
+bytes = "1"
+rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs"] }
+serde_json = "1"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+url = "2"

nestsClient/tests/hang-interop/hang-publish/src/main.rs

@@ -0,0 +1,311 @@
+//! hang-publish — reference moq-lite / hang audio publisher.
+//!
+//! Opens a broadcast at `<relay>/<broadcast>`, publishes a hang
+//! catalog with one Opus / Container::Legacy audio rendition, and
+//! pumps Opus-encoded sine-wave frames in groups of 5 for
+//! `--duration` seconds. Used by the cross-stack interop tests for
+//! the Rust → Amethyst direction. See
+//! `nestsClient/plans/2026-05-06-cross-stack-interop-test.md`.
+
+use std::time::Duration;
+
+use anyhow::{Context, anyhow};
+use bytes::Bytes;
+use clap::Parser;
+use hang::catalog::{Audio, AudioCodec, AudioConfig, Catalog, Container};
+
+const SAMPLE_RATE_HZ: u32 = 48_000;
+/// 20 ms at 48 kHz — same frame size Amethyst speakers send.
+const FRAME_SIZE_SAMPLES: usize = 960;
+/// Microseconds per frame: 20_000 = 1_000_000 * 960 / 48_000.
+const FRAME_DURATION_US: u64 = 20_000;
+/// 5 frames per group → 100 ms group cadence, matching nests speaker.
+const FRAMES_PER_GROUP: usize = 5;
+/// Default audio rendition track name in the catalog. Amethyst's
+/// listener subscribes to `audio/data` per `MoqLiteNestsListener.AUDIO_TRACK`,
+/// so that's what we ship by default. Override via `--track-name`.
+const DEFAULT_TRACK_NAME: &str = "audio/data";
+
+#[derive(Parser, Debug)]
+#[command(
+    name = "hang-publish",
+    about = "Reference moq-lite / hang audio publisher for cross-stack interop"
+)]
+struct Args {
+    /// HTTPS URL of the relay, e.g. `https://127.0.0.1:34721`.
+    #[arg(long)]
+    relay_url: String,
+
+    /// Optional JWT for the `?jwt=` query string.
+    #[arg(long)]
+    jwt: Option<String>,
+
+    /// Broadcast namespace (path under the relay root).
+    #[arg(long)]
+    broadcast: String,
+
+    /// Sine-wave frequency in Hz. Used as the default for every
+    /// channel; override per-channel via `--freq-hz-l` / `--freq-hz-r`.
+    #[arg(long, default_value_t = 440)]
+    freq_hz: u32,
+
+    /// Per-channel frequency override for the LEFT channel. Falls
+    /// back to `--freq-hz` when unset.
+    #[arg(long)]
+    freq_hz_l: Option<u32>,
+
+    /// Per-channel frequency override for the RIGHT channel.
+    /// Ignored when `--channels 1`. Falls back to `--freq-hz` when
+    /// unset.
+    #[arg(long)]
+    freq_hz_r: Option<u32>,
+
+    /// Maximum runtime in seconds.
+    #[arg(long, default_value_t = 5)]
+    duration: u64,
+
+    /// Channel count: 1 (mono) or 2 (stereo). With `2` and
+    /// `--freq-hz-l` / `--freq-hz-r` set, the L/R channels carry
+    /// independent tones — useful for the I4 stereo cross-stack
+    /// scenario.
+    #[arg(long, default_value_t = 1)]
+    channels: u32,
+
+    /// Audio rendition track name. Default `audio/data` matches
+    /// Amethyst's `MoqLiteNestsListener.AUDIO_TRACK`. Override for
+    /// custom interop scenarios.
+    #[arg(long, default_value_t = DEFAULT_TRACK_NAME.to_string())]
+    track_name: String,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    // rustls 0.23 requires an explicit crypto-provider install.
+    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .with_writer(std::io::stderr)
+        .try_init();
+
+    let args = Args::parse();
+
+    let result = tokio::time::timeout(
+        Duration::from_secs(args.duration + 5),
+        run(args),
+    )
+    .await
+    .context("hang-publish wallclock timeout")?;
+
+    result
+}
+
+async fn run(args: Args) -> anyhow::Result<()> {
+    let url = build_url(&args.relay_url, args.jwt.as_deref())?;
+
+    let cfg = moq_native::ClientConfig::parse_from([
+        "hang-publish",
+        "--client-bind",
+        "127.0.0.1:0",
+        "--client-version",
+        "moq-lite-03",
+        "--tls-disable-verify=true",
+    ]);
+    let client = cfg.init().context("init moq client")?;
+
+    // Set up the publish side: a producer this binary writes to,
+    // and a consumer the moq-native client forwards to the relay.
+    let origin = moq_lite::Origin::produce();
+    let publish_consumer = origin.consume();
+
+    // Start the reconnect loop in the background. It owns the
+    // session lifecycle.
+    let session_url = url.clone();
+    let session = tokio::spawn(async move {
+        let reconnect = client.with_publish(publish_consumer).reconnect(session_url);
+        if let Err(err) = reconnect.closed().await {
+            tracing::warn!(%err, "reconnect loop exited");
+        }
+    });
+
+    // Result of the publish loop is what determines test pass/fail.
+    let publish_result = publish(&origin, &args).await;
+
+    // Once we drop origin all published broadcasts unannounce; the
+    // reconnect task exits when the session closes.
+    drop(session);
+
+    publish_result
+}
+
+async fn publish(origin: &moq_lite::OriginProducer, args: &Args) -> anyhow::Result<()> {
+    let mut broadcast = origin
+        .create_broadcast(args.broadcast.as_str())
+        .ok_or_else(|| anyhow!("broadcast '{}' not allowed by origin", args.broadcast))?;
+
+    // 1. Catalog track. We declare one Opus rendition; the JSON
+    //    payload mirrors what Amethyst's MoqLiteHangCatalog produces.
+    let mut catalog_track = broadcast
+        .create_track(hang::Catalog::default_track())
+        .context("create catalog track")?;
+
+    let mut renditions = std::collections::BTreeMap::new();
+    renditions.insert(
+        args.track_name.clone(),
+        AudioConfig {
+            codec: AudioCodec::Opus,
+            sample_rate: SAMPLE_RATE_HZ,
+            channel_count: args.channels,
+            bitrate: Some(32_000),
+            description: None,
+            container: Container::Legacy,
+            jitter: None,
+        },
+    );
+    let catalog = Catalog {
+        audio: Audio { renditions },
+        ..Default::default()
+    };
+    let catalog_json =
+        serde_json::to_vec(&catalog).context("serialize catalog json")?;
+
+    let mut catalog_group = catalog_track
+        .create_group(moq_lite::Group { sequence: 0 })
+        .context("create catalog group")?;
+    catalog_group
+        .write_frame(catalog_json)
+        .context("publish catalog frame")?;
+    catalog_group.finish().ok();
+    // We don't finish() the catalog_track itself yet — moq-lite
+    // treats track-end as broadcast-end, and we want the audio
+    // track to keep streaming.
+
+    // 2. Audio track.
+    let mut audio_track = broadcast
+        .create_track(moq_lite::Track {
+            name: args.track_name.clone(),
+            priority: 1,
+        })
+        .context("create audio track")?;
+
+    let channels = match args.channels {
+        1 => opus::Channels::Mono,
+        2 => opus::Channels::Stereo,
+        n => anyhow::bail!("unsupported channel count: {n}"),
+    };
+    let mut encoder = opus::Encoder::new(SAMPLE_RATE_HZ, channels, opus::Application::Audio)
+        .context("init opus encoder")?;
+    encoder
+        .set_bitrate(opus::Bitrate::Bits(32_000))
+        .context("set opus bitrate")?;
+
+    let total_frames = (args.duration * 1_000_000 / FRAME_DURATION_US) as usize;
+    // Per-channel phase step: each channel may have its own
+    // frequency (I4 stereo). Defaults to args.freq_hz on every
+    // channel.
+    let mut phase_steps: Vec<f64> = Vec::with_capacity(args.channels as usize);
+    for ch in 0..(args.channels as usize) {
+        let f = match (ch, args.freq_hz_l, args.freq_hz_r) {
+            (0, Some(l), _) => l,
+            (1, _, Some(r)) => r,
+            _ => args.freq_hz,
+        };
+        phase_steps
+            .push(2.0_f64 * std::f64::consts::PI * (f as f64) / (SAMPLE_RATE_HZ as f64));
+    }
+    let mut sample_idx: u64 = 0;
+    // Sized to libopus's worst-case output for one 20 ms frame.
+    let mut opus_buf = vec![0u8; 4_000];
+
+    let frame_period = Duration::from_micros(FRAME_DURATION_US);
+    let mut next_send = tokio::time::Instant::now();
+    let mut group_idx: u64 = 0;
+    let mut frames_in_group = 0usize;
+    let mut group: Option<moq_lite::GroupProducer> = None;
+
+    for frame_no in 0..total_frames {
+        // Generate one PCM frame, possibly with a different sine
+        // tone on each channel.
+        let mut pcm = vec![0i16; FRAME_SIZE_SAMPLES * args.channels as usize];
+        for i in 0..FRAME_SIZE_SAMPLES {
+            let t = (sample_idx + i as u64) as f64;
+            for ch in 0..(args.channels as usize) {
+                let v = (t * phase_steps[ch]).sin();
+                let s = (v * 16_383.0) as i16;
+                pcm[i * args.channels as usize + ch] = s;
+            }
+        }
+        sample_idx += FRAME_SIZE_SAMPLES as u64;
+
+        let n = encoder
+            .encode(&pcm, &mut opus_buf)
+            .context("encode opus packet")?;
+        let opus_packet = Bytes::copy_from_slice(&opus_buf[..n]);
+
+        // Wrap the Opus packet in a hang Legacy frame: VarInt
+        // timestamp prefix + raw codec payload.
+        let frame = hang::container::Frame {
+            timestamp: hang::container::Timestamp::from_micros(
+                (frame_no as u64) * FRAME_DURATION_US,
+            )
+            .context("frame timestamp out of range")?,
+            payload: opus_packet.into(),
+        };
+
+        // Start a new group every FRAMES_PER_GROUP frames. The 5-frame
+        // group cadence matches Amethyst's NestMoqLiteBroadcaster default
+        // and produces ~100 ms groups.
+        if frames_in_group == 0 {
+            if let Some(mut g) = group.take() {
+                g.finish().ok();
+            }
+            group = Some(
+                audio_track
+                    .create_group(moq_lite::Group { sequence: group_idx })
+                    .context("create audio group")?,
+            );
+            group_idx += 1;
+        }
+
+        let g = group.as_mut().expect("group always Some after init");
+        frame.encode(g).context("encode hang frame into group")?;
+        frames_in_group += 1;
+        if frames_in_group == FRAMES_PER_GROUP {
+            frames_in_group = 0;
+        }
+
+        next_send += frame_period;
+        tokio::time::sleep_until(next_send).await;
+    }
+
+    if let Some(mut g) = group.take() {
+        g.finish().ok();
+    }
+    audio_track.finish().ok();
+    catalog_track.finish().ok();
+
+    tracing::info!(
+        frames = total_frames,
+        groups = group_idx,
+        "hang-publish done"
+    );
+    Ok(())
+}
+
+/// Build the WebTransport URL the publisher connects to.
+///
+/// `relay_url` is taken as the *full* URL the publisher connects to
+/// (scheme + authority + optional path). `broadcast` is the relative
+/// announce-suffix passed to `Origin::create_broadcast`, NOT appended
+/// to the URL. Callers that want the publisher's URL path to also be
+/// `broadcast` should pass `--relay-url=<host>/<broadcast>` and
+/// `--broadcast=<broadcast>` (the simple Rust↔Rust shape).
+fn build_url(relay_url: &str, jwt: Option<&str>) -> anyhow::Result<url::Url> {
+    let trimmed = relay_url.trim_end_matches('/');
+    let raw = if let Some(jwt) = jwt {
+        format!("{trimmed}?jwt={jwt}")
+    } else {
+        trimmed.to_string()
+    };
+    url::Url::parse(&raw).with_context(|| format!("malformed relay url: {raw}"))
+}

nestsClient/tests/hang-interop/udp-loss-shim/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "udp-loss-shim"
+version.workspace = true
+edition.workspace = true
+publish.workspace = true
+license.workspace = true
+
+# 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"
+path = "src/main.rs"
+
+[dependencies]
+anyhow.workspace = true
+clap.workspace = true
+tokio.workspace = true
+rand = "0.8"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }

nestsClient/tests/hang-interop/udp-loss-shim/src/main.rs

@@ -0,0 +1,152 @@
+//! udp-loss-shim — UDP loopback that drops a configurable fraction
+//! 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 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 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();
+    anyhow::ensure!(
+        (0.0..=1.0).contains(&args.loss_rate),
+        "loss-rate must be in 0.0..=1.0, got {}",
+        args.loss_rate
+    );
+
+    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");
+        }
+    }
+}