feat(neswire): open the audio server

Captures a session's audio and hands it to the transport over a local socket.
Third component in, imported as a tree from `nestrilabs/neswire` on the same
terms as the previous two.

Wired to the workspace, `nesprotocol` by path. 4 tests pass.

`bin/hub-stub.rs` is a stand-in for the transport's listener, which is what lets
this be developed and tested without the rest of a box existing. It names the
transport by its old name, and is left for the rename commit along with the two
in the compositor.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

apps/neswire/.gitignore

@@ -0,0 +1,45 @@
+```
+# Rust/Cargo specific
+/target/
+Cargo.lock
+**/Cargo.lock
+.cargo/
+cargo_home/
+
+# Git related
+.git/
+
+# Temporary files
+*.tmp
+*.swp
+*~
+
+# Logs
+*.log
+
+# OS generated files
+.DS_Store
+Thumbs.db
+
+# Environment files
+.env
+.env.local
+*.env.*
+
+# IDE files
+.vscode/
+.idea/
+*.swp
+*.swo
+
+# Build artifacts
+build/
+dist/
+*.o
+*.obj
+*.exe
+*.dll
+*.so
+*.a
+*.out
+```

apps/neswire/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "neswire"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+pipewire = "0.10"
+crossbeam-channel = "0.5"
+anyhow = "1"
+opus-head-sys = "0.3"
+bytemuck = { version = "1", features = ["extern_crate_alloc"] }
+clap = { version = "4", features = ["derive", "env"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+nesprotocol = { path = "../../crates/nesprotocol" }

apps/neswire/README.md

@@ -0,0 +1,36 @@
+## neswire
+
+A small custom PipeWire sink for cloud gaming audio capture.
+
+Currently for debugging uses RTP to send Opus (with FEC enabled by default) over to target address.
+
+
+### Testing
+
+#### Mono/Stereo
+
+Launch gstreamer pipeline to receive audio as so (will save incoming RTP audio into test.mkv):
+```bash
+gst-launch-1.0 udpsrc port=12345 caps="application/x-rtp,media=audio,encoding-name=OPUS,clock-rate=48000,payload=111" ! rtpopusdepay2 ! opusdec ! matroskamux ! filesink location=test.mkv sync=false
+```
+
+Then run neswire like so for example:
+```bash
+cargo run --release --bin neswire -- --rtp-addr 127.0.0.1:12345
+```
+
+#### Surround
+
+Launch gstreamer pipeline to receive audio as so (will save incoming RTP audio into test_multi.mkv):
+```bash
+gst-launch-1.0 udpsrc port=12345 caps='application/x-rtp,media=audio,encoding-name=MULTIOPUS,clock-rate=48000,payload=111,encoding-params=(string)8,num_streams=(string)5,coupled_streams=(string)3,channel_mapping=(string)"0,6,1,2,3,4,5,7"' ! rtpopusdepay2 ! opusdec ! matroskamux ! filesink location=test.mkv sync=false
+```
+
+Then run neswire like so for example:
+```bash
+cargo run --release --bin neswire -- --rtp-addr 127.0.0.1:12345 --channels 8
+```
+
+
+For both afterwards, set the neswire sink as audio output source in your system (or specify it as output in some app/game),
+ then play some audio, stop gst pipeline and sink, listen to results after.

apps/neswire/src/bin/hub-stub.rs

@@ -0,0 +1,205 @@
+//! A stand-in for nestri-guest-hub's audio IPC listener.
+//!
+//! neswire's only output is a Unix datagram socket that the hub binds, so
+//! running it outside a guest means having nothing to talk to: it retries
+//! `connect` forever and there is no way to see what it would have sent. This
+//! binds that socket and reports what arrives.
+//!
+//! It decodes the Opus rather than only counting bytes, because byte counts
+//! cannot tell the two interesting failures apart. Opus codes digital silence
+//! in about two bytes a packet, so a sink receiving nothing but zeros still
+//! produces a steady ~3 kbps and looks, from every meter downstream, exactly
+//! like one that is working. Peak amplitude is what distinguishes them.
+//!
+//! Mirrors `ipc_listener::run_audio_listener` in the hub: same bind, same 0o666,
+//! same stream-type and codec checks. Where it differs from the hub, the hub is
+//! right and this should be corrected.
+
+use std::os::unix::net::UnixDatagram;
+use std::time::{Duration, Instant};
+
+use anyhow::{Result, bail};
+use clap::Parser;
+use opus_head_sys::*;
+
+use nesprotocol::{CODEC_OPUS, STREAM_AUDIO, decode_ipc_frame};
+
+#[derive(Parser, Debug)]
+#[command(about = "Receive and measure neswire's audio IPC stream")]
+struct Args {
+    /// Path to bind, matching neswire's --ipc-path.
+    #[arg(long, default_value = "/tmp/nestri-audio.sock")]
+    ipc_path: String,
+
+    /// Channels neswire is configured for. The decoder has to agree with the
+    /// encoder; a mismatch here reads as garbage, not as an error.
+    #[arg(long, default_value_t = 2)]
+    channels: u32,
+
+    /// How often to print a line.
+    #[arg(long, default_value_t = 1.0)]
+    interval_secs: f64,
+}
+
+struct MsDecoder {
+    ptr: *mut OpusMSDecoder,
+    channels: usize,
+}
+
+impl MsDecoder {
+    /// Built to match `MsEncoder::create_surround` for the same channel count:
+    /// stereo is mapping family 0, one stream, one coupled pair.
+    fn new(sample_rate: u32, channels: u32) -> Result<Self> {
+        let (streams, coupled, mapping): (i32, i32, Vec<u8>) = match channels {
+            2 => (1, 1, vec![0, 1]),
+            6 => (4, 2, vec![0, 4, 1, 2, 3, 5]),
+            8 => (5, 3, vec![0, 6, 1, 2, 3, 4, 5, 7]),
+            n => bail!("unsupported channel count: {n}"),
+        };
+
+        let mut error: i32 = 0;
+        let ptr = unsafe {
+            opus_multistream_decoder_create(
+                sample_rate as i32,
+                channels as i32,
+                streams,
+                coupled,
+                mapping.as_ptr(),
+                &mut error,
+            )
+        };
+        if error != OPUS_OK as i32 || ptr.is_null() {
+            bail!("opus_multistream_decoder_create failed: {error}");
+        }
+        Ok(Self {
+            ptr,
+            channels: channels as usize,
+        })
+    }
+
+    /// Returns the decoded samples, interleaved.
+    fn decode(&self, packet: &[u8], pcm: &mut [f32]) -> Result<usize> {
+        let frame_size = (pcm.len() / self.channels) as i32;
+        let decoded = unsafe {
+            opus_multistream_decode_float(
+                self.ptr,
+                packet.as_ptr(),
+                packet.len() as i32,
+                pcm.as_mut_ptr(),
+                frame_size,
+                0,
+            )
+        };
+        if decoded < 0 {
+            bail!("opus decode failed: {decoded}");
+        }
+        Ok(decoded as usize * self.channels)
+    }
+}
+
+impl Drop for MsDecoder {
+    fn drop(&mut self) {
+        unsafe { opus_multistream_decoder_destroy(self.ptr) };
+    }
+}
+
+fn main() -> Result<()> {
+    let args = Args::parse();
+
+    // Same sequence as the hub: clear a stale socket, bind, widen the mode.
+    // neswire runs as a different user there, and 0o666 is what makes that
+    // work; keeping it here means this stub cannot pass a case the hub fails.
+    let _ = std::fs::remove_file(&args.ipc_path);
+    let socket = UnixDatagram::bind(&args.ipc_path)?;
+    std::fs::set_permissions(
+        &args.ipc_path,
+        std::os::unix::fs::PermissionsExt::from_mode(0o666),
+    )?;
+    socket.set_read_timeout(Some(Duration::from_millis(200)))?;
+
+    println!("listening on {}", args.ipc_path);
+    println!("waiting for neswire...");
+
+    let decoder = MsDecoder::new(48_000, args.channels)?;
+    // 120ms at 48kHz is the largest frame Opus can produce, so nothing that
+    // decodes at all can overrun this.
+    let mut pcm = vec![0f32; 5760 * args.channels as usize];
+    let mut buf = vec![0u8; 65536];
+
+    let interval = Duration::from_secs_f64(args.interval_secs);
+    let mut window_start = Instant::now();
+    let mut packets = 0u64;
+    let mut bytes = 0u64;
+    let mut peak = 0f32;
+    let mut samples = 0u64;
+    let mut seen_anything = false;
+
+    loop {
+        match socket.recv(&mut buf) {
+            Ok(n) => {
+                let Some(frame) = decode_ipc_frame(&buf[..n]) else {
+                    eprintln!("invalid IPC frame ({n} bytes)");
+                    continue;
+                };
+                if frame.stream_type != STREAM_AUDIO {
+                    eprintln!("unexpected stream type: {}", frame.stream_type);
+                    continue;
+                }
+                if frame.codec != CODEC_OPUS {
+                    eprintln!("unexpected codec: {}", frame.codec);
+                    continue;
+                }
+
+                packets += 1;
+                bytes += frame.data.len() as u64;
+                seen_anything = true;
+
+                match decoder.decode(frame.data, &mut pcm) {
+                    Ok(count) => {
+                        samples += count as u64;
+                        for sample in &pcm[..count] {
+                            peak = peak.max(sample.abs());
+                        }
+                    }
+                    Err(e) => eprintln!("{e}"),
+                }
+            }
+            Err(ref e)
+                if e.kind() == std::io::ErrorKind::WouldBlock
+                    || e.kind() == std::io::ErrorKind::TimedOut => {}
+            Err(e) => return Err(e.into()),
+        }
+
+        let elapsed = window_start.elapsed();
+        if elapsed >= interval {
+            let secs = elapsed.as_secs_f64();
+            let kbps = bytes as f64 * 8.0 / 1000.0 / secs;
+
+            if packets == 0 {
+                println!(
+                    "{}",
+                    if seen_anything {
+                        "no packets — neswire stopped sending"
+                    } else {
+                        "no packets yet"
+                    }
+                );
+            } else if peak == 0.0 {
+                println!(
+                    "{packets:>4} pkt  {kbps:>6.1} kbps  {samples:>6} samples  \
+                     SILENT — decodes cleanly, every sample is zero"
+                );
+            } else {
+                println!(
+                    "{packets:>4} pkt  {kbps:>6.1} kbps  {samples:>6} samples  peak {peak:.4}"
+                );
+            }
+
+            window_start = Instant::now();
+            packets = 0;
+            bytes = 0;
+            samples = 0;
+            peak = 0.0;
+        }
+    }
+}

apps/neswire/src/encoder.rs

@@ -0,0 +1,421 @@
+use anyhow::{Result, bail};
+use crossbeam_channel::Receiver;
+use opus_head_sys::*;
+use std::os::unix::net::UnixDatagram;
+use std::time::Instant;
+
+use nesprotocol::{CODEC_OPUS, STREAM_AUDIO, encode_ipc_frame};
+
+pub struct EncoderConfig {
+    pub channels: u32,
+    pub sample_rate: u32,
+    pub frame_size: u32,
+    pub ipc_path: String,
+    pub bitrate_per_channel: u32,
+}
+
+struct MsEncoder {
+    ptr: *mut OpusMSEncoder,
+}
+impl MsEncoder {
+    fn create_surround(config: &EncoderConfig) -> Result<Self> {
+        let mut error: i32 = 0;
+        let mut streams: i32 = 0;
+        let mut coupled_streams: i32 = 0;
+        let mut mapping = [0u8; 255];
+
+        let mapping_family = if config.channels > 2 { 1 } else { 0 };
+
+        let encoder = unsafe {
+            opus_multistream_surround_encoder_create(
+                config.sample_rate as i32,
+                config.channels as i32,
+                mapping_family,
+                &mut streams,
+                &mut coupled_streams,
+                mapping.as_mut_ptr(),
+                OPUS_APPLICATION_AUDIO as i32,
+                &mut error,
+            )
+        };
+
+        if error != OPUS_OK as i32 || encoder.is_null() {
+            bail!(
+                "opus_multistream_surround_encoder_create failed: {}",
+                opus_error(error)
+            );
+        }
+
+        tracing::info!(
+            "opus multistream mapping: streams={}, coupled={}, mapping={:?}",
+            streams,
+            coupled_streams,
+            &mapping[..config.channels as usize],
+        );
+
+        let bitrate = (config.bitrate_per_channel * config.channels) * 1000;
+
+        let ret = unsafe {
+            opus_multistream_encoder_ctl(encoder, OPUS_SET_BITRATE_REQUEST as i32, bitrate)
+        };
+        if ret != OPUS_OK as i32 {
+            unsafe { opus_multistream_encoder_destroy(encoder) };
+            bail!("failed to set bitrate: {}", opus_error(ret));
+        }
+
+        tracing::info!("opus bitrate: {}kbps", bitrate / 1000);
+
+        let ret = unsafe {
+            opus_multistream_encoder_ctl(encoder, OPUS_SET_COMPLEXITY_REQUEST as i32, 10 as i32)
+        };
+        if ret != OPUS_OK as i32 {
+            unsafe { opus_multistream_encoder_destroy(encoder) };
+            bail!("failed to set complexity: {}", opus_error(ret));
+        }
+
+        tracing::info!("opus complexity set to 10 (max quality)");
+
+        let ret = unsafe {
+            opus_multistream_encoder_ctl(
+                encoder,
+                OPUS_SET_SIGNAL_REQUEST as i32,
+                OPUS_SIGNAL_MUSIC as i32,
+            )
+        };
+        if ret != OPUS_OK as i32 {
+            unsafe { opus_multistream_encoder_destroy(encoder) };
+            bail!("failed to set signal type: {}", opus_error(ret));
+        }
+
+        tracing::info!("opus signal type set to MUSIC");
+
+        Ok(Self { ptr: encoder })
+    }
+
+    fn encode_float(&self, pcm: &[f32], frame_size: u32, output: &mut [u8]) -> Result<usize> {
+        let ret = unsafe {
+            opus_multistream_encode_float(
+                self.ptr,
+                pcm.as_ptr(),
+                frame_size as i32,
+                output.as_mut_ptr(),
+                output.len() as i32,
+            )
+        };
+
+        if ret < 0 {
+            bail!("opus encode failed: {}", opus_error(ret));
+        }
+
+        Ok(ret as usize)
+    }
+}
+impl Drop for MsEncoder {
+    fn drop(&mut self) {
+        unsafe {
+            opus_multistream_encoder_destroy(self.ptr);
+        }
+    }
+}
+
+fn opus_error(code: i32) -> String {
+    unsafe {
+        let ptr = opus_strerror(code);
+        if ptr.is_null() {
+            format!("unknown error {}", code)
+        } else {
+            std::ffi::CStr::from_ptr(ptr).to_string_lossy().into_owned()
+        }
+    }
+}
+
+fn send_frame(
+    encoder: &MsEncoder,
+    socket: &UnixDatagram,
+    config: &EncoderConfig,
+    frame: &[f32],
+    start: &Instant,
+    opus_buf: &mut Vec<u8>,
+) -> Result<()> {
+    let encoded_len = encoder.encode_float(frame, config.frame_size, opus_buf)?;
+    let encoded_data = &opus_buf[..encoded_len];
+    let timestamp_ms = start.elapsed().as_millis() as u32;
+    let ipc_frame = encode_ipc_frame(
+        STREAM_AUDIO,
+        CODEC_OPUS,
+        0,
+        timestamp_ms,
+        0,
+        0,
+        encoded_data,
+    );
+    socket.send(&ipc_frame)?;
+    Ok(())
+}
+
+pub fn run(config: EncoderConfig, rx: Receiver<Vec<f32>>) -> Result<()> {
+    tracing::info!(
+        "encoder started — IPC {}, {}ch, {} samples/frame",
+        config.ipc_path,
+        config.channels,
+        config.frame_size,
+    );
+
+    let socket = UnixDatagram::unbound()?;
+    let encoder = MsEncoder::create_surround(&config)?;
+
+    let mut opus_buf = vec![0u8; 4000];
+    let start = Instant::now();
+
+    // Connect to hub with retry
+    loop {
+        match socket.connect(&config.ipc_path) {
+            Ok(()) => {
+                tracing::info!("IPC connected to {}", config.ipc_path);
+                break;
+            }
+            Err(e) => {
+                tracing::warn!(
+                    "IPC connect to {} failed (retrying in 2s): {e}",
+                    config.ipc_path
+                );
+                std::thread::sleep(std::time::Duration::from_secs(2));
+            }
+        }
+    }
+
+    // Send loop
+    while let Ok(frame) = rx.recv() {
+        send_frame(&encoder, &socket, &config, &frame, &start, &mut opus_buf)?;
+    }
+
+    tracing::info!("encoder shut down");
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nesprotocol::decode_ipc_frame;
+    use std::os::unix::net::UnixDatagram;
+
+    const SAMPLE_RATE: u32 = 48_000;
+    const CHANNELS: u32 = 2;
+    /// 5ms, the default. 240 samples per channel.
+    const FRAME_SIZE: u32 = 240;
+
+    fn config(ipc_path: &str) -> EncoderConfig {
+        EncoderConfig {
+            channels: CHANNELS,
+            sample_rate: SAMPLE_RATE,
+            frame_size: FRAME_SIZE,
+            ipc_path: ipc_path.to_string(),
+            bitrate_per_channel: 64,
+        }
+    }
+
+    fn silence() -> Vec<f32> {
+        vec![0.0; (FRAME_SIZE * CHANNELS) as usize]
+    }
+
+    /// One frame of a 440Hz tone, interleaved stereo, starting at `frame_index`
+    /// so consecutive frames form a continuous wave rather than restarting.
+    fn tone(frame_index: usize) -> Vec<f32> {
+        let start = frame_index * FRAME_SIZE as usize;
+        (0..FRAME_SIZE as usize)
+            .flat_map(|i| {
+                let t = (start + i) as f32 / SAMPLE_RATE as f32;
+                let v = (t * 440.0 * std::f32::consts::TAU).sin() * 0.5;
+                [v, v]
+            })
+            .collect()
+    }
+
+    /// The decoder the hub's client ends up using, in miniature: same
+    /// parameters the desktop app builds for stereo.
+    struct Decoder(*mut OpusMSDecoder);
+
+    impl Decoder {
+        fn new() -> Self {
+            let mut error = 0i32;
+            let mapping = [0u8, 1];
+            let ptr = unsafe {
+                opus_multistream_decoder_create(
+                    SAMPLE_RATE as i32,
+                    CHANNELS as i32,
+                    1,
+                    1,
+                    mapping.as_ptr(),
+                    &mut error,
+                )
+            };
+            assert_eq!(error, OPUS_OK as i32, "decoder create failed");
+            Self(ptr)
+        }
+
+        fn decode(&self, packet: &[u8]) -> Vec<f32> {
+            let mut pcm = vec![0f32; (FRAME_SIZE * CHANNELS) as usize];
+            let n = unsafe {
+                opus_multistream_decode_float(
+                    self.0,
+                    packet.as_ptr(),
+                    packet.len() as i32,
+                    pcm.as_mut_ptr(),
+                    FRAME_SIZE as i32,
+                    0,
+                )
+            };
+            assert!(n > 0, "decode failed: {n}");
+            pcm.truncate(n as usize * CHANNELS as usize);
+            pcm
+        }
+    }
+
+    impl Drop for Decoder {
+        fn drop(&mut self) {
+            unsafe { opus_multistream_decoder_destroy(self.0) };
+        }
+    }
+
+    fn peak(pcm: &[f32]) -> f32 {
+        pcm.iter().fold(0f32, |acc, s| acc.max(s.abs()))
+    }
+
+    /// Silence is nearly free to encode, and that is a trap worth pinning down.
+    ///
+    /// The configured bitrate is a ceiling, not a floor. At 200 packets a second
+    /// these two-byte packets come to roughly 3kbps of perfectly valid,
+    /// perfectly silent Opus — which looks, to every byte counter downstream,
+    /// like a working stream. A whole debugging session was spent on the
+    /// difference. If this test starts failing because silence got expensive,
+    /// that reasoning needs revisiting.
+    #[test]
+    fn silence_costs_almost_nothing_to_encode() {
+        let config = config("/nonexistent");
+        let encoder = MsEncoder::create_surround(&config).expect("encoder");
+        let mut buf = vec![0u8; 4000];
+
+        // Past the encoder's warm-up, where the first packets are larger.
+        for _ in 0..10 {
+            encoder
+                .encode_float(&silence(), FRAME_SIZE, &mut buf)
+                .expect("encode");
+        }
+
+        let len = encoder
+            .encode_float(&silence(), FRAME_SIZE, &mut buf)
+            .expect("encode");
+        assert!(
+            len <= 8,
+            "silence took {len} bytes; the ~3kbps silent-stream signature no \
+             longer holds and the diagnostics that rely on it are wrong"
+        );
+    }
+
+    /// The other half of the above: real audio must cost real bytes, or the
+    /// test before this one would pass on a permanently broken encoder.
+    #[test]
+    fn a_tone_costs_far_more_than_silence() {
+        let config = config("/nonexistent");
+        // One encoder each. Opus codes a stream, not isolated frames, so
+        // feeding both signals to one encoder makes every "silent" frame the
+        // tail of a tone and costs it accordingly -- which is a measurement of
+        // nothing.
+        let quiet_encoder = MsEncoder::create_surround(&config).expect("encoder");
+        let loud_encoder = MsEncoder::create_surround(&config).expect("encoder");
+        let mut buf = vec![0u8; 4000];
+
+        let mut quiet = 0;
+        let mut loud = 0;
+        for i in 0..40 {
+            quiet += quiet_encoder
+                .encode_float(&silence(), FRAME_SIZE, &mut buf)
+                .expect("encode");
+            loud += loud_encoder
+                .encode_float(&tone(i), FRAME_SIZE, &mut buf)
+                .expect("encode");
+        }
+
+        assert!(
+            loud > quiet * 4,
+            "a tone encoded to {loud} bytes against {quiet} for silence; the \
+             encoder is not responding to its input"
+        );
+    }
+
+    /// A tone goes in and comes back out, through the exact encoder the guest
+    /// runs and the exact decoder parameters the desktop client builds.
+    ///
+    /// This is the one that would have caught a channel-mapping or sample-rate
+    /// disagreement between the two ends, which no byte count can see.
+    #[test]
+    fn a_tone_survives_the_round_trip() {
+        let config = config("/nonexistent");
+        let encoder = MsEncoder::create_surround(&config).expect("encoder");
+        let decoder = Decoder::new();
+        let mut buf = vec![0u8; 4000];
+
+        // Opus needs a few frames before its output is representative, so the
+        // assertion is on the tail rather than the first packet.
+        let mut last = Vec::new();
+        for i in 0..40 {
+            let len = encoder
+                .encode_float(&tone(i), FRAME_SIZE, &mut buf)
+                .expect("encode");
+            last = decoder.decode(&buf[..len]);
+        }
+
+        assert_eq!(
+            last.len(),
+            (FRAME_SIZE * CHANNELS) as usize,
+            "decoded frame is the wrong length"
+        );
+        assert!(
+            peak(&last) > 0.1,
+            "a 0.5-amplitude tone decoded to a peak of {:.4}",
+            peak(&last)
+        );
+    }
+
+    /// What actually crosses the socket is what the hub knows how to read.
+    ///
+    /// The hub rejects any frame whose stream type is not `STREAM_AUDIO` or
+    /// whose codec is not `CODEC_OPUS`, and drops it with a warning nobody
+    /// reads. This asserts the contract from the sending side.
+    #[test]
+    fn the_hub_receives_a_frame_it_can_parse() {
+        let path = std::env::temp_dir().join(format!(
+            "neswire-test-{}-{}.sock",
+            std::process::id(),
+            line!()
+        ));
+        let _ = std::fs::remove_file(&path);
+
+        // Bound first: the hub binds and neswire connects, so a test that did
+        // this the other way round would not be testing the real sequence.
+        let listener = UnixDatagram::bind(&path).expect("bind");
+        let sender = UnixDatagram::unbound().expect("socket");
+        sender.connect(&path).expect("connect");
+
+        let config = config(path.to_str().expect("utf-8 path"));
+        let encoder = MsEncoder::create_surround(&config).expect("encoder");
+        let start = Instant::now();
+        let mut opus_buf = vec![0u8; 4000];
+
+        send_frame(&encoder, &sender, &config, &tone(0), &start, &mut opus_buf)
+            .expect("send");
+
+        let mut buf = vec![0u8; 65536];
+        let n = listener.recv(&mut buf).expect("recv");
+        let frame = decode_ipc_frame(&buf[..n]).expect("hub could not parse the frame");
+
+        assert_eq!(frame.stream_type, STREAM_AUDIO);
+        assert_eq!(frame.codec, CODEC_OPUS);
+        assert!(!frame.data.is_empty(), "frame carried no payload");
+
+        // And the payload is Opus the far end can actually decode.
+        Decoder::new().decode(frame.data);
+
+        let _ = std::fs::remove_file(&path);
+    }
+}

apps/neswire/src/main.rs

@@ -0,0 +1,58 @@
+mod encoder;
+mod sink;
+
+use clap::Parser;
+
+#[derive(Parser, Debug)]
+struct Args {
+    /// Path for the audio IPC socket (neswire → nestri-guest-hub)
+    #[arg(
+        long,
+        env = "NESWIRE_IPC_PATH",
+        default_value = "/tmp/nestri-audio.sock"
+    )]
+    ipc_path: String,
+
+    /// Output channels: 2, 6, or 8
+    #[arg(long, env = "NESWIRE_CHANNELS", default_value_t = 2)]
+    channels: u32,
+
+    /// Packet duration in ms (5, 10..)
+    #[arg(long, env = "NESWIRE_PACKET_DURATION_MS", default_value_t = 5)]
+    packet_duration_ms: u32,
+
+    /// Bitrate per channel in kbps
+    #[arg(long, env = "NESWIRE_BITRATE_PER_CHANNEL", default_value_t = 64)]
+    bitrate_per_channel: u32,
+}
+
+fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .init();
+
+    let args = Args::parse();
+
+    let sample_rate = 48_000u32;
+    let frame_size = (sample_rate * args.packet_duration_ms) / 1000;
+
+    let (tx, rx) = crossbeam_channel::bounded::<Vec<f32>>(256);
+
+    let encoder_config = encoder::EncoderConfig {
+        channels: args.channels,
+        sample_rate,
+        frame_size,
+        ipc_path: args.ipc_path,
+        bitrate_per_channel: args.bitrate_per_channel,
+    };
+
+    std::thread::spawn(move || {
+        if let Err(e) = encoder::run(encoder_config, rx) {
+            tracing::error!("encoder thread died: {e:#}");
+        }
+    });
+
+    sink::run(sample_rate, args.channels, frame_size, tx)?;
+
+    Ok(())
+}

apps/neswire/src/sink.rs

@@ -0,0 +1,183 @@
+use anyhow::Result;
+use crossbeam_channel::Sender;
+use pipewire::{self as pw, loop_::Signal, spa, stream::*};
+use spa::param::audio::{AudioFormat, AudioInfoRaw};
+
+pub fn run(sample_rate: u32, channels: u32, frame_size: u32, tx: Sender<Vec<f32>>) -> Result<()> {
+    pw::init();
+    tracing::info!("pipewire init ok");
+
+    let mainloop =
+        pw::main_loop::MainLoopRc::new(None).map_err(|e| anyhow::anyhow!("mainloop: {e}"))?;
+    tracing::info!("mainloop created");
+
+    let context = pw::context::ContextRc::new(&mainloop, None)
+        .map_err(|e| anyhow::anyhow!("context: {e}"))?;
+    tracing::info!("context created");
+
+    let core = context
+        .connect_rc(None)
+        .map_err(|e| anyhow::anyhow!("core connect: {e}"))?;
+    tracing::info!("core connected");
+
+    let stream = pw::stream::StreamRc::new(
+        core,
+        "neswire",
+        pw::properties::properties! {
+            *pw::keys::MEDIA_TYPE => "Audio",
+            *pw::keys::MEDIA_CATEGORY => "Capture",
+            *pw::keys::MEDIA_CLASS => "Audio/Sink",
+            *pw::keys::NODE_NAME => "neswire",
+            *pw::keys::NODE_DESCRIPTION => "Neswire Cloud Gaming Audio Sink",
+            // Prevent session manager from suspending us
+            "node.always-process" => "true",
+            // Desired latency in samples
+            "node.latency" => format!("{}/{}", frame_size, sample_rate),
+        },
+    )
+    .map_err(|e| anyhow::anyhow!("stream create: {e}"))?;
+
+    // Accumulation buffer for collecting enough samples before sending a frame
+    let samples_per_frame = (frame_size * channels) as usize;
+
+    struct State {
+        tx: Sender<Vec<f32>>,
+        accum: Vec<f32>,
+        samples_per_frame: usize,
+    }
+
+    let state = State {
+        tx,
+        accum: Vec::with_capacity(samples_per_frame * 2),
+        samples_per_frame,
+    };
+
+    let _listener = stream
+        .add_local_listener_with_user_data(state)
+        .param_changed(move |_, _state, id, pod| {
+            if id != spa::param::ParamType::Format.as_raw() {
+                return;
+            }
+            // You can parse the negotiated format here if needed
+            // For now we're requesting a fixed format so it should match
+            if let Some(_pod) = pod {
+                tracing::info!("format negotiated");
+            }
+        })
+        .process(move |stream, state| {
+            // Dequeue the buffer from PipeWire
+            if let Some(mut buffer) = stream.dequeue_buffer() {
+                let datas = buffer.datas_mut();
+                if let Some(data) = datas.first_mut() {
+                    let chunk = data.chunk();
+                    let offset = chunk.offset() as usize;
+                    let size = chunk.size() as usize;
+
+                    if let Some(slice) = data.data() {
+                        let audio_bytes = &slice[offset..offset + size];
+
+                        // Reinterpret as f32 samples (we requested F32LE)
+                        let samples: &[f32] = bytemuck::cast_slice(audio_bytes);
+
+                        state.accum.extend_from_slice(samples);
+
+                        tracing::trace!(
+                            "pw delivered {} samples, accum now {}, frame size {}",
+                            samples.len(),
+                            state.accum.len(),
+                            state.samples_per_frame,
+                        );
+
+                        // Drain complete frames
+                        while state.accum.len() >= state.samples_per_frame {
+                            let frame: Vec<f32> =
+                                state.accum.drain(..state.samples_per_frame).collect();
+
+                            if state.tx.try_send(frame).is_err() {
+                                tracing::warn!("encoder falling behind, dropping frame");
+                            }
+                        }
+                    }
+                }
+            }
+        })
+        .register()
+        .map_err(|e| anyhow::anyhow!("stream register: {e}"))?;
+
+    let mut position = [0u32; 64];
+    let channel_map: &[u32] = match channels {
+        2 => &[
+            spa::sys::SPA_AUDIO_CHANNEL_FL,
+            spa::sys::SPA_AUDIO_CHANNEL_FR,
+        ],
+        6 => &[
+            spa::sys::SPA_AUDIO_CHANNEL_FL,
+            spa::sys::SPA_AUDIO_CHANNEL_FC,
+            spa::sys::SPA_AUDIO_CHANNEL_FR,
+            spa::sys::SPA_AUDIO_CHANNEL_RL,
+            spa::sys::SPA_AUDIO_CHANNEL_RR,
+            spa::sys::SPA_AUDIO_CHANNEL_LFE,
+        ],
+        8 => &[
+            spa::sys::SPA_AUDIO_CHANNEL_FL,
+            spa::sys::SPA_AUDIO_CHANNEL_FC,
+            spa::sys::SPA_AUDIO_CHANNEL_FR,
+            spa::sys::SPA_AUDIO_CHANNEL_SL,
+            spa::sys::SPA_AUDIO_CHANNEL_SR,
+            spa::sys::SPA_AUDIO_CHANNEL_RL,
+            spa::sys::SPA_AUDIO_CHANNEL_RR,
+            spa::sys::SPA_AUDIO_CHANNEL_LFE,
+        ],
+        _ => anyhow::bail!("unsupported channel count: {channels}"),
+    };
+    position[..channel_map.len()].copy_from_slice(channel_map);
+
+    // Build the format we want: f32le, 48kHz, N channels
+    let mut audio_info = AudioInfoRaw::new();
+    audio_info.set_format(AudioFormat::F32LE);
+    audio_info.set_rate(sample_rate);
+    audio_info.set_channels(channels);
+    audio_info.set_position(position);
+
+    let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
+        std::io::Cursor::new(Vec::new()),
+        &pw::spa::pod::Value::Object(pw::spa::pod::Object {
+            type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
+            id: pw::spa::param::ParamType::EnumFormat.as_raw(),
+            properties: audio_info.into(),
+        }),
+    )?
+    .0
+    .into_inner();
+
+    let mut params = [pw::spa::pod::Pod::from_bytes(&values).unwrap()];
+
+    stream
+        .connect(
+            spa::utils::Direction::Input, // We receive audio (we're a sink)
+            Some(pw::constants::ID_ANY),
+            StreamFlags::AUTOCONNECT | StreamFlags::MAP_BUFFERS | StreamFlags::RT_PROCESS,
+            &mut params,
+        )
+        .map_err(|e| anyhow::anyhow!("stream connect: {e}"))?;
+    tracing::info!("stream connected");
+
+    tracing::info!(
+        "neswire sink running — {}ch, {}Hz, {} samples/frame",
+        channels,
+        sample_rate,
+        frame_size
+    );
+
+    let weak = mainloop.downgrade();
+    let _sigint = mainloop.loop_().add_signal_local(Signal::INT, move || {
+        if let Some(mainloop) = weak.upgrade() {
+            mainloop.quit();
+        }
+    });
+
+    mainloop.run();
+
+    tracing::info!("shutting down");
+    Ok(())
+}