peerspeak/src/core/mod.rs

pub mod messages;
pub mod jitter;

use crate::audio::{AudioBackend, pipewire_impl::PipeWireBackend};
use crate::audio::eq::{Eq, EqSettings};
use crate::codec::{AudioEncoder, opus_impl::OpusEncoder};
use crate::core::jitter::{JitterBuffer, FRAME_SAMPLES};
use crate::network::{
    NetworkTransport, RoomState, PeerState, RoomEvent, ConnEvent, PeerSpeakTicket,
    iroh_impl::{IrohTransport, AudioRouter},
    gossip::IrohGossipState,
};
use crate::core::messages::{CoreCommand, UiEvent};

use crate::config::{NetworkMode, RecordingMode};
use crate::presence::PresenceMode;
use crate::audio::multitrack::MultitrackRecorder;
use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey, endpoint::presets, protocol::Router};
use iroh_gossip::net::Gossip;
use tokio::sync::{mpsc, Mutex};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::Duration;

pub struct CoreController {
    cmd_tx: mpsc::Sender<CoreCommand>,
}

impl CoreController {
    pub fn new(ui_tx: mpsc::Sender<UiEvent>) -> Self {
        let (cmd_tx, cmd_rx) = mpsc::channel(100);

        std::thread::spawn(move || {
            let rt = tokio::runtime::Runtime::new().expect("Failed to create Tokio runtime");
            rt.block_on(async move {
                crate::log_msg("Starting core network loop in dedicated Tokio runtime");
                if let Err(e) = run_core_loop(cmd_rx, ui_tx).await {
                    crate::log_msg(&format!("App core loop failed: {:?}", e));
                }
            });
        });

        Self { cmd_tx }
    }

    /// Queue a command for the core loop, best-effort. Returns `true` if it was
    /// accepted, `false` if the channel is full or closed. (We return a plain
    /// bool rather than the channel's `Result` so the bulky `CoreCommand` isn't
    /// carried back by value in every caller's error type.)
    pub fn send(&self, cmd: CoreCommand) -> bool {
        self.cmd_tx.try_send(cmd).is_ok()
    }

    /// Clone the command sender for asynchronous one-shot sends that should wait
    /// for channel capacity instead of failing immediately on a full queue.
    pub fn command_sender(&self) -> mpsc::Sender<CoreCommand> {
        self.cmd_tx.clone()
    }
}

/// How long a peer may stay "reconnecting" after a transient drop before we give
/// up and evict it. Comfortably past the QUIC idle timeout (~30s) so a genuine
/// reconnect has time to complete, but short enough that a crashed/departed peer
/// clears from the room promptly.
const RECONNECT_GRACE: Duration = Duration::from_secs(45);

/// Opus frames sent by our encoder are one 20 ms mono frame, normally far below
/// this. 4000 bytes still leaves room for large valid Opus packets (well above a
/// 48 kHz / 60 ms frame) while bounding malicious datagram copy/decode churn.
const MAX_OPUS_PAYLOAD: usize = 4000;

/// If the Discoverable time-box tries to revert but discovery service reconfiguration
/// fails, retry soon while keeping the UI in the still-possible publishing state.
const DISCOVERY_REVERT_RETRY: Duration = Duration::from_secs(60);

fn audio_datagram_len_ok(len: usize) -> bool {
    (4..=4 + MAX_OPUS_PAYLOAD).contains(&len)
}

fn arm_discovery_retry(
    discovery_deadline: &mut Option<tokio::time::Instant>,
    now: tokio::time::Instant,
) {
    let retry_deadline = now + DISCOVERY_REVERT_RETRY;
    if discovery_deadline
        .map(|current| current > retry_deadline)
        .unwrap_or(true)
    {
        *discovery_deadline = Some(retry_deadline);
    }
}

/// Per-peer reconnect grace timers (see [`RECONNECT_GRACE`]). Shared between the
/// room-event task (which arms one on a transient drop and cancels it on a
/// gossip rejoin) and the conn-event task (which cancels it when the audio link
/// actually comes back).
type GraceTimers = Arc<std::sync::Mutex<HashMap<EndpointId, tokio::task::JoinHandle<()>>>>;

/// Peers we've completed at least one audio link with. Lets the conn-event task
/// tell a genuine reconnect (arm an eviction timer) from a first-ever dial (don't).
/// Scrubbed whenever a peer is evicted or leaves so a later rejoin starts clean.
type SeenConnected = Arc<std::sync::Mutex<HashSet<EndpointId>>>;

/// Cancel and forget a peer's pending grace timer, if any. No-op if none is armed.
fn cancel_grace_timer(timers: &GraceTimers, peer_id: &EndpointId) {
    if let Some(handle) = timers.lock().unwrap().remove(peer_id) {
        handle.abort();
    }
}

/// Arm a per-peer reconnect grace timer that evicts the peer if its link hasn't
/// recovered within [`RECONNECT_GRACE`]. No-op if a timer is already pending for
/// the peer, so the earliest drop notice — whether the gossip `PeerConnectionLost`
/// or the transport `Connecting` — sets one hard deadline, rather than a flapping
/// link repeatedly resetting the clock and dodging eviction forever. On firing it
/// also scrubs the peer from `seen_connected` so a later rejoin isn't treated as a
/// reconnect on its initial dial.
fn arm_grace_timer(
    timers: &GraceTimers,
    seen_connected: &SeenConnected,
    transport: &Arc<IrohTransport>,
    jitter: &Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
    ui_tx: &mpsc::Sender<UiEvent>,
    grace: Duration,
    peer_id: EndpointId,
) {
    let mut timers_guard = timers.lock().unwrap();
    if timers_guard.contains_key(&peer_id) {
        return;
    }
    let transport_evict = transport.clone();
    let jitter_evict = jitter.clone();
    let ui_evict = ui_tx.clone();
    let timers_evict = timers.clone();
    let seen_evict = seen_connected.clone();
    let handle = tokio::spawn(async move {
        tokio::time::sleep(grace).await;
        crate::log_msg(&format!("Reconnect grace expired; evicting peer {:?}", peer_id));
        transport_evict.remove_audio_sender(peer_id);
        transport_evict.disconnect_peer(peer_id).await;
        jitter_evict.lock().await.remove(&peer_id);
        // Scrub our internal state *before* announcing the eviction, so anything
        // that observes `PeerConnectionFailed` (or a rejoin racing it) sees a clean
        // slate — a later dial for this identity is then a fresh first-dial, not a
        // reconnect.
        timers_evict.lock().unwrap().remove(&peer_id);
        seen_evict.lock().unwrap().remove(&peer_id);
        let _ = ui_evict.send(UiEvent::PeerConnectionFailed { id: peer_id }).await;
    });
    timers_guard.insert(peer_id, handle);
}

/// Scale a frame in place by a per-peer volume factor, saturating to the i16
/// range. A volume within `f32::EPSILON` of 1.0 is treated as unity and skipped,
/// matching the mixer hot path that avoids touching unmodified frames.
fn apply_volume(frame: &mut [i16], vol: f32) {
    if (vol - 1.0).abs() <= f32::EPSILON {
        return;
    }
    for sample in frame.iter_mut() {
        *sample = (*sample as f32 * vol).clamp(i16::MIN as f32, i16::MAX as f32) as i16;
    }
}

/// Normalized RMS level of a frame in `[0.0, 1.0]` (32768 = full scale), for the
/// UI level meter. An empty frame reads as 0.0.
fn frame_level(frame: &[i16]) -> f32 {
    let sum_sq: f32 = frame.iter().map(|&x| (x as f32).powi(2)).sum();
    let rms = (sum_sq / frame.len().max(1) as f32).sqrt();
    (rms / 32768.0).clamp(0.0, 1.0)
}

/// Peak-hold every this many captured samples (~100ms @ 48kHz) before emitting a
/// [`UiEvent::MicLevel`], so the meter doesn't flood the UI runtime at frame rate.
const MIC_LEVEL_REPORT_SAMPLES: usize = 4800;

/// Peak-holds the raw mic level across captured frames and yields a value to
/// report roughly every [`MIC_LEVEL_REPORT_SAMPLES`] samples. Shared by the
/// in-call capture thread and the standalone monitor so both throttle and
/// peak-hold identically.
struct MicLevelMeter {
    peak: f32,
    acc: usize,
}

impl MicLevelMeter {
    fn new() -> Self {
        Self { peak: 0.0, acc: 0 }
    }

    /// Folds one frame into the running peak. Returns `Some(peak)` (and resets)
    /// once enough samples have accumulated to emit a reading, else `None`.
    fn push(&mut self, frame: &[i16]) -> Option<f32> {
        self.peak = self.peak.max(frame_level(frame));
        self.acc += frame.len();
        if self.acc >= MIC_LEVEL_REPORT_SAMPLES {
            let peak = self.peak;
            self.peak = 0.0;
            self.acc = 0;
            Some(peak)
        } else {
            None
        }
    }
}

/// A standalone, capture-only mic monitor for gate calibration outside a call.
/// Owns the worker thread that reads raw PCM and reports its level; the PipeWire
/// capture stream itself lives in the shared backend. Tear down by stopping the
/// backend's capture (which closes the channel) and joining this thread.
struct MicMonitor {
    thread: std::thread::JoinHandle<()>,
}

/// Drains a capture channel, reporting the raw (un-gated) mic level to the UI.
/// Returns when the channel closes (i.e. the backend's capture stream stopped).
fn run_mic_monitor(
    rx: std::sync::mpsc::Receiver<Vec<i16>>,
    ui_tx: mpsc::Sender<UiEvent>,
    input_gain: Arc<std::sync::atomic::AtomicU32>,
) {
    let mut meter = MicLevelMeter::new();
    while let Ok(mut pcm) = rx.recv() {
        // Mirror the in-call path: apply the input gain before metering so the
        // test meter reflects the gained signal (and the input slider moves it).
        apply_volume(&mut pcm, f32::from_bits(input_gain.load(Ordering::Relaxed)));
        if let Some(peak) = meter.push(&pcm) {
            // Drop on a full channel — a stale meter reading is harmless.
            let _ = ui_tx.try_send(UiEvent::MicLevel(peak));
        }
    }
    // Channel closed: the monitor was stopped. Snap the meter back to zero.
    let _ = ui_tx.try_send(UiEvent::MicLevel(0.0));
}

/// Stops a standalone mic monitor if one is running. MUST NOT be called while a
/// room session is active — `backend.stop()` would also tear down the call's
/// capture/playback. Monitor and session are mutually exclusive by construction.
fn stop_mic_monitor(backend: &PipeWireBackend, monitor: Option<MicMonitor>) {
    if let Some(m) = monitor {
        let _ = backend.stop();
        let _ = m.thread.join();
    }
}

/// Sum per-peer frames sample-by-sample into one `frame_len`-sample bus, **without**
/// clamping — the lossless `i32` sum preserves the true peak so the mix-bus soft
/// limiter (see [`crate::audio::limiter`]) can ride it down to the ceiling instead
/// of the old hard clip shattering loud moments. Peers shorter than `frame_len`
/// contribute 0 past their end; an empty peer set yields a silent bus.
#[cfg(test)]
fn mix_frames(peer_frames: &[Vec<i16>], frame_len: usize) -> Vec<i32> {
    let mut mixed = vec![0i32; frame_len];
    for frame in peer_frames {
        for (out, &sample) in mixed.iter_mut().zip(frame.iter()) {
            *out += sample as i32;
        }
    }
    mixed
}

/// Sum per-peer mono frames into one interleaved stereo `i32` bus. Center pan is
/// a special exact dual-mono path so the default listener mix is bit-for-bit the
/// old mono sum duplicated to both ears.
fn mix_stereo_frames(peer_frames: &[(Vec<i16>, f32)], frame_len: usize) -> Vec<i32> {
    let mut mixed = vec![0i32; frame_len * crate::audio::PLAYBACK_CHANNELS];
    for (frame, pan) in peer_frames {
        if pan.abs() <= f32::EPSILON {
            for (i, &sample) in frame.iter().take(frame_len).enumerate() {
                let idx = i * crate::audio::PLAYBACK_CHANNELS;
                let s = sample as i32;
                mixed[idx] += s;
                mixed[idx + 1] += s;
            }
            continue;
        }

        let (left_gain, right_gain) = crate::audio::pan::playback_pan_gains(*pan);
        for (i, &sample) in frame.iter().take(frame_len).enumerate() {
            let idx = i * crate::audio::PLAYBACK_CHANNELS;
            let x = sample as f32;
            mixed[idx] += (x * left_gain).round() as i32;
            mixed[idx + 1] += (x * right_gain).round() as i32;
        }
    }
    mixed
}

/// Fold an interleaved stereo frame to mono for the existing mixed WAV writers.
/// Center/default pan folds back to the exact old mono mix.
fn stereo_to_mono(stereo: &[i16]) -> Vec<i16> {
    let mut mono = Vec::with_capacity(stereo.len() / crate::audio::PLAYBACK_CHANNELS);
    for pair in stereo.chunks_exact(crate::audio::PLAYBACK_CHANNELS) {
        let sum = pair[0] as i32 + pair[1] as i32;
        mono.push((sum / 2).clamp(i16::MIN as i32, i16::MAX as i32) as i16);
    }
    mono
}

/// Handles the transport's per-peer link-state stream (`ConnEvent`): arms/cancels
/// reconnect grace timers, tracks which peers we've linked with, and forwards
/// link state to the UI. Pulled out of the conn-event task as a unit so the
/// reconnect-eviction behavior can be tested without standing up a full session.
/// Exposed (with a `grace` override) for that purpose; not part of the public API.
pub struct ConnEventHandler {
    ui_tx: mpsc::Sender<UiEvent>,
    grace_timers: GraceTimers,
    seen_connected: SeenConnected,
    transport: Arc<IrohTransport>,
    jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
    grace: Duration,
}

impl ConnEventHandler {
    pub fn new(
        ui_tx: mpsc::Sender<UiEvent>,
        grace_timers: GraceTimers,
        seen_connected: SeenConnected,
        transport: Arc<IrohTransport>,
        jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
    ) -> Self {
        Self {
            ui_tx,
            grace_timers,
            seen_connected,
            transport,
            jitter,
            grace: RECONNECT_GRACE,
        }
    }

    /// Override the eviction grace window. For tests that can't wait 45s.
    pub fn with_grace(mut self, grace: Duration) -> Self {
        self.grace = grace;
        self
    }

    pub async fn handle(&self, event: ConnEvent) {
        match event {
            ConnEvent::Connecting(id) => {
                // A reconnect (we've linked with this peer before): arm an eviction
                // timer so a peer that never comes back is cleared even when gossip
                // doesn't re-report the drop — the transport reliably re-emits this
                // on every outage, gossip's NeighborDown does not. A first-ever dial
                // (not yet in seen_connected) gets no timer; Connected cancels it on
                // recovery.
                if self.seen_connected.lock().unwrap().contains(&id) {
                    arm_grace_timer(
                        &self.grace_timers,
                        &self.seen_connected,
                        &self.transport,
                        &self.jitter,
                        &self.ui_tx,
                        self.grace,
                        id,
                    );
                }
                let _ = self.ui_tx.send(UiEvent::PeerConnecting { id }).await;
            }
            ConnEvent::Connected(id) => {
                // The audio link came back — the peer recovered within the grace
                // window, so cancel its eviction.
                cancel_grace_timer(&self.grace_timers, &id);
                self.seen_connected.lock().unwrap().insert(id);
                let _ = self.ui_tx.send(UiEvent::PeerConnected { id }).await;
            }
            ConnEvent::Left(id) => {
                // The peer closed its link gracefully (intentional leave) — evict
                // immediately, like a PeerLeft, instead of leaving it "reconnecting"
                // until the grace timer or the slow gossip Leave.
                cancel_grace_timer(&self.grace_timers, &id);
                self.seen_connected.lock().unwrap().remove(&id);
                self.transport.remove_audio_sender(id);
                self.transport.disconnect_peer(id).await;
                self.jitter.lock().await.remove(&id);
                let _ = self.ui_tx.send(UiEvent::PeerLeft { id }).await;
            }
        }
    }
}

struct ActiveSession {
    room_state: Arc<IrohGossipState>,
    capture_thread: std::thread::JoinHandle<()>,
    datagram_task: tokio::task::JoinHandle<()>,
    mixer_task: tokio::task::JoinHandle<()>,
    event_task: tokio::task::JoinHandle<()>,
    conn_event_task: tokio::task::JoinHandle<()>,
    grace_timers: GraceTimers,
    transport: Arc<IrohTransport>,
    /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop.
    echo_cancel: Option<crate::audio::echo_cancel::EchoCancelGuard>,
    /// Our pixelpass screen-share host child while sharing (`kill_on_drop`, so it
    /// also dies if the session is dropped without an explicit stop).
    screenshare_host: Option<tokio::process::Child>,
    /// pixelpass viewer children we spawned to watch peers' shares; killed on
    /// session teardown (each also self-exits when its player window closes).
    screenshare_viewers: Vec<tokio::process::Child>,
}

impl ActiveSession {
    async fn shutdown(mut self, audio_backend: Arc<PipeWireBackend>) {
        crate::log_msg("ActiveSession::shutdown started");
        // Tear down any screen-share children first so the host stops streaming
        // promptly (kill_on_drop is the backstop, but kill explicitly so viewers
        // see the stream end without waiting on drop ordering).
        if let Some(mut host) = self.screenshare_host.take() {
            let _ = host.kill().await;
        }
        for mut viewer in self.screenshare_viewers.drain(..) {
            let _ = viewer.kill().await;
        }
        self.datagram_task.abort();
        self.mixer_task.abort();
        self.event_task.abort();
        self.conn_event_task.abort();
        // Abort any pending reconnect grace timers so they can't fire a stray
        // eviction (or touch a torn-down transport) after the session is gone.
        for (_, handle) in self.grace_timers.lock().unwrap().drain() {
            handle.abort();
        }
        crate::log_msg("Aborted tasks");

        let audio_backend_clone = audio_backend.clone();
        let _ = tokio::task::spawn_blocking(move || {
            crate::log_msg("Stopping audio backend...");
            let _ = audio_backend_clone.stop();
            crate::log_msg("Audio backend stopped");
        }).await;

        // Unload the echo-cancel module now that the audio streams releasing its
        // virtual nodes have stopped. (Dropping the guard runs `pactl unload`.)
        drop(self.echo_cancel);

        crate::log_msg("Leaving room...");
        let _ = self.room_state.leave().await;
        // Close peer links with the graceful goodbye code so remotes evict us
        // promptly (not after the reconnect grace / slow gossip Leave), and stop
        // our supervisors so none redial the about-to-close endpoint.
        self.transport.leave().await;
        crate::log_msg("Room left");

        // NOTE: the router + endpoint are persistent (owned by the NetStack), so we
        // deliberately do NOT shut them down here — only this session's peer links
        // (closed by `transport.leave()` above) and tasks are torn down. The caller
        // clears the persistent `audio_router` so stray inbound links are dropped.

        crate::log_msg("Joining capture thread...");
        let _ = self.capture_thread.join();
        crate::log_msg("ActiveSession::shutdown complete");
    }
}

/// The app's persistent network stack: ONE endpoint + gossip + router, built once
/// and kept alive for the whole `run_core_loop` lifetime (rebuilt only on a network
/// -mode or identity change). Room sessions come and go on top of this — `join`
/// subscribes a gossip topic + binds a per-session transport to `audio_router`,
/// `leave` drops them — but the endpoint stays reachable between calls. That's the
/// W7 prerequisite: the friends presence listener needs an endpoint up while idle,
/// and a node id can only have ONE live endpoint instance (proven by the dual-
/// endpoint spike — see `docs/contacts-plan.md` P4).
struct NetStack {
    endpoint: Endpoint,
    gossip: Gossip,
    router: Router,
    /// The persistent inbound-audio handler on `router`; per-join we bind the
    /// active session's transport into it, and clear it on leave.
    audio_router: AudioRouter,
    /// In-memory address book (ticket + gossip fed), shared with every session.
    memory_lookup: iroh::address_lookup::memory::MemoryLookup,
}

impl NetStack {
    async fn shutdown(self) {
        crate::log_msg("Shutting down network stack (router + endpoint)...");
        let _ = tokio::time::timeout(Duration::from_secs(1), self.router.shutdown()).await;
        self.endpoint.close().await;
        crate::log_msg("Network stack shut down");
    }
}

/// Install the n0 DNS address-lookup services for a discovery `plan` (W7 P6), at
/// runtime, on an already-bound endpoint. The in-memory lookup (server-free, fed by
/// tickets + gossip) is always re-added; the n0 DNS *resolver* (`PkarrResolver` +
/// `DnsAddressLookup`, mirroring the `N0` preset) is added when `plan.resolver`; the
/// n0 DNS *publisher* (`PkarrPublisher`) when `plan.publisher`.
///
/// Idempotent and reversible: it builds the replacement services first, then clears
/// the service set and reinstalls exactly what the plan wants. Flipping `publisher`
/// off drops the publisher (its republish task ends when the last clone is dropped,
/// and the already-published record TTL-expires within ~30s) without an endpoint
/// rebuild and without disturbing resolution.
fn apply_discovery(
    endpoint: &Endpoint,
    memory_lookup: &iroh::address_lookup::memory::MemoryLookup,
    plan: crate::discovery::LookupPlan,
) -> Result<(), anyhow::Error> {
    use iroh::address_lookup::{
        AddressLookupBuilder, dns::DnsAddressLookup,
        pkarr::{PkarrPublisher, PkarrResolver},
    };
    let services = endpoint.address_lookup()?;
    let pkarr_resolver = if plan.resolver {
        Some(PkarrResolver::n0_dns().into_address_lookup(endpoint)?)
    } else {
        None
    };
    let dns_resolver = if plan.resolver {
        Some(DnsAddressLookup::n0_dns().into_address_lookup(endpoint)?)
    } else {
        None
    };
    let publisher = if plan.publisher {
        Some(PkarrPublisher::n0_dns().into_address_lookup(endpoint)?)
    } else {
        None
    };

    services.clear();
    // Always keep the local, server-free lookup (this is what ticket/gossip dialing
    // depends on — it must survive every posture, including DirectOnly).
    services.add(memory_lookup.clone());
    if let Some(pkarr_resolver) = pkarr_resolver {
        services.add(pkarr_resolver);
    }
    if let Some(dns_resolver) = dns_resolver {
        services.add(dns_resolver);
    }
    if let Some(publisher) = publisher {
        services.add(publisher);
    }
    Ok(())
}

/// Build the persistent network stack for the given identity + relay/discovery
/// posture. Binds the endpoint (the per-`NetworkMode` build hoisted out of the old
/// per-join path), spawns one gossip instance + one router accepting gossip and
/// audio, and kicks off `online()` in the background so app launch isn't blocked on
/// the relay handshake. `publish` is whether the presence posture is `Discoverable`
/// at build time (W7 P6) — it seeds the initial n0 DNS publish state.
async fn build_net_stack(
    secret_key: SecretKey,
    network_mode: NetworkMode,
    memory_lookup: iroh::address_lookup::memory::MemoryLookup,
    friends_handler: crate::presence_net::Handler,
    publish: bool,
) -> Result<NetStack, anyhow::Error> {
    // Bind with only the relay posture baked in (`Minimal` = crypto provider only,
    // relay on/off per mode); n0 DNS discovery is installed uniformly below via
    // `apply_discovery` so the Discoverable publish toggle is independent of the
    // network mode and can be flipped later at runtime. The in-memory lookup is added
    // here so dialing works even before the first `apply_discovery` (which re-adds it).
    let relay_mode = match network_mode {
        NetworkMode::DirectOnly => RelayMode::Disabled,
        NetworkMode::RelayNoDiscovery | NetworkMode::N0Full => RelayMode::Default,
    };
    let endpoint = Endpoint::builder(presets::Minimal)
        .secret_key(secret_key.clone())
        .relay_mode(relay_mode)
        .address_lookup(memory_lookup.clone())
        .bind()
        .await?;

    // Install the n0 DNS services for the current (mode, publish) posture.
    apply_discovery(&endpoint, &memory_lookup, crate::discovery::lookup_plan(network_mode, publish))?;

    // Bring the endpoint online in the background so launch isn't blocked on the
    // relay handshake; joins/sends just work once it's up (online() is idempotent).
    let ep = endpoint.clone();
    tokio::spawn(async move { ep.online().await });

    // One gossip instance for all rooms; sessions subscribe/unsubscribe topics on
    // it. Frame budget raised above the 4 KB default so a hard-capped custom avatar
    // (W4) can ride presence inline — all peers must use the same value.
    let gossip = Gossip::builder()
        .max_message_size(65536)
        .spawn(endpoint.clone());

    let audio_router = AudioRouter::new();
    // The friends presence listener (W7 B2) rides this same persistent router as a
    // third ALPN — it MUST be a handler here, not a standalone accept loop, since
    // the router owns endpoint.accept(). Policy (who we answer / what room we
    // report) is injected via `friends_handler`.
    let router = Router::builder(endpoint.clone())
        .accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone())
        .accept(crate::protocol::AUDIO_ALPN, audio_router.clone())
        .accept(
            crate::presence_net::FRIENDS_ALPN,
            crate::presence_net::FriendsProtocol::new(friends_handler),
        )
        .spawn();

    Ok(NetStack {
        endpoint,
        gossip,
        router,
        audio_router,
        memory_lookup,
    })
}

/// Finalize and clear the active recording, if any, emitting `RecordingStopped`.
/// No-op when not recording. Called on stop, room leave, and room switch so a
/// recording is always closed cleanly (its WAV size fields patched).
async fn stop_recording(
    recorder: &Arc<std::sync::Mutex<Option<crate::audio::recorder::Recorder>>>,
    is_recording: &Arc<AtomicBool>,
    multitrack: &Arc<std::sync::Mutex<Option<MultitrackRecorder>>>,
    is_multitrack: &Arc<AtomicBool>,
    ui_tx: &mpsc::Sender<UiEvent>,
) {
    is_recording.store(false, Ordering::Relaxed);
    is_multitrack.store(false, Ordering::Relaxed);
    // Exactly one slot is ever active for a given recording, but finalize both
    // defensively. The reported path is the file (mixed) or the session dir
    // (multitrack).
    let rec = recorder.lock().unwrap().take();
    if let Some(rec) = rec {
        let path = rec.path().to_string_lossy().to_string();
        if let Err(e) = rec.finalize() {
            crate::log_msg(&format!("Failed to finalize recording: {e}"));
        }
        crate::log_msg(&format!("Recording saved: {path}"));
        let _ = ui_tx.send(UiEvent::RecordingStopped { path }).await;
    }
    let mt = multitrack.lock().unwrap().take();
    if let Some(mt) = mt {
        let path = mt.dir().to_string_lossy().to_string();
        if let Err(e) = mt.finalize() {
            crate::log_msg(&format!("Failed to finalize multitrack recording: {e}"));
        }
        crate::log_msg(&format!("Multitrack recording saved: {path}/"));
        let _ = ui_tx.send(UiEvent::RecordingStopped { path }).await;
    }
}

/// Persist the (core-owned) friends store and tell the GUI the new snapshot. Skips
/// the disk write when `read_only` (a malformed load — A16: don't clobber it) but
/// still emits so the UI reflects the in-memory change. Snapshots under the lock,
/// then releases it before the async send.
async fn persist_and_emit_friends(
    friends: &Arc<std::sync::Mutex<crate::friends::FriendStore>>,
    read_only: bool,
    ui_tx: &mpsc::Sender<UiEvent>,
) {
    let store = friends.lock().unwrap().clone();
    if !read_only
        && let Err(e) = crate::friends::save(&store)
    {
        crate::log_msg(&format!("friends: save failed: {e:#}"));
    }
    let _ = ui_tx
        .send(UiEvent::FriendsUpdated { friends: store.friends, read_only })
        .await;
}

/// How often the outbound presence scheduler refreshes friends' status. Slow on
/// purpose — presence is best-effort, not real-time, and each pass opens a short
/// connection per friend.
const PING_INTERVAL: Duration = Duration::from_secs(60);
/// Delay before the FIRST presence pass, so the endpoint's background `online()`
/// has a moment to finish (otherwise the first probes fail and friends flash offline).
const PING_STARTUP_DELAY: Duration = Duration::from_secs(3);

/// One outbound presence-refresh pass (W7 B2): probe every friend that has a saved
/// address and emit their interpreted status. Friends with no saved address are
/// skipped (a bare id can't resolve without discovery) and stay offline in the UI
/// until first contact populates their address via `note_seen`. Probes run in
/// parallel (friend counts are small); an unreachable friend just yields nothing.
async fn probe_friends_once(
    endpoint: Endpoint,
    friends: crate::friends::FriendStore,
    ui_tx: mpsc::Sender<UiEvent>,
) {
    let mut set = tokio::task::JoinSet::new();
    for f in friends.list() {
        let Some(addr) = f.last_addr.clone() else { continue };
        let id = f.id;
        let ep = endpoint.clone();
        set.spawn(async move {
            match crate::presence_net::probe(&ep, addr).await {
                Ok((from, reply)) => crate::presence::interpret_pong(&reply, from).map(|p| (id, p)),
                Err(_) => None,
            }
        });
    }
    while let Some(res) = set.join_next().await {
        if let Ok(Some((id, presence))) = res {
            let _ = ui_tx.send(UiEvent::FriendPresence { id, presence }).await;
        }
    }
}

async fn run_core_loop(
    mut cmd_rx: mpsc::Receiver<CoreCommand>,
    ui_tx: mpsc::Sender<UiEvent>,
) -> Result<(), anyhow::Error> {
    let memory_lookup = iroh::address_lookup::memory::MemoryLookup::new();
    // Persistent identity (W7 P1): load a stable key so our node id survives
    // launches — the foundation for the friends-first contacts model. Fall back
    // to an ephemeral key only if the key file can't be read/created (e.g. no
    // writable config dir), so a bad disk never blocks getting on a call. The
    // `persisted` flag + error reason are surfaced to the UI (degraded state).
    let (mut secret_key, mut identity_error) = match crate::identity::load_or_create() {
        Ok(key) => (key, None),
        Err(e) => {
            let reason = format!("{e:#}");
            crate::log_msg(&format!(
                "identity: falling back to an ephemeral key (persistent load failed: {reason})"
            ));
            (iroh::SecretKey::generate(), Some(reason))
        }
    };
    // Tell the UI our node id + whether it's persisted. Re-sent after a regenerate.
    let _ = ui_tx
        .send(UiEvent::IdentityStatus {
            node_id: secret_key.public().to_string(),
            persisted: identity_error.is_none(),
            error: identity_error.clone(),
        })
        .await;
    // Peers seen per room, keyed by the room's join ticket, retained ACROSS leaves
    // so a rejoin can bootstrap to them. This is the fix for A8: the room creator's
    // own ticket lists only themselves as host, so on rejoin these retained peers
    // are their only dial targets. Keyed by TICKET (not one flat set) so that
    // revisiting a room after a detour through another one still remembers the
    // first room's peers — the old single-set version cleared them on any ticket
    // change, so an A→B→A bounce stranded the rejoiner with an empty bootstrap.
    // Inner map keyed by peer id so updates refresh the address.
    let known_peers: Arc<std::sync::Mutex<HashMap<String, HashMap<EndpointId, EndpointAddr>>>> =
        Arc::new(std::sync::Mutex::new(HashMap::new()));

    let audio_backend = Arc::new(PipeWireBackend::new());

    let is_muted = Arc::new(AtomicBool::new(false));
    let is_deafened = Arc::new(AtomicBool::new(false));
    let ptt_mode = Arc::new(AtomicBool::new(false));
    let ptt_active = Arc::new(AtomicBool::new(false));
    let noise_gate_threshold = Arc::new(std::sync::atomic::AtomicU32::new(0.01f32.to_bits()));
    // App-internal capture/playback gains (f32 bits), live-read by the audio loops.
    let input_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits()));
    let output_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits()));
    // Call recording: an optional live recorder (mic FIFO + WAV writer), shared
    // by the capture thread (pushes mic) and the mixer task (writes mix frames).
    // `is_recording` is a fast-path gate so the audio loops only take the lock
    // while a recording is actually running.
    let recorder: Arc<std::sync::Mutex<Option<crate::audio::recorder::Recorder>>> =
        Arc::new(std::sync::Mutex::new(None));
    let is_recording = Arc::new(AtomicBool::new(false));
    // Multitrack (stem) recording: a parallel recorder used instead of the mixed
    // `recorder` when `recording_mode` is Multitrack/Both. Exactly one of the two
    // slots is ever active. `is_multitrack` is the fast-path gate the audio loops
    // read (cheap) to decide whether to tap raw per-peer stems this cycle.
    let multitrack: Arc<std::sync::Mutex<Option<MultitrackRecorder>>> =
        Arc::new(std::sync::Mutex::new(None));
    let is_multitrack = Arc::new(AtomicBool::new(false));
    let mut recording_mode = RecordingMode::default();
    let peer_volumes = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new()));
    let peer_eq = Arc::new(Mutex::new(HashMap::<EndpointId, EqSettings>::new()));
    let peer_pan = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new()));
    // Peers locally muted by us: decoded for level metering but not mixed.
    let locally_muted = Arc::new(Mutex::new(HashSet::<EndpointId>::new()));
    let mut current_name = "Anonymous".to_string();
    // Our chosen avatar (W4), set on Join and changeable via SetAvatar; included
    // in every self-state we announce over presence.
    let mut current_avatar = crate::avatar::Avatar::default();
    let mut network_mode = NetworkMode::default();
    // Pixelpass binary override (config), and the ticket of our own active screen
    // share (rides our presence so the room — incl. late joiners — can watch).
    let mut pixelpass_override: Option<String> = None;
    let mut current_sharing: Option<String> = None;

    let mut active_session: Option<ActiveSession> = None;
    // Standalone capture-only mic meter, live only when no session exists.
    let mut mic_monitor: Option<MicMonitor> = None;

    // Friends store (W7) — core now OWNS it (was the GUI's). Shared so the idle
    // listener + ping scheduler read it. A malformed load yields an EMPTY store
    // flagged read-only, so we never save over the damaged file (A16); the GUI
    // surfaces the degraded state from `FriendsUpdated { read_only: true }`.
    let (initial_friends, friends_read_only) = match crate::friends::load() {
        Ok(store) => (store, false),
        Err(e) => {
            crate::log_msg(&format!(
                "friends: load failed, starting read-only to avoid clobbering the file: {e:#}"
            ));
            (crate::friends::FriendStore::default(), true)
        }
    };
    let friends = Arc::new(std::sync::Mutex::new(initial_friends));
    // Our presence posture, shared with the listener + scheduler. The GUI pushes it
    // at startup + on change via SetPresenceMode; persistence stays in AppConfig.
    let presence_mode = Arc::new(std::sync::Mutex::new(crate::presence::PresenceMode::default()));
    // The gathering we're currently in (its restamped member ticket + label), so the
    // listener can offer friends a one-click Join. `None` when not in a call. Set on
    // Join, cleared on Leave.
    let current_room: Arc<std::sync::Mutex<Option<crate::presence::RoomPresence>>> =
        Arc::new(std::sync::Mutex::new(None));

    // Reply policy for the idle friends listener (B2): answer friends only, never
    // while invisible (`should_answer`), and report our current gathering so a friend
    // can one-click join. Reads the shared snapshots, so it stays correct as they
    // change and survives a network-stack rebuild. Pure-sync (no awaits, no lock held
    // across one). Built once and handed to every `build_net_stack`.
    let friends_handler: crate::presence_net::Handler = {
        let friends = friends.clone();
        let presence_mode = presence_mode.clone();
        let current_room = current_room.clone();
        Arc::new(move |from| {
            let mode = *presence_mode.lock().unwrap();
            let allowed = crate::presence::should_answer(&from, &friends.lock().unwrap(), mode);
            if !allowed {
                return None;
            }
            let room = current_room.lock().unwrap().clone();
            Some(crate::presence::ControlMsg::Pong { room })
        })
    };

    // The persistent network stack (endpoint + gossip + router), built once at
    // startup and kept alive for the app's lifetime. Room sessions ride on top of
    // it (subscribe a topic + bind the audio router on join, clear on leave); it's
    // rebuilt only when the network mode or identity changes. Moving `memory_lookup`
    // in — all later access is via `net.memory_lookup`.
    let initial_publish = presence_mode.lock().unwrap().publishes_to_discovery();
    let mut net = match build_net_stack(secret_key.clone(), network_mode, memory_lookup, friends_handler.clone(), initial_publish).await {
        Ok(stack) => stack,
        Err(e) => {
            // Only a local socket bind can fail here (the relay handshake is
            // backgrounded), so this is fatal to networking — surface it rather
            // than dying silently.
            let _ = ui_tx
                .send(UiEvent::Error(format!("Failed to start networking: {e}")))
                .await;
            return Err(e);
        }
    };
    // Set when a network-mode / identity change arrives mid-call; the stack is
    // rebuilt on the next Leave (or before the next Join), preserving the old
    // "applies on next join" semantics while keeping the endpoint up while idle.
    let mut net_rebuild_pending = false;

    // When Discoverable is on, the instant it auto-reverts to Normal (W7 P6 time-box).
    // `None` = not Discoverable, no pending revert. Set on SetPresenceMode(Discoverable),
    // cleared on any other posture, and consumed by the select! branch below.
    let mut discovery_deadline: Option<tokio::time::Instant> = None;

    // Tell the GUI the loaded friends list (it renders from this, no longer owning
    // it). Snapshot under the lock, then release it before the async send.
    let initial_snapshot = friends.lock().unwrap().list().to_vec();
    let _ = ui_tx
        .send(UiEvent::FriendsUpdated {
            friends: initial_snapshot,
            read_only: friends_read_only,
        })
        .await;

    // Drive the command loop AND the outbound presence ping scheduler together: the
    // scheduler can't block command handling, so a probe pass runs in a detached task
    // on each tick. `interval_at` delays the first pass so the endpoint can come
    // online first.
    let mut ping_interval = tokio::time::interval_at(
        tokio::time::Instant::now() + PING_STARTUP_DELAY,
        PING_INTERVAL,
    );
    ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    loop {
        let cmd = tokio::select! {
            maybe_cmd = cmd_rx.recv() => match maybe_cmd {
                Some(cmd) => cmd,
                None => break,
            },
            _ = ping_interval.tick() => {
                // Fully dark while Invisible (the user's choice): don't even probe,
                // so nothing we do touches a friend's machine. Otherwise refresh in a
                // detached task capturing the CURRENT endpoint (a stack rebuild between
                // ticks is naturally picked up next tick).
                if *presence_mode.lock().unwrap() != crate::presence::PresenceMode::Invisible {
                    tokio::spawn(probe_friends_once(
                        net.endpoint.clone(),
                        friends.lock().unwrap().clone(),
                        ui_tx.clone(),
                    ));
                }
                continue;
            }
            // W7 P6 time-box: Discoverable auto-reverts to Normal after DISCOVERY_TIMEBOX
            // so a publish beacon never stands indefinitely. The branch is disabled
            // (`if` guard) unless a deadline is armed; `unwrap_or_else` is unreachable
            // belt-and-braces. On fire: stop publishing first, then commit Normal only
            // if the endpoint's discovery services accepted the non-publishing plan.
            _ = tokio::time::sleep_until(
                discovery_deadline.unwrap_or_else(tokio::time::Instant::now),
            ), if discovery_deadline.is_some() => {
                let previous_mode = *presence_mode.lock().unwrap();
                if previous_mode != PresenceMode::Discoverable {
                    discovery_deadline = None;
                    continue;
                }

                let requested_mode = PresenceMode::Normal;
                let now = tokio::time::Instant::now();
                let plan = crate::discovery::lookup_plan(
                    network_mode,
                    requested_mode.publishes_to_discovery(),
                );
                let apply_result = apply_discovery(&net.endpoint, &net.memory_lookup, plan);
                let (committed_mode, transition_error) =
                    crate::discovery::resolve_presence_transition(
                        previous_mode,
                        requested_mode,
                        apply_result.is_ok(),
                    );
                *presence_mode.lock().unwrap() = committed_mode;
                discovery_deadline = if committed_mode == PresenceMode::Discoverable {
                    Some(now + DISCOVERY_REVERT_RETRY)
                } else {
                    None
                };

                match apply_result {
                    Ok(()) => {
                        crate::log_msg(
                            "discovery: Discoverable time-box elapsed → reverting to Normal",
                        );
                        let _ = ui_tx
                            .send(UiEvent::PresenceModeReverted {
                                mode: PresenceMode::Normal,
                            })
                            .await;
                    }
                    Err(e) => {
                        crate::log_msg(&format!("discovery: time-box revert failed: {e:#}"));
                        if committed_mode != requested_mode {
                            let _ = ui_tx
                                .send(UiEvent::PresenceModeReverted {
                                    mode: committed_mode,
                                })
                                .await;
                        }
                        if let Some(message) = transition_error {
                            let _ = ui_tx
                                .send(UiEvent::Error(format!("{message} ({e:#})")))
                                .await;
                        }
                    }
                }
                continue;
            }
        };
        match cmd {
            CoreCommand::Shutdown => {
                crate::log_msg("Core shutdown requested");
                // Finalize recordings while capture/mixer feeders are still alive.
                stop_recording(&recorder, &is_recording, &multitrack, &is_multitrack, &ui_tx).await;
                stop_mic_monitor(&audio_backend, mic_monitor.take());

                if let Some(session) = active_session.take() {
                    session.shutdown(audio_backend.clone()).await;
                    net.audio_router.clear();
                }
                *current_room.lock().unwrap() = None;

                net.shutdown().await;
                let _ = ui_tx.send(UiEvent::ShutdownComplete).await;
                break;
            }

            CoreCommand::Join { name, ticket, room_name, input_device, output_device, echo_cancellation, avatar } => {
                current_name = name.clone();
                current_avatar = avatar;

                // Finalize any recording before tearing down the old session — its
                // capture/mixer feeders are about to stop.
                stop_recording(&recorder, &is_recording, &multitrack, &is_multitrack, &ui_tx).await;

                // Clean up any existing session FIRST (this calls `transport.leave()`
                // on the current endpoint), before any stack rebuild closes it.
                if let Some(session) = active_session.take() {
                    crate::log_msg("Shutting down existing active session");
                    session.shutdown(audio_backend.clone()).await;
                    net.audio_router.clear();
                }

                // If a network-mode / identity change was deferred while a call was
                // active, rebuild the persistent stack now — after the old session is
                // gone, before the new one binds — so this join uses the new posture.
                if net_rebuild_pending {
                    let lookup = net.memory_lookup.clone();
                    net.shutdown().await;
                    let publish = presence_mode.lock().unwrap().publishes_to_discovery();
                    net = build_net_stack(secret_key.clone(), network_mode, lookup, friends_handler.clone(), publish).await?;
                    net_rebuild_pending = false;
                }

                // Release a standalone mic monitor if running — it shares the
                // backend's single capture stream, so it must stop before the
                // call claims it. (Safe here: any session was just shut down.)
                stop_mic_monitor(&audio_backend, mic_monitor.take());

                // Use the app's persistent endpoint (bound once at startup); the
                // per-join endpoint build is gone. A node id has exactly one live
                // endpoint instance (W7 spike), so the friends listener and room
                // audio necessarily share this one. online() is idempotent — it just
                // ensures we're relay-reachable before joining the gossip swarm.
                let endpoint = net.endpoint.clone();
                endpoint.online().await;

                // Determine target ticket
                let ticket_str = if ticket.trim().is_empty() || ticket == "create" {
                    let topic_id: [u8; 32] = rand::random();
                    let host_addr = endpoint.addr();
                    crate::log_msg(&format!(
                        "Creating room. host_id={}, host_addrs={}, topic={}",
                        crate::short_id(&host_addr.id.to_string()),
                        host_addr.addrs.len(),
                        crate::short_bytes_hex(&topic_id)
                    ));
                    // The creator's chosen cosmetic label rides in the ticket so
                    // every joiner inherits it; sanitize it before it leaves here.
                    let label = crate::sanitize::sanitize_name(&room_name);
                    let ticket = PeerSpeakTicket { host_addr, topic_id, name: label };
                    ticket.to_string()
                } else {
                    let ticket_str = ticket.trim().to_string();
                    crate::log_msg(&format!(
                        "Joining room with existing ticket={}",
                        crate::redact_for_log(&ticket_str)
                    ));
                    ticket_str
                };

                // Per-session transport over the persistent endpoint, bound to the
                // persistent audio router so this call's inbound audio links route
                // to it (cleared on leave). Gossip + router are persistent on the
                // NetStack; the session just subscribes its topic below.
                let transport = Arc::new(IrohTransport::new(endpoint.clone()));
                net.audio_router.bind(&transport);

                let room_state = Arc::new(IrohGossipState::new(
                    endpoint.clone(),
                    net.gossip.clone(),
                    net.memory_lookup.clone(),
                    secret_key.clone(),
                ));

                // Fresh join starts not sharing; clear any stale share ticket.
                current_sharing = None;
                let self_state = PeerState {
                    name: current_name.clone(),
                    is_muted: is_muted.load(Ordering::Relaxed),
                    addr: endpoint.addr(),
                    sharing: None,
                    avatar: current_avatar.clone(),
                };

                // Snapshot THIS room's retained peers (by ticket) as extra bootstrap
                // targets so a rejoin can dial them (A8) — including after a detour
                // through another room, since the per-ticket archive isn't cleared.
                // Resolution rides the persistent address book.
                let extra_bootstrap: Vec<EndpointAddr> = known_peers
                    .lock()
                    .unwrap()
                    .get(&ticket_str)
                    .map(|peers| peers.values().cloned().collect())
                    .unwrap_or_default();

                let extra_bootstrap_ids = extra_bootstrap
                    .iter()
                    .map(|a| crate::short_id(&a.id.to_string()))
                    .collect::<Vec<_>>();
                crate::log_msg(&format!(
                    "Attempting room_state.join self_id={}, self_name={:?}, sharing={}, extra_bootstrap={:?}",
                    crate::short_id(&self_state.addr.id.to_string()),
                    self_state.name,
                    self_state.sharing.is_some(),
                    extra_bootstrap_ids
                ));
                if let Err(e) = room_state.join(&ticket_str, self_state.clone(), extra_bootstrap).await {
                    crate::log_msg(&format!("Error room_state.join failed: {:?}", e));
                    let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await;
                    net.audio_router.clear();
                    continue;
                }
                crate::log_msg("Joined room successfully via room_state");

                // Setup raw audio channels
                let (capture_tx, capture_rx) = std::sync::mpsc::channel();
                let (playback_tx, playback_rx) = std::sync::mpsc::channel();

                // Echo cancellation: if enabled, load PipeWire's echo-cancel module
                // bound to the chosen real devices and route capture/playback
                // through its virtual nodes (the sink doubles as the AEC reference).
                // The guard unloads the module on drop — including the early-return
                // paths below, since it's a local until moved into the session. On
                // any failure, warn and fall back to the direct devices.
                let mut echo_cancel_guard = None;
                let (capture_target, playback_target) = if echo_cancellation {
                    match crate::audio::echo_cancel::enable(
                        input_device.as_deref(),
                        output_device.as_deref(),
                    ) {
                        Ok(guard) => {
                            echo_cancel_guard = Some(guard);
                            crate::log_msg("Echo cancellation enabled");
                            (
                                Some(crate::audio::echo_cancel::EC_SOURCE.to_string()),
                                Some(crate::audio::echo_cancel::EC_SINK.to_string()),
                            )
                        }
                        Err(e) => {
                            crate::log_msg(&format!(
                                "Echo cancellation unavailable, using direct devices: {e}"
                            ));
                            let _ = ui_tx
                                .send(UiEvent::Error(format!("Echo cancellation unavailable: {e}")))
                                .await;
                            (input_device.clone(), output_device.clone())
                        }
                    }
                } else {
                    (input_device.clone(), output_device.clone())
                };

                if let Err(e) = audio_backend.start_capture(capture_tx, capture_target) {
                    let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await;
                    let _ = room_state.leave().await;
                    net.audio_router.clear();
                    continue;
                }

                // Shared gauge: PipeWire publishes the playback ring's live depth
                // here (drain side + fill side); the mixer reads it to pace
                // production to the hardware clock instead of a fixed timer.
                let ring_fill = Arc::new(AtomicUsize::new(0));
                if let Err(e) = audio_backend.start_playback(playback_rx, playback_target, ring_fill.clone()) {
                    let _ = ui_tx.send(UiEvent::Error(format!("Failed to start playback: {}", e))).await;
                    let _ = audio_backend.stop();
                    let _ = room_state.leave().await;
                    net.audio_router.clear();
                    continue;
                }

                let jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>> = Arc::new(Mutex::new(HashMap::new()));

                // 1. Capture & encoding thread
                let is_muted_clone = is_muted.clone();
                let ptt_mode_clone = ptt_mode.clone();
                let ptt_active_clone = ptt_active.clone();
                let noise_gate_threshold_clone = noise_gate_threshold.clone();
                let input_gain_clone = input_gain.clone();
                let transport_clone = transport.clone();
                let ui_tx_capture = ui_tx.clone();
                let recorder_capture = recorder.clone();
                let is_recording_capture = is_recording.clone();
                let multitrack_capture = multitrack.clone();
                let is_multitrack_capture = is_multitrack.clone();

                let capture_thread = std::thread::spawn(move || {
                    use opus::{Channels, Application};
                    let mut encoder = match OpusEncoder::new(48000, Channels::Mono, Application::Voip) {
                        Ok(enc) => enc,
                        Err(e) => {
                            crate::log_msg(&format!("Capture thread error: {:?}", e));
                            return;
                        }
                    };
                    // Per-sender packet sequence number, prepended to every frame so
                    // receivers can reorder and conceal loss. Wraps after ~years.
                    let mut seq: u32 = 0;
                    // Smooth noise gate (hysteresis + attack/release + hangover),
                    // carrying envelope state across frames. The live slider value
                    // is read per frame so changes apply immediately.
                    let mut gate = crate::audio::gate::NoiseGate::new(48000);
                    // Peak-held raw mic level for the settings meter, reported
                    // pre-gate/pre-mute so calibration reflects the true input.
                    let mut mic_meter = MicLevelMeter::new();

                    while let Ok(mut pcm) = capture_rx.recv() {
                        // Apply the input gain first so the meter, gate, and what we
                        // transmit all reflect the same (gained) signal.
                        apply_volume(&mut pcm, f32::from_bits(input_gain_clone.load(Ordering::Relaxed)));

                        if let Some(peak) = mic_meter.push(&pcm) {
                            let _ = ui_tx_capture.try_send(UiEvent::MicLevel(peak));
                        }

                        if is_muted_clone.load(Ordering::Relaxed) {
                            continue;
                        }
                        if ptt_mode_clone.load(Ordering::Relaxed) && !ptt_active_clone.load(Ordering::Relaxed) {
                            continue;
                        }

                        let ng_bits = noise_gate_threshold_clone.load(Ordering::Relaxed);
                        let ng_thresh = f32::from_bits(ng_bits);
                        // Apply the gate in place; skip transmitting a fully-closed
                        // frame so we don't send pure silence (the receiver's jitter
                        // buffer conceals the gap).
                        if !gate.process(&mut pcm, ng_thresh) {
                            continue;
                        }

                        // Record what we transmit (post-gain, post-gate, post-mute):
                        // this is exactly the mic audio peers receive from us. The
                        // mixer task pairs it with the incoming mix.
                        if is_recording_capture.load(Ordering::Relaxed) {
                            if is_multitrack_capture.load(Ordering::Relaxed) {
                                if let Some(mt) = multitrack_capture.lock().unwrap().as_mut() {
                                    mt.push_mic(&pcm);
                                }
                            } else if let Some(rec) = recorder_capture.lock().unwrap().as_mut() {
                                rec.push_mic(&pcm);
                            }
                        }

                        if let Ok(encoded) = encoder.encode(&pcm) {
                            // Frame on the wire: [seq: u32 LE][opus payload].
                            let mut packet = Vec::with_capacity(4 + encoded.len());
                            packet.extend_from_slice(&seq.to_le_bytes());
                            packet.extend_from_slice(&encoded);
                            seq = seq.wrapping_add(1);
                            transport_clone.broadcast(bytes::Bytes::from(packet));
                        }
                    }
                });

                // 2. Receiver task: parse the sequence header and hand each packet
                //    to that peer's jitter buffer. Decoding happens later, on the
                //    playout side, so loss can be concealed at the right moment.
                let transport_recv = transport.clone();
                let jitter_recv = jitter.clone();
                let datagram_task = tokio::spawn(async move {
                    let mut datagram_rx = match transport_recv.receive_datagrams().await {
                        Ok(rx) => rx,
                        Err(e) => {
                            crate::log_msg(&format!("Receiver task error: {:?}", e));
                            return;
                        }
                    };

                    while let Some((from_peer, bytes)) = datagram_rx.recv().await {
                        if !transport_recv.audio_sender_admitted(from_peer) {
                            continue;
                        }
                        if !audio_datagram_len_ok(bytes.len()) {
                            // Malformed (< sequence header) or oversized Opus payload.
                            continue;
                        }
                        let seq = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
                        let payload = bytes[4..].to_vec();

                        let mut guard = jitter_recv.lock().await;
                        let buffer = match guard.entry(from_peer) {
                            std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(),
                            std::collections::hash_map::Entry::Vacant(entry) => {
                                match JitterBuffer::new() {
                                    Ok(jb) => entry.insert(jb),
                                    Err(e) => {
                                        crate::log_msg(&format!("Failed to init jitter buffer for {:?}: {:?}", from_peer, e));
                                        continue;
                                    }
                                }
                            }
                        };
                        buffer.insert(seq, payload);
                    }
                });

                // 3. Mixing & level extraction loop task. Production is paced by
                //    the playback ring's fill level (the PipeWire hardware clock),
                //    NOT a fixed software timer: we produce a 20ms frame only when
                //    the ring is below its target depth, so the long-run mix rate
                //    auto-matches the device drain rate and the producer/consumer
                //    beat (which otherwise churns ~20% of audio) disappears. Each
                //    produced frame pulls one concealed frame per peer from its
                //    jitter buffer, applies per-peer volume, and sums.
                let jitter_mixer = jitter.clone();
                let is_deafened_clone = is_deafened.clone();
                let peer_volumes_mixer = peer_volumes.clone();
                let peer_eq_mixer = peer_eq.clone();
                let peer_pan_mixer = peer_pan.clone();
                let locally_muted_mixer = locally_muted.clone();
                let output_gain_mixer = output_gain.clone();
                let ui_tx_mixer = ui_tx.clone();
                let ring_fill_mixer = ring_fill.clone();
                let recorder_mixer = recorder.clone();
                let is_recording_mixer = is_recording.clone();
                let multitrack_mixer = multitrack.clone();
                let is_multitrack_mixer = is_multitrack.clone();
                let mixer_task = tokio::spawn(async move {
                    // Mix-bus soft limiter: rides loud multi-peer moments down to
                    // the ceiling instead of hard-clipping. State carries across
                    // frames (see audio::limiter).
                    let mut limiter = crate::audio::limiter::SoftLimiter::new(48_000);
                    // Per-peer EQ filter state. Settings are live-cloned each
                    // cycle; state is rebuilt only when a peer's EQ changes.
                    let mut peer_eqs: HashMap<EndpointId, Eq> = HashMap::new();
                    // When the ring is at/above target we have nothing to do; nap
                    // briefly and re-check. Short enough (relative to the ~60ms
                    // target and ~21ms device quantum) that we always refill well
                    // before the ring can run dry.
                    const IDLE_NAP: Duration = Duration::from_millis(2);

                    // Pushing a level event per frame floods the UI runtime at
                    // ~50/sec. We peak-hold per-peer levels across this many
                    // produced frames and emit once per window (~10/sec) —
                    // peak-hold so a brief transient still lights the indicator.
                    const LEVEL_EMIT_FRAMES: u32 = 5;
                    let mut level_peaks: HashMap<EndpointId, f32> = HashMap::new();
                    let mut frames_since_emit: u32 = 0;

                    loop {
                        // Pace to the hardware clock: only produce while the ring
                        // is draining below target. Otherwise yield and re-check.
                        if ring_fill_mixer.load(Ordering::Relaxed) >= crate::audio::PLAYBACK_TARGET_SAMPLES {
                            tokio::time::sleep(IDLE_NAP).await;
                            continue;
                        }

                        let current_volumes = peer_volumes_mixer.lock().await.clone();
                        let current_eq = peer_eq_mixer.lock().await.clone();
                        let current_pans = peer_pan_mixer.lock().await.clone();
                        let muted_peers = locally_muted_mixer.lock().await.clone();
                        let mut peer_frames: Vec<(Vec<i16>, f32)> = Vec::new();
                        let mut peers_seen = HashSet::new();

                        // Multitrack stem capture: tap each peer's RAW decoded frame
                        // (pre-volume, pre-mute, pre-limiter) so the stems are clean
                        // source. Only collected while a multitrack recording is live.
                        let mt_active = is_recording_mixer.load(Ordering::Relaxed)
                            && is_multitrack_mixer.load(Ordering::Relaxed);
                        let mut stems: Vec<(EndpointId, Vec<i16>)> = Vec::new();

                        {
                            let mut guard = jitter_mixer.lock().await;
                            for (&peer_id, buffer) in guard.iter_mut() {
                                // `None` means idle/buffering: contribute nothing,
                                // but keep a (zero) entry so the UI sees it idle.
                                let Some(mut frame) = buffer.pop_frame() else {
                                    level_peaks.entry(peer_id).or_insert(0.0);
                                    continue;
                                };

                                if mt_active {
                                    stems.push((peer_id, frame.clone()));
                                }

                                let vol = current_volumes.get(&peer_id).copied().unwrap_or(1.0);
                                apply_volume(&mut frame, vol);

                                let eq_settings = current_eq
                                    .get(&peer_id)
                                    .copied()
                                    .unwrap_or_default()
                                    .clamped();
                                if eq_settings.is_flat() {
                                    peer_eqs.remove(&peer_id);
                                } else {
                                    let needs_rebuild = peer_eqs
                                        .get(&peer_id)
                                        .map(|eq| eq.settings() != eq_settings)
                                        .unwrap_or(true);
                                    if needs_rebuild {
                                        peer_eqs.insert(peer_id, Eq::new(eq_settings));
                                    }
                                    if let Some(eq) = peer_eqs.get_mut(&peer_id) {
                                        eq.process_frame(&mut frame);
                                    }
                                }

                                // Level is recorded even for locally-muted peers so
                                // the UI still shows that they're speaking.
                                let peak = level_peaks.entry(peer_id).or_insert(0.0);
                                *peak = peak.max(frame_level(&frame));
                                peers_seen.insert(peer_id);

                                // Locally muted: decoded above (jitter buffer advances,
                                // level shown) but not mixed into our output.
                                if muted_peers.contains(&peer_id) {
                                    continue;
                                }

                                let pan = current_pans
                                    .get(&peer_id)
                                    .copied()
                                    .unwrap_or(0.0)
                                    .clamp(-1.0, 1.0);
                                peer_frames.push((frame, pan));
                            }
                        }
                        peer_eqs.retain(|id, _| peers_seen.contains(id) || current_eq.contains_key(id));

                        // Lossless i32 sum, then the limiter applies the master
                        // output gain (in f32, so a boost past the ceiling is
                        // limited too) and rides peaks down to the ceiling.
                        let mixed_sum = mix_stereo_frames(&peer_frames, FRAME_SAMPLES);
                        let out_gain = f32::from_bits(output_gain_mixer.load(Ordering::Relaxed));
                        let mixed = limiter.process(&mixed_sum, out_gain);
                        let record_mix = stereo_to_mono(&mixed);

                        // Record the true call audio, independent of local deafen —
                        // deafen only silences our own monitor, not what the call
                        // carried. Multitrack writes raw per-peer stems (+ the mixed
                        // track in Both mode) one aligned frame per cycle; Mixed mode
                        // writes the single blended file as before.
                        if mt_active {
                            if let Some(mt) = multitrack_mixer.lock().unwrap().as_mut() {
                                let res = (|| -> std::io::Result<()> {
                                    for (id, f) in &stems {
                                        mt.write_peer(*id, f)?;
                                    }
                                    mt.write_mix(&record_mix)?;
                                    mt.end_cycle()
                                })();
                                if let Err(e) = res {
                                    crate::log_msg(&format!("Multitrack write failed: {e}"));
                                }
                            }
                        } else if is_recording_mixer.load(Ordering::Relaxed)
                            && let Some(rec) = recorder_mixer.lock().unwrap().as_mut()
                            && let Err(e) = rec.write_frame(&record_mix)
                        {
                            crate::log_msg(&format!("Recording write failed: {e}"));
                        }

                        let frame_to_send = if is_deafened_clone.load(Ordering::Relaxed) {
                            vec![0i16; mixed.len()]
                        } else {
                            mixed
                        };

                        if playback_tx.send(frame_to_send).is_err() {
                            break;
                        }

                        // Emit coalesced peaks once per window, then reset.
                        frames_since_emit += 1;
                        if frames_since_emit >= LEVEL_EMIT_FRAMES {
                            let levels: Vec<(EndpointId, f32)> = level_peaks.drain().collect();
                            let _ = ui_tx_mixer.send(UiEvent::AudioLevels(levels)).await;
                            frames_since_emit = 0;
                        }
                    }
                });

                // 4. Room event subscriber task
                let mut room_events = match room_state.subscribe_events().await {
                    Ok(rx) => rx,
                    Err(e) => {
                        let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe events: {}", e))).await;
                        continue;
                    }
                };
                let ui_tx_events = ui_tx.clone();
                let jitter_events = jitter.clone();
                let transport_events = transport.clone();
                let grace_timers: GraceTimers = Arc::new(std::sync::Mutex::new(HashMap::new()));
                let grace_timers_events = grace_timers.clone();
                let seen_connected: SeenConnected = Arc::new(std::sync::Mutex::new(HashSet::new()));
                let seen_connected_events = seen_connected.clone();
                let multitrack_events = multitrack.clone();
                let is_multitrack_events = is_multitrack.clone();
                let known_peers_events = known_peers.clone();
                // The ticket of the room this event loop serves, so peer add/remove
                // updates the right per-ticket bucket in `known_peers` (A8 archive).
                let ticket_events = ticket_str.clone();
                // Friends store + ui sender, so a connected peer who is a friend has
                // their saved address auto-healed (W7) — populates `last_addr` so the
                // presence scheduler can reach them later.
                let friends_events = friends.clone();
                let friends_read_only_events = friends_read_only;
                let event_task = tokio::spawn(async move {
                    while let Some(event) = room_events.recv().await {
                        match event {
                            RoomEvent::PeerJoined(peer_id, state) => {
                                // A (re)join means the peer is back — cancel any
                                // pending reconnect grace timer before re-adding it.
                                cancel_grace_timer(&grace_timers_events, &peer_id);
                                transport_events.admit_audio_sender(peer_id);
                                // Establish the audio connection as soon as the peer
                                // is known (the transport dedupes the full-mesh race).
                                // Hand over the full address so reconnects can dial
                                // it directly rather than via the gossip lookup.
                                transport_events.connect_peer(state.addr.clone()).await;
                                // Auto-heal a friend's saved address (W7): if this
                                // peer is a friend, refresh their last_addr so the
                                // presence scheduler can reach them between rooms.
                                let healed = friends_events
                                    .lock()
                                    .unwrap()
                                    .note_seen(&peer_id, state.addr.clone());
                                if healed {
                                    persist_and_emit_friends(
                                        &friends_events,
                                        friends_read_only_events,
                                        &ui_tx_events,
                                    )
                                    .await;
                                }
                                // Retain this peer under this room's ticket as a
                                // future rejoin bootstrap target (A8).
                                known_peers_events
                                    .lock()
                                    .unwrap()
                                    .entry(ticket_events.clone())
                                    .or_default()
                                    .insert(peer_id, state.addr.clone());
                                // If a multitrack recording is live, give this peer
                                // its own stem track (silence-padded back to t=0).
                                if is_multitrack_events.load(Ordering::Relaxed)
                                    && let Some(mt) = multitrack_events.lock().unwrap().as_mut()
                                    && let Err(e) = mt.add_peer(peer_id, &state.name)
                                {
                                    crate::log_msg(&format!("multitrack add_peer (join) failed: {e}"));
                                }
                                let _ = ui_tx_events.send(UiEvent::PeerJoined { id: peer_id, state }).await;
                            }
                            RoomEvent::PeerLeft(peer_id) => {
                                // Graceful leave — evict immediately.
                                cancel_grace_timer(&grace_timers_events, &peer_id);
                                seen_connected_events.lock().unwrap().remove(&peer_id);
                                // Graceful leave: drop them as a rejoin dial target
                                // for this room (a transient PeerConnectionLost
                                // deliberately does NOT, so we can still re-dial a
                                // peer who's still up).
                                if let Some(peers) =
                                    known_peers_events.lock().unwrap().get_mut(&ticket_events)
                                {
                                    peers.remove(&peer_id);
                                }
                                transport_events.remove_audio_sender(peer_id);
                                transport_events.disconnect_peer(peer_id).await;
                                jitter_events.lock().await.remove(&peer_id);
                                let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await;
                            }
                            RoomEvent::PeerUpdated(peer_id, state) => {
                                // A re-announce means the peer is alive — cancel any
                                // pending grace timer. It may also carry a fresh
                                // address (peer back on a new network); refresh the
                                // retained dial target so a later reconnect re-reaches
                                // it. Idempotent: an ordinary mute/unmute update just
                                // re-records the same address.
                                cancel_grace_timer(&grace_timers_events, &peer_id);
                                transport_events.admit_audio_sender(peer_id);
                                transport_events.connect_peer(state.addr.clone()).await;
                                // Auto-heal a friend's saved address (W7) on the
                                // re-announce too — this is the path that catches a
                                // friend who moved networks mid-session.
                                let healed = friends_events
                                    .lock()
                                    .unwrap()
                                    .note_seen(&peer_id, state.addr.clone());
                                if healed {
                                    persist_and_emit_friends(
                                        &friends_events,
                                        friends_read_only_events,
                                        &ui_tx_events,
                                    )
                                    .await;
                                }
                                // Refresh this room's retained rejoin target with the
                                // fresh addr (A8).
                                known_peers_events
                                    .lock()
                                    .unwrap()
                                    .entry(ticket_events.clone())
                                    .or_default()
                                    .insert(peer_id, state.addr.clone());
                                let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await;
                            }
                            RoomEvent::ChatMessage { from, name, text, .. } => {
                                let _ = ui_tx_events.send(UiEvent::ChatMessage {
                                    from: from.to_string(),
                                    name,
                                    text,
                                }).await;
                            }
                            RoomEvent::PeerConnectionLost(peer_id) => {
                                // Transient drop: do NOT tear down the peer. Its audio
                                // supervisor stays alive and keeps redialing the
                                // retained address, so show "reconnecting" and arm a
                                // grace timer that evicts the peer only if the link
                                // hasn't recovered within RECONNECT_GRACE. A gossip
                                // rejoin (PeerJoined/PeerUpdated) or a transport
                                // reconnect (ConnEvent::Connected) cancels it first.
                                transport_events.keep_audio_sender_for_reconnect_grace(peer_id);
                                let _ = ui_tx_events.send(UiEvent::PeerConnecting { id: peer_id }).await;
                                arm_grace_timer(
                                    &grace_timers_events,
                                    &seen_connected_events,
                                    &transport_events,
                                    &jitter_events,
                                    &ui_tx_events,
                                    RECONNECT_GRACE,
                                    peer_id,
                                );
                            }
                        }
                    }
                });

                // 5. Connection-state forwarder: turns transport link state into
                //    per-peer UI indicators (connecting / reconnecting vs. live).
                let mut conn_events = match transport.subscribe_conn_events().await {
                    Ok(rx) => rx,
                    Err(e) => {
                        let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe conn events: {}", e))).await;
                        continue;
                    }
                };
                let conn_handler = ConnEventHandler::new(
                    ui_tx.clone(),
                    grace_timers.clone(),
                    seen_connected.clone(),
                    transport.clone(),
                    jitter.clone(),
                );
                let conn_event_task = tokio::spawn(async move {
                    while let Some(event) = conn_events.recv().await {
                        conn_handler.handle(event).await;
                    }
                });

                let session = ActiveSession {
                    room_state: room_state.clone(),
                    capture_thread,
                    datagram_task,
                    mixer_task,
                    event_task,
                    conn_event_task,
                    grace_timers,
                    transport: transport.clone(),
                    echo_cancel: echo_cancel_guard,
                    screenshare_host: None,
                    screenshare_viewers: Vec::new(),
                };

                let self_id = endpoint.id().to_string();
                // Member-issued ticket (W7 P3): the ticket we hand to the UI for
                // sharing is stamped with OUR OWN live address + the room's topic,
                // not the (possibly someone else's) ticket we joined with. So every
                // member — not just the creator — hands out a working door pointing
                // at themselves, which is what lets a room outlive its creator. We
                // keep `ticket_str` untouched for joining + the A8 retain logic; we
                // only re-stamp the *display* copy. Re-stamping is a no-op for the
                // creator (same addr+topic), and a no-op if the ticket can't be
                // parsed (a malformed join, which fails anyway).
                let share_ticket = PeerSpeakTicket::restamp(&ticket_str, endpoint.addr());
                // Advertise this gathering to friends who ping us (W7 B2): our own
                // restamped member ticket → a one-click Join. The cosmetic label
                // rides in the ticket (set by the creator), so every member —
                // creator or joiner — reports the same room name. Re-sanitize the
                // parsed label since the ticket is peer-supplied (untrusted).
                let room_label =
                    crate::sanitize::sanitize_name(&PeerSpeakTicket::label_of(&share_ticket));
                *current_room.lock().unwrap() = Some(crate::presence::RoomPresence {
                    name: room_label,
                    ticket: share_ticket.clone(),
                });
                let _ = ui_tx.send(UiEvent::RoomJoined { ticket: share_ticket, self_id }).await;
                active_session = Some(session);
            }

            CoreCommand::Leave => {
                // Finalize any recording first, while the audio feeders are alive.
                stop_recording(&recorder, &is_recording, &multitrack, &is_multitrack, &ui_tx).await;
                current_sharing = None;
                if let Some(session) = active_session.take() {
                    session.shutdown(audio_backend.clone()).await;
                    // Stop routing inbound audio links — the endpoint/router stay up.
                    net.audio_router.clear();
                    // No longer in a gathering — friends who ping see us as just online.
                    *current_room.lock().unwrap() = None;
                    let _ = ui_tx.send(UiEvent::RoomLeft).await;
                }
                // Apply any network-mode / identity change that was deferred while we
                // were in the call (rebuild while idle keeps the endpoint reachable).
                if net_rebuild_pending {
                    let lookup = net.memory_lookup.clone();
                    net.shutdown().await;
                    let publish = presence_mode.lock().unwrap().publishes_to_discovery();
                    net = build_net_stack(secret_key.clone(), network_mode, lookup, friends_handler.clone(), publish).await?;
                    net_rebuild_pending = false;
                }
            }

            CoreCommand::ToggleMute => {
                let current = is_muted.load(Ordering::Relaxed);
                let new_state = !current;
                is_muted.store(new_state, Ordering::Relaxed);

                if let Some(session) = &active_session {
                    let self_state = PeerState {
                        name: current_name.clone(),
                        is_muted: new_state,
                        addr: net.endpoint.addr(),
                        sharing: current_sharing.clone(),
                        avatar: current_avatar.clone(),
                    };
                    let _ = session.room_state.update_self_state(self_state).await;
                }
            }

            CoreCommand::SetAvatar(avatar) => {
                current_avatar = avatar;
                // Re-announce presence so the room (incl. late joiners, via the
                // retained presence) picks up the new avatar (W4).
                if let Some(session) = &active_session {
                    let self_state = PeerState {
                        name: current_name.clone(),
                        is_muted: is_muted.load(Ordering::Relaxed),
                        addr: net.endpoint.addr(),
                        sharing: current_sharing.clone(),
                        avatar: current_avatar.clone(),
                    };
                    let _ = session.room_state.update_self_state(self_state).await;
                }
            }

            CoreCommand::ToggleDeafen => {
                let current = is_deafened.load(Ordering::Relaxed);
                is_deafened.store(!current, Ordering::Relaxed);
            }

            CoreCommand::SetPttMode(enabled) => {
                ptt_mode.store(enabled, Ordering::Relaxed);
            }

            CoreCommand::SetPttActive(active) => {
                ptt_active.store(active, Ordering::Relaxed);
            }

            CoreCommand::SetPeerVolume(peer_id, vol) => {
                let mut guard = peer_volumes.lock().await;
                guard.insert(peer_id, vol);
            }

            CoreCommand::SetPeerEq(peer_id, settings) => {
                let settings = settings.clamped();
                let mut guard = peer_eq.lock().await;
                if settings.is_flat() {
                    guard.remove(&peer_id);
                } else {
                    guard.insert(peer_id, settings);
                }
            }

            CoreCommand::SetPeerPan(peer_id, pan) => {
                let pan = pan.clamp(-1.0, 1.0);
                let mut guard = peer_pan.lock().await;
                if pan.abs() <= 0.001 {
                    guard.remove(&peer_id);
                } else {
                    guard.insert(peer_id, pan);
                }
            }

            CoreCommand::SetPeerMuted(peer_id, muted) => {
                let mut guard = locally_muted.lock().await;
                if muted {
                    guard.insert(peer_id);
                } else {
                    guard.remove(&peer_id);
                }
            }

            CoreCommand::SetNoiseGateThreshold(threshold) => {
                noise_gate_threshold.store(threshold.to_bits(), Ordering::Relaxed);
            }

            CoreCommand::SetInputVolume(vol) => {
                input_gain.store(vol.to_bits(), Ordering::Relaxed);
            }

            CoreCommand::SetOutputVolume(vol) => {
                output_gain.store(vol.to_bits(), Ordering::Relaxed);
            }

            CoreCommand::SetMicMonitor { enabled, input_device } => {
                // During a call the in-call capture thread already reports the
                // mic level, and it owns the backend's capture stream — leave it be.
                if active_session.is_some() {
                    continue;
                }
                if enabled {
                    if mic_monitor.is_none() {
                        let (tx, rx) = std::sync::mpsc::channel();
                        match audio_backend.start_capture(tx, input_device) {
                            Ok(()) => {
                                let ui = ui_tx.clone();
                                let gain = input_gain.clone();
                                let thread = std::thread::spawn(move || run_mic_monitor(rx, ui, gain));
                                mic_monitor = Some(MicMonitor { thread });
                            }
                            Err(e) => {
                                let _ = ui_tx
                                    .send(UiEvent::Error(format!("Mic test unavailable: {e}")))
                                    .await;
                            }
                        }
                    }
                } else {
                    stop_mic_monitor(&audio_backend, mic_monitor.take());
                }
            }

            CoreCommand::SetNetworkMode(mode) => {
                network_mode = mode;
                // Rebuild the persistent stack to the new posture immediately if
                // idle; if a call is active, defer to the next Leave/Join so the
                // live call isn't disrupted (preserves "applies on next join").
                if active_session.is_none() {
                    let lookup = net.memory_lookup.clone();
                    net.shutdown().await;
                    let publish = presence_mode.lock().unwrap().publishes_to_discovery();
                    net = build_net_stack(secret_key.clone(), network_mode, lookup, friends_handler.clone(), publish).await?;
                } else {
                    net_rebuild_pending = true;
                }
            }

            CoreCommand::RegenerateIdentity => {
                // Mint + persist a fresh identity, discarding the old one. The
                // persistent endpoint is rebuilt with the new key (now if idle, else
                // on the next Leave/Join) — consistent with "applies on next join."
                let mut regenerated = false;
                match crate::identity::regenerate() {
                    Ok(key) => {
                        secret_key = key;
                        identity_error = None;
                        regenerated = true;
                        crate::log_msg("identity: regenerated to a fresh persistent id");
                    }
                    Err(e) => {
                        // Couldn't write the new key — keep the current one in
                        // memory but report the disk problem as a degraded state.
                        identity_error = Some(format!("{e:#}"));
                        crate::log_msg(&format!("identity: regenerate failed: {e:#}"));
                    }
                }
                // Re-bind the endpoint under the new id so our node id actually
                // changes. Only on a successful regenerate (a failed one left the
                // key unchanged, so a rebuild would be pointless churn).
                if regenerated {
                    if active_session.is_none() {
                        let lookup = net.memory_lookup.clone();
                        net.shutdown().await;
                        let publish = presence_mode.lock().unwrap().publishes_to_discovery();
                    net = build_net_stack(secret_key.clone(), network_mode, lookup, friends_handler.clone(), publish).await?;
                    } else {
                        net_rebuild_pending = true;
                    }
                }
                let _ = ui_tx
                    .send(UiEvent::IdentityStatus {
                        node_id: secret_key.public().to_string(),
                        persisted: identity_error.is_none(),
                        error: identity_error.clone(),
                    })
                    .await;
            }

            CoreCommand::AddFriend { id, name, addr } => {
                // Idempotent: re-adding an existing id is a no-op (preserves the
                // local name/addr), so only persist+emit when something changed.
                let changed = friends.lock().unwrap().add(id, name, addr);
                if changed {
                    persist_and_emit_friends(&friends, friends_read_only, &ui_tx).await;
                }
            }

            CoreCommand::RemoveFriend(id) => {
                if friends.lock().unwrap().remove(&id) {
                    persist_and_emit_friends(&friends, friends_read_only, &ui_tx).await;
                }
            }

            CoreCommand::RenameFriend(id, new_name) => {
                if friends.lock().unwrap().rename(&id, new_name) {
                    persist_and_emit_friends(&friends, friends_read_only, &ui_tx).await;
                }
            }

            CoreCommand::SetPresenceMode(mode) => {
                let previous_mode = *presence_mode.lock().unwrap();
                let now = tokio::time::Instant::now();

                if previous_mode == mode {
                    // Same-mode requests are no-ops for discovery wiring, but keep the
                    // existing UX: re-selecting Discoverable restarts the clock.
                    discovery_deadline = if mode == PresenceMode::Discoverable {
                        Some(now + crate::discovery::DISCOVERY_TIMEBOX)
                    } else {
                        None
                    };
                    continue;
                }

                // W7 P6/S11: re-apply n0 DNS discovery for the requested posture
                // first, then commit the presence mode only if the endpoint accepted
                // that discovery plan. This keeps the UI truthful when dropping the
                // publisher fails.
                let plan =
                    crate::discovery::lookup_plan(network_mode, mode.publishes_to_discovery());
                let apply_result = apply_discovery(&net.endpoint, &net.memory_lookup, plan);
                let (committed_mode, transition_error) =
                    crate::discovery::resolve_presence_transition(
                        previous_mode,
                        mode,
                        apply_result.is_ok(),
                    );
                *presence_mode.lock().unwrap() = committed_mode;

                if committed_mode == PresenceMode::Discoverable {
                    if apply_result.is_ok() && mode == PresenceMode::Discoverable {
                        discovery_deadline = Some(now + crate::discovery::DISCOVERY_TIMEBOX);
                    } else {
                        arm_discovery_retry(&mut discovery_deadline, now);
                    }
                } else {
                    discovery_deadline = None;
                }

                if let Err(e) = apply_result {
                    crate::log_msg(&format!("discovery: apply failed: {e:#}"));
                    if committed_mode != mode {
                        let _ = ui_tx
                            .send(UiEvent::PresenceModeReverted {
                                mode: committed_mode,
                            })
                            .await;
                    }
                    if let Some(message) = transition_error {
                        let _ = ui_tx
                            .send(UiEvent::Error(format!("{message} ({e:#})")))
                            .await;
                    }
                }
            }

            CoreCommand::SetRecordingMode(mode) => {
                recording_mode = mode;
            }

            CoreCommand::SetRecording(enabled) => {
                if enabled {
                    // Only record while in a call, and not already recording.
                    if active_session.is_none() {
                        let _ = ui_tx
                            .send(UiEvent::Error("Join a call before recording".into()))
                            .await;
                    } else if !is_recording.load(Ordering::Relaxed) {
                        match dirs::home_dir() {
                            Some(home) => {
                                let base = home.join("peerspeak-recordings");
                                let now = std::time::SystemTime::now()
                                    .duration_since(std::time::UNIX_EPOCH)
                                    .map(|d| d.as_secs())
                                    .unwrap_or(0);
                                let result: Result<String, String> = if recording_mode.is_multitrack() {
                                    // Multitrack/Both: a per-session directory of stems.
                                    let stamp = crate::audio::recorder::timestamp_filename(now);
                                    let session_dir = base.join(stamp.trim_end_matches(".wav"));
                                    std::fs::create_dir_all(&session_dir)
                                        .map_err(|e| e.to_string())
                                        .and_then(|_| {
                                            MultitrackRecorder::create(
                                                &session_dir,
                                                FRAME_SAMPLES,
                                                matches!(recording_mode, RecordingMode::Both),
                                            )
                                            .map_err(|e| e.to_string())
                                        })
                                        .map(|mut mt| {
                                            // Register everyone already in the room so their
                                            // stems are silence-aligned from t=0.
                                            if let Some(session) = &active_session {
                                                for (id, st) in session.room_state.active_peers() {
                                                    if let Err(e) = mt.add_peer(id, &st.name) {
                                                        crate::log_msg(&format!("multitrack add_peer failed: {e}"));
                                                    }
                                                }
                                            }
                                            let path = mt.dir().to_string_lossy().to_string();
                                            *multitrack.lock().unwrap() = Some(mt);
                                            is_multitrack.store(true, Ordering::Relaxed);
                                            path
                                        })
                                } else {
                                    // Mixed: one file (original behaviour).
                                    std::fs::create_dir_all(&base)
                                        .map_err(|e| e.to_string())
                                        .and_then(|_| {
                                            crate::audio::recorder::Recorder::create(&base, now)
                                                .map_err(|e| e.to_string())
                                        })
                                        .map(|rec| {
                                            let path = rec.path().to_string_lossy().to_string();
                                            *recorder.lock().unwrap() = Some(rec);
                                            path
                                        })
                                };
                                match result {
                                    Ok(path) => {
                                        is_recording.store(true, Ordering::Relaxed);
                                        crate::log_msg(&format!("Recording started ({recording_mode:?}): {path}"));
                                        let _ = ui_tx
                                            .send(UiEvent::RecordingStarted { path })
                                            .await;
                                    }
                                    Err(e) => {
                                        let _ = ui_tx
                                            .send(UiEvent::Error(format!("Recording failed: {e}")))
                                            .await;
                                    }
                                }
                            }
                            None => {
                                let _ = ui_tx
                                    .send(UiEvent::Error("No home directory for recordings".into()))
                                    .await;
                            }
                        }
                    }
                } else {
                    stop_recording(&recorder, &is_recording, &multitrack, &is_multitrack, &ui_tx).await;
                }
            }

            CoreCommand::SendChat(text) => {
                if let Some(session) = &active_session
                    && let Err(e) = session.room_state.send_chat(text).await
                {
                    crate::log_msg(&format!("Failed to send chat: {e}"));
                }
            }

            CoreCommand::SetPixelpassPath(path) => {
                pixelpass_override = path.filter(|p| !p.trim().is_empty());
            }

            CoreCommand::StartScreenShare => {
                let Some(session) = &mut active_session else {
                    let _ = ui_tx
                        .send(UiEvent::Error("Join a call before sharing your screen".into()))
                        .await;
                    continue;
                };
                if session.screenshare_host.is_some() {
                    continue; // already sharing
                }
                let bin = match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) {
                    Some(b) => b,
                    None => {
                        let _ = ui_tx
                            .send(UiEvent::Error(
                                "pixelpass not found — install it to share your screen".into(),
                            ))
                            .await;
                        continue;
                    }
                };
                match crate::screenshare::spawn_host(&bin).await {
                    Ok((child, ticket)) => {
                        crate::log_msg("Screen share host started");
                        session.screenshare_host = Some(child);
                        current_sharing = Some(ticket.clone());
                        let self_state = PeerState {
                            name: current_name.clone(),
                            is_muted: is_muted.load(Ordering::Relaxed),
                            addr: net.endpoint.addr(),
                            sharing: Some(ticket),
                            avatar: current_avatar.clone(),
                        };
                        let _ = session.room_state.update_self_state(self_state).await;
                        let _ = ui_tx.send(UiEvent::ScreenShareStarted).await;
                    }
                    Err(e) => {
                        let _ = ui_tx
                            .send(UiEvent::Error(format!("Couldn't start screen share: {e}")))
                            .await;
                    }
                }
            }

            CoreCommand::StopScreenShare => {
                current_sharing = None;
                if let Some(session) = &mut active_session {
                    if let Some(mut child) = session.screenshare_host.take() {
                        let _ = child.kill().await;
                        crate::log_msg("Screen share host stopped");
                    }
                    let self_state = PeerState {
                        name: current_name.clone(),
                        is_muted: is_muted.load(Ordering::Relaxed),
                        addr: net.endpoint.addr(),
                        sharing: None,
                        avatar: current_avatar.clone(),
                    };
                    let _ = session.room_state.update_self_state(self_state).await;
                }
                let _ = ui_tx.send(UiEvent::ScreenShareStopped).await;
            }

            CoreCommand::ViewShare(ticket) => {
                let bin = match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) {
                    Some(b) => b,
                    None => {
                        let _ = ui_tx
                            .send(UiEvent::Error(
                                "pixelpass not found — install it to watch screen shares".into(),
                            ))
                            .await;
                        continue;
                    }
                };
                match crate::screenshare::spawn_viewer(&bin, &ticket).await {
                    Ok(child) => {
                        crate::log_msg("Screen share viewer started");
                        if let Some(session) = &mut active_session {
                            session.screenshare_viewers.push(child);
                        }
                    }
                    Err(e) => {
                        let _ = ui_tx
                            .send(UiEvent::Error(format!("Couldn't watch screen share: {e}")))
                            .await;
                    }
                }
            }
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{
        apply_volume, audio_datagram_len_ok, frame_level, mix_frames, mix_stereo_frames,
        stereo_to_mono, MicLevelMeter, MAX_OPUS_PAYLOAD, MIC_LEVEL_REPORT_SAMPLES,
    };

    /// A frame of constant amplitude with the given sample count.
    fn frame(amp: i16, len: usize) -> Vec<i16> {
        vec![amp; len]
    }

    #[test]
    fn mic_meter_reports_only_after_enough_samples() {
        let mut m = MicLevelMeter::new();
        // One short frame well under the report window yields nothing yet.
        assert_eq!(m.push(&frame(1000, 480)), None);
        // A frame that crosses the window boundary triggers a report.
        assert!(m.push(&frame(1000, MIC_LEVEL_REPORT_SAMPLES)).is_some());
    }

    #[test]
    fn audio_datagram_length_gate_preserves_header_and_caps_payload() {
        assert!(!audio_datagram_len_ok(0));
        assert!(!audio_datagram_len_ok(3));
        assert!(audio_datagram_len_ok(4));
        assert!(audio_datagram_len_ok(4 + MAX_OPUS_PAYLOAD));
        assert!(!audio_datagram_len_ok(5 + MAX_OPUS_PAYLOAD));
    }

    #[test]
    fn mic_meter_holds_the_peak_across_the_window() {
        let mut m = MicLevelMeter::new();
        let chunk = MIC_LEVEL_REPORT_SAMPLES / 4;
        // Loud frame first, then quiet ones — the reported value is the loud peak.
        assert_eq!(m.push(&frame(8000, chunk)), None);
        assert_eq!(m.push(&frame(100, chunk)), None);
        assert_eq!(m.push(&frame(100, chunk)), None);
        let reported = m.push(&frame(100, chunk)).expect("window complete");
        let loud = frame_level(&frame(8000, chunk));
        assert!((reported - loud).abs() < 1e-6, "peak should hold the loud frame");
    }

    #[test]
    fn mic_meter_resets_after_reporting() {
        let mut m = MicLevelMeter::new();
        // Fill and report a loud window.
        assert!(m.push(&frame(8000, MIC_LEVEL_REPORT_SAMPLES)).is_some());
        // The next window of silence must report ~zero, not the stale loud peak.
        let reported = m.push(&frame(0, MIC_LEVEL_REPORT_SAMPLES)).expect("second window");
        assert_eq!(reported, 0.0, "peak and accumulator reset between windows");
    }

    #[test]
    fn mic_meter_silence_reports_zero() {
        let mut m = MicLevelMeter::new();
        let reported = m.push(&frame(0, MIC_LEVEL_REPORT_SAMPLES)).expect("window complete");
        assert_eq!(reported, 0.0);
    }

    #[test]
    fn mix_of_no_peers_is_silence() {
        let mixed = mix_frames(&[], 4);
        assert_eq!(mixed, vec![0i32; 4]);
    }

    #[test]
    fn single_peer_passes_through_unchanged() {
        let frame = vec![100, -200, 300, -400];
        let mixed = mix_frames(std::slice::from_ref(&frame), 4);
        assert_eq!(mixed, vec![100i32, -200, 300, -400]);
    }

    #[test]
    fn centered_stereo_mix_is_exact_dual_mono() {
        let a = vec![100, -200, 300, -400];
        let b = vec![50, 200, -100, 400];
        let mixed = mix_stereo_frames(&[(a, 0.0), (b, 0.0)], 4);
        assert_eq!(mixed, vec![150, 150, 0, 0, 200, 200, 0, 0]);
    }

    #[test]
    fn hard_left_pan_only_contributes_left_channel() {
        let frame = vec![100, 200];
        let mixed = mix_stereo_frames(&[(frame, -1.0)], 2);
        assert_eq!(mixed, vec![141, 0, 283, 0]);
    }

    #[test]
    fn stereo_fold_down_averages_pairs() {
        let mono = stereo_to_mono(&[100, 100, 200, 0, i16::MAX, i16::MAX]);
        assert_eq!(mono, vec![100, 100, i16::MAX]);
    }

    #[test]
    fn two_peers_sum_sample_by_sample() {
        let a = vec![100, -200, 300, -400];
        let b = vec![50, 200, -100, 400];
        let mixed = mix_frames(&[a, b], 4);
        assert_eq!(mixed, vec![150i32, 0, 200, 0]);
    }

    #[test]
    fn loud_positive_mix_is_lossless_not_clamped() {
        // The bus is a lossless i32 sum now — the true peak (~2x i16::MAX) is
        // preserved so the limiter can ride it down. (The old mixer clamped here.)
        let a = vec![30_000; 4];
        let b = vec![30_000; 4];
        let mixed = mix_frames(&[a, b], 4);
        assert_eq!(mixed, vec![60_000i32; 4]);
    }

    #[test]
    fn loud_negative_mix_is_lossless_not_clamped() {
        let a = vec![i16::MIN; 4];
        let b = vec![i16::MIN; 4];
        let mixed = mix_frames(&[a, b], 4);
        assert_eq!(mixed, vec![2 * i16::MIN as i32; 4]);
    }

    #[test]
    fn shorter_peer_frame_contributes_zero_past_its_end() {
        let full = vec![100, 100, 100, 100];
        let short = vec![10, 20]; // only first two samples
        let mixed = mix_frames(&[full, short], 4);
        assert_eq!(mixed, vec![110i32, 120, 100, 100]);
    }

    #[test]
    fn volume_unity_is_a_noop() {
        let mut frame = vec![100, -200, 300, -400];
        apply_volume(&mut frame, 1.0);
        assert_eq!(frame, vec![100, -200, 300, -400]);
    }

    #[test]
    fn volume_zero_mutes() {
        let mut frame = vec![100, -200, 300, -400];
        apply_volume(&mut frame, 0.0);
        assert_eq!(frame, vec![0, 0, 0, 0]);
    }

    #[test]
    fn volume_half_scales_samples() {
        let mut frame = vec![100, -200, 300, -400];
        apply_volume(&mut frame, 0.5);
        // 100*0.5=50, -200*0.5=-100, 300*0.5=150, -400*0.5=-200 (exact in f32 here)
        assert_eq!(frame, vec![50, -100, 150, -200]);
    }

    #[test]
    fn volume_boost_saturates_not_wraps() {
        // 20000 * 4.0 = 80000, well past i16::MAX — must clamp, not wrap.
        let mut frame = vec![20_000, -20_000, 20_000, -20_000];
        apply_volume(&mut frame, 4.0);
        assert_eq!(frame, vec![i16::MAX, i16::MIN, i16::MAX, i16::MIN]);
    }

    #[test]
    fn frame_level_of_silence_is_zero() {
        assert_eq!(frame_level(&[0, 0, 0, 0]), 0.0);
        assert_eq!(frame_level(&[]), 0.0);
    }

    #[test]
    fn frame_level_of_full_scale_is_about_one() {
        let full = vec![i16::MAX; 64];
        let level = frame_level(&full);
        assert!(level > 0.99 && level <= 1.0, "full-scale level was {level}");
    }

    #[test]
    fn volume_truncates_toward_zero() {
        let mut frame = vec![3, -3, 5, -5];
        apply_volume(&mut frame, 0.5);
        assert_eq!(frame, vec![1, -1, 2, -2]);
    }

    #[test]
    fn volume_amplifies_without_saturating() {
        let mut frame = vec![1000, -1000];
        apply_volume(&mut frame, 2.0);
        assert_eq!(frame, vec![2000, -2000]);
    }

    #[test]
    fn three_peers_sum_without_saturation() {
        let a = vec![10, 20];
        let b = vec![3, 4];
        let c = vec![100, -50];
        let mixed = mix_frames(&[a, b, c], 2);
        assert_eq!(mixed, vec![113i32, -26]);
    }

    #[test]
    fn mix_zero_pads_output_longer_than_peer_frames() {
        let a = vec![100, 200];
        let mixed = mix_frames(&[a], 4);
        assert_eq!(mixed, vec![100i32, 200, 0, 0]);
    }

    #[test]
    fn frame_level_mid_range() {
        let frame = vec![16384; 64];
        let level = frame_level(&frame);
        assert!((level - 0.5).abs() < 1e-3, "mid-range level was {level}");
    }
}