chore: 初始化 client v2

packages/client-v2/src/app/entry.rs

@@ -0,0 +1,35 @@
+#![cfg(target_os = "linux")]
+
+use crate::app::master::run_master;
+use crate::app::slave::run_slave;
+use crate::net::protocol::ChannelRole;
+use anyhow::Result;
+use std::env;
+
+pub async fn run_xiao() -> Result<()> {
+    let args: Vec<String> = env::args().collect();
+    if args.len() < 3 {
+        eprintln!("用法: {} [master|slave] [left|right]", args[0]);
+        return Ok(());
+    }
+
+    let mode = if args[1].to_lowercase() == "master" {
+        "主节点"
+    } else {
+        "从节点"
+    };
+
+    let role = if args[2].to_lowercase() == "left" {
+        ChannelRole::Left
+    } else {
+        ChannelRole::Right
+    };
+
+    println!("🚗 当前为: {} {}", mode, role.to_string());
+
+    if mode == "主节点" {
+        run_master(role).await
+    } else {
+        run_slave(role).await
+    }
+}

packages/client-v2/src/app/master.rs

@@ -0,0 +1,365 @@
+#![cfg(target_os = "linux")]
+
+use crate::audio::codec::OpusCodec;
+use crate::audio::config::AudioConfig;
+use crate::audio::player::AudioPlayer;
+use crate::net::discovery::Discovery;
+use crate::net::network::{ControlConnection, MasterNetwork};
+use crate::net::protocol::{AudioPacket, ChannelRole, ControlPacket};
+use crate::utils::alsa::AlsaRedirector;
+use crate::utils::sync::now_us;
+use anyhow::{Result, anyhow};
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::signal::unix::{SignalKind, signal};
+use tokio::sync::Mutex;
+
+pub const SERVER_TCP_PORT: u16 = 53531;
+
+#[derive(Clone)]
+struct SlaveSession {
+    udp_addr: SocketAddr,
+    role: ChannelRole,
+}
+
+pub async fn run_master(master_role: ChannelRole) -> Result<()> {
+    // 0. 设置 ALSA 重定向
+    println!("🔥 启动中，请稍等...");
+    let _alsa_guard = AlsaRedirector::new()?;
+
+    // 1. 设置网络 (UDP + TCP)
+    let network = MasterNetwork::setup(SERVER_TCP_PORT).await?;
+    let audio_socket = network.audio_socket().clone_inner();
+
+    // 2. 启动服务发现广播
+    Discovery::start_broadcast(SERVER_TCP_PORT).await?;
+
+    println!("✅ 服务已启动，等待连接...");
+
+    let shutdown_flag = Arc::new(AtomicBool::new(false));
+    let slaves = Arc::new(Mutex::new(Vec::<SlaveSession>::new()));
+
+    // 3. 启动连接监听任务
+    let slaves_clone = slaves.clone();
+    let audio_socket_clone = audio_socket.clone();
+    tokio::spawn(async move {
+        loop {
+            match network.accept().await {
+                Ok((control_conn, client_addr)) => {
+                    let slaves_for_session = slaves_clone.clone();
+                    let audio_socket_for_session = audio_socket_clone.clone();
+                    tokio::spawn(async move {
+                        if let Err(e) = handle_master_session(
+                            control_conn,
+                            audio_socket_for_session,
+                            slaves_for_session,
+                            client_addr.to_string(),
+                        )
+                        .await
+                        {
+                            eprintln!("❌ 会话错误: {:?}", e);
+                        }
+                    });
+                }
+                Err(e) => {
+                    eprintln!("❌ Accept 错误: {:?}", e);
+                }
+            }
+        }
+    });
+
+    // 4. 音频处理主循环
+    let config = AudioConfig::music();
+    let encode_config = AudioConfig {
+        channels: 1,
+        vbr: true,
+        ..AudioConfig::music()
+    };
+    let player = AudioPlayer::new(&AudioConfig {
+        channels: 2,
+        playback_device: "plug:original_default".into(),
+        ..config.clone()
+    })?;
+
+    let mut raw_buf = vec![0u8; config.frame_size * 2 * 2];
+    let mut pcm_out = vec![0i16; config.frame_size * 2];
+    let mut left_pcm = vec![0i16; config.frame_size];
+    let mut right_pcm = vec![0i16; config.frame_size];
+    let mut opus_out = vec![0u8; 1500];
+    let mut seq = 0u32;
+
+    // 播放延迟: 只需覆盖网络延迟 + 时钟偏移
+    let delay_us = 100_000; // 100ms 基础延迟
+    let frame_duration_us =
+        (config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128;
+
+    let mut stream_start_ts = 0;
+    let mut stream_start_seq = 0;
+
+    let shutdown_flag_clone = shutdown_flag.clone();
+    let audio_loop = async move {
+        loop {
+            if shutdown_flag_clone.load(Ordering::Relaxed) {
+                break;
+            }
+
+            // 打开 FIFO
+            let mut fifo = match tokio::fs::File::open(AlsaRedirector::fifo_path()).await {
+                Ok(f) => f,
+                Err(_) => {
+                    if shutdown_flag_clone.load(Ordering::Relaxed) {
+                        break;
+                    }
+                    tokio::time::sleep(Duration::from_secs(1)).await;
+                    continue;
+                }
+            };
+
+            // 每个新流开始时，重置编码器状态以避免残留音频导致爆音
+            let mut left_encoder = OpusCodec::new(&encode_config)?;
+            let mut right_encoder = OpusCodec::new(&encode_config)?;
+
+            loop {
+                if shutdown_flag_clone.load(Ordering::Relaxed) {
+                    break;
+                }
+
+                // 从 FIFO 读取
+                if let Err(_) = fifo.read_exact(&mut raw_buf).await {
+                    break; // FIFO 关闭，重新打开
+                }
+
+                let active_slaves = {
+                    let s = slaves.lock().await;
+                    if s.is_empty() { None } else { Some(s.clone()) }
+                };
+
+                let now = now_us();
+                if stream_start_ts == 0 {
+                    stream_start_ts = now;
+                    stream_start_seq = seq;
+                }
+
+                // 计算该帧应当播放的目标时间
+                // target_ts = 数据包发送时间 + 播放延迟
+                let target_ts = stream_start_ts
+                    + ((seq - stream_start_seq) as u128 * frame_duration_us)
+                    + delay_us;
+
+                // 提取 PCM 数据
+                for i in 0..config.frame_size {
+                    left_pcm[i] = i16::from_le_bytes([raw_buf[i * 4], raw_buf[i * 4 + 1]]);
+                    right_pcm[i] = i16::from_le_bytes([raw_buf[i * 4 + 2], raw_buf[i * 4 + 3]]);
+                }
+
+                if let Some(slaves_list) = active_slaves {
+                    // 情况 1: 有从节点，进行网络传输，并本地构造静音声道回放
+
+                    // 1. 检查各声道是否有从节点需要
+                    let needs_left = slaves_list.iter().any(|s| s.role == ChannelRole::Left);
+                    let needs_right = slaves_list.iter().any(|s| s.role == ChannelRole::Right);
+
+                    // 2. 编码需要的声道
+                    let mut left_bytes = None;
+                    let mut right_bytes = None;
+
+                    if needs_left {
+                        let len = left_encoder.encode(&left_pcm, &mut opus_out)?;
+                        let packet = AudioPacket {
+                            seq,
+                            timestamp: target_ts,
+                            data: opus_out[..len].to_vec(),
+                        };
+                        left_bytes = Some(postcard::to_allocvec(&packet)?);
+                    }
+
+                    if needs_right {
+                        let len = right_encoder.encode(&right_pcm, &mut opus_out)?;
+                        let packet = AudioPacket {
+                            seq,
+                            timestamp: target_ts,
+                            data: opus_out[..len].to_vec(),
+                        };
+                        right_bytes = Some(postcard::to_allocvec(&packet)?);
+                    }
+
+                    // 3. 发送给对应的从节点
+                    for slave in &slaves_list {
+                        let bytes = match slave.role {
+                            ChannelRole::Left => left_bytes.as_ref(),
+                            ChannelRole::Right => right_bytes.as_ref(),
+                        };
+                        if let Some(b) = bytes {
+                            let _ = audio_socket.send_to(b, slave.udp_addr).await;
+                        }
+                    }
+
+                    // 4. 将非本节点的声道置为静音
+                    for i in 0..config.frame_size {
+                        match master_role {
+                            ChannelRole::Left => {
+                                pcm_out[i * 2] = left_pcm[i];
+                                pcm_out[i * 2 + 1] = 0;
+                            }
+                            ChannelRole::Right => {
+                                pcm_out[i * 2] = 0;
+                                pcm_out[i * 2 + 1] = right_pcm[i];
+                            }
+                        }
+                    }
+
+                    // 5. 等待播放
+                    let now = now_us();
+                    if now < target_ts {
+                        let wait = target_ts - now;
+                        if wait > 1000 {
+                            tokio::time::sleep(Duration::from_micros(wait as u64)).await;
+                        } else {
+                            // 小于 1ms，直接播放，让播放时机稍微早一点点
+                        }
+                    }
+                } else {
+                    // 情况 2: 没有从节点，本地立体声播放
+                    for i in 0..config.frame_size {
+                        pcm_out[i * 2] = left_pcm[i];
+                        pcm_out[i * 2 + 1] = right_pcm[i];
+                    }
+                }
+
+                // 统一写入播放器 (始终是立体声)
+                if let Err(_) = player.write(&pcm_out) {
+                    if shutdown_flag_clone.load(Ordering::Relaxed) {
+                        break;
+                    }
+                }
+
+                seq += 1;
+            }
+
+            // 重置流计时
+            stream_start_ts = 0;
+        }
+
+        Ok::<(), anyhow::Error>(())
+    };
+
+    tokio::select! {
+        res = audio_loop => {
+            if let Err(e) = res {
+                eprintln!("❌ 音频循环错误: {:?}", e);
+            }
+        },
+        _ = shutdown_signal() => {
+            // 设置退出标志，通知音频循环停止
+            shutdown_flag.store(true, Ordering::Relaxed);
+        },
+    }
+
+    // 显式清理
+    println!("👋 正在退出...");
+    AlsaRedirector::cleanup();
+
+    // 强制退出
+    std::process::exit(0);
+}
+
+/// 监听系统退出信号 (SIGINT, SIGTERM, SIGQUIT)
+async fn shutdown_signal() {
+    let mut sigint = signal(SignalKind::interrupt()).expect("无法注册 SIGINT 处理器");
+    let mut sigterm = signal(SignalKind::terminate()).expect("无法注册 SIGTERM 处理器");
+    let mut sigquit = signal(SignalKind::quit()).expect("无法注册 SIGQUIT 处理器");
+
+    tokio::select! {
+        _ = sigint.recv() => {},
+        _ = sigterm.recv() => {},
+        _ = sigquit.recv() => {},
+    }
+}
+
+/// 处理主节点与从节点的会话
+async fn handle_master_session(
+    mut control: ControlConnection,
+    audio_socket: Arc<tokio::net::UdpSocket>,
+    slaves: Arc<Mutex<Vec<SlaveSession>>>,
+    client_tcp_addr: String,
+) -> Result<()> {
+    let mut buf = [0u8; 1024];
+
+    // 握手
+    let pkt = control.recv_packet(&mut buf).await?;
+    let slave_role = match pkt {
+        ControlPacket::ClientIdentify { role } => role,
+        _ => return Err(anyhow!("无效的握手协议")),
+    };
+
+    let xiao = ControlPacket::ServerHello {
+        udp_port: audio_socket.local_addr()?.port(),
+    };
+    control.send_packet(&xiao).await?;
+
+    // 等待 UDP 打洞/确认
+    let mut buf = [0u8; 128];
+    let (_, client_udp_addr) = audio_socket.recv_from(&mut buf).await?;
+
+    println!(
+        "✅ 从节点已连接: {} {}",
+        client_tcp_addr,
+        slave_role.to_string(),
+    );
+
+    // 添加到从节点列表
+    let session = SlaveSession {
+        udp_addr: client_udp_addr,
+        role: slave_role,
+    };
+    {
+        let mut s = slaves.lock().await;
+        s.push(session.clone());
+    }
+
+    // 分离 TCP 读写，处理控制消息和心跳
+    let (mut tcp_rx, mut tcp_tx) = control.split();
+
+    let mut buf = [0u8; 1024];
+    loop {
+        match tcp_rx.read(&mut buf).await {
+            Ok(0) | Err(_) => {
+                break;
+            }
+            Ok(n) => {
+                if let Ok(ControlPacket::Ping { client_ts, seq }) = postcard::from_bytes(&buf[..n])
+                {
+                    let pong = ControlPacket::Pong {
+                        client_ts,
+                        server_ts: now_us(),
+                        seq,
+                    };
+                    if tcp_tx
+                        .write_all(&postcard::to_allocvec(&pong).unwrap())
+                        .await
+                        .is_err()
+                    {
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    println!(
+        "❌ 从节点已断开: {} {}",
+        client_tcp_addr,
+        slave_role.to_string(),
+    );
+
+    // 从列表中移除
+    {
+        let mut s = slaves.lock().await;
+        s.retain(|x| x.udp_addr != client_udp_addr);
+    }
+
+    Ok(())
+}

packages/client-v2/src/app/mod.rs

@@ -0,0 +1,3 @@
+pub mod entry;
+pub mod master;
+pub mod slave;

packages/client-v2/src/app/slave.rs

@@ -0,0 +1,199 @@
+#![cfg(target_os = "linux")]
+
+use crate::audio::codec::OpusCodec;
+use crate::audio::config::AudioConfig;
+use crate::audio::player::AudioPlayer;
+use crate::net::discovery::Discovery;
+use crate::net::network::SlaveNetwork;
+use crate::net::protocol::{AudioPacket, ChannelRole, ControlPacket};
+use crate::utils::sync::{ClockSync, now_us};
+use anyhow::{Result, anyhow};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::sync::{Mutex, mpsc};
+
+/// 运行从节点模式
+pub async fn run_slave(role: ChannelRole) -> Result<()> {
+    loop {
+        match handle_connection(role.clone()).await {
+            Err(e) => {
+                eprintln!("❌ {:?}", e);
+                tokio::time::sleep(Duration::from_secs(3)).await;
+            }
+            Ok(_) => {}
+        }
+    }
+}
+
+async fn handle_connection(role: ChannelRole) -> Result<()> {
+    // 1. 发现主节点
+    println!("🔍 正在扫描主节点...");
+    let (master_ip, master_tcp_port) = Discovery::discover_master().await?;
+    let master_tcp_addr = format!("{}:{}", master_ip, master_tcp_port);
+
+    // 2. 建立 TCP 连接
+    println!("🔥 发现主节点: {}", master_tcp_addr);
+    let network = SlaveNetwork::connect(master_tcp_addr.parse()?).await?;
+    let (mut control, audio) = network.split();
+
+    // 3. 身份认证
+    control
+        .send_packet(&ControlPacket::ClientIdentify { role: role.clone() })
+        .await?;
+
+    let mut buf = [0u8; 1024];
+    let pkt = control.recv_packet(&mut buf).await?;
+    let server_udp_port = match pkt {
+        ControlPacket::ServerHello { udp_port } => udp_port,
+        _ => return Err(anyhow!("身份认证应答异常")),
+    };
+
+    // 4. UDP 打洞
+    audio
+        .punch(format!("{}:{}", master_ip, server_udp_port).parse()?)
+        .await?;
+
+    // 5. 初始化音频与同步组件
+    let config = AudioConfig {
+        channels: 1,
+        ..AudioConfig::music()
+    };
+    let player = AudioPlayer::new(&config)?;
+    let mut codec = OpusCodec::new(&config)?;
+    let clock = Arc::new(Mutex::new(ClockSync::new(100)));
+
+    // 用于通知主循环 TCP 已断开的消息通道
+    let (disconnect_tx, mut disconnect_rx) = mpsc::channel::<()>(1);
+
+    // 6. 分离 TCP 读写
+    let (mut tcp_rx, mut tcp_tx) = control.split();
+    let clock_updater = clock.clone();
+    let d_tx_ping = disconnect_tx.clone();
+    let d_tx_pong = disconnect_tx.clone();
+
+    // 定时发送 Ping (心跳 & 时间同步)
+    let _sync_handle = tokio::spawn(async move {
+        let mut seq = 0;
+        loop {
+            let t1 = now_us();
+            let msg = ControlPacket::Ping { client_ts: t1, seq };
+            let data = postcard::to_allocvec(&msg).unwrap();
+            if tcp_tx.write_all(&data).await.is_err() {
+                let _ = d_tx_ping.send(()).await; // 通知主线程 TCP 失败
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(200)).await;
+            seq += 1;
+        }
+    });
+
+    // 接收 Pong
+    tokio::spawn(async move {
+        let mut buf = [0u8; 1024];
+        loop {
+            match tcp_rx.read(&mut buf).await {
+                Ok(n) if n > 0 => {
+                    if let Ok(ControlPacket::Pong {
+                        client_ts,
+                        server_ts,
+                        ..
+                    }) = postcard::from_bytes(&buf[..n])
+                    {
+                        let t4 = now_us();
+                        clock_updater.lock().await.update(client_ts, server_ts, t4);
+                    }
+                }
+                _ => {
+                    let _ = d_tx_pong.send(()).await; // TCP 断开
+                    break;
+                }
+            }
+        }
+    });
+
+    // 7. 接收音频数据包 (UDP)
+    let (audio_tx, mut audio_rx) = mpsc::channel(100);
+    let audio_socket = audio.clone_inner();
+    tokio::spawn(async move {
+        let mut buf = [0u8; 2048];
+        loop {
+            if let Ok((len, _)) = audio_socket.recv_from(&mut buf).await {
+                if let Ok(packet) = postcard::from_bytes::<AudioPacket>(&buf[..len]) {
+                    if audio_tx.send(packet).await.is_err() {
+                        break;
+                    }
+                }
+            }
+        }
+    });
+
+    // 8. 播放提示
+    println!("✅ 主节点已连接，音频串流中...");
+    let role_str = role.to_string();
+    tokio::spawn(async move {
+        let _ = tokio::process::Command::new("sh")
+            .arg("-c")
+            .arg(format!(
+                "/usr/sbin/tts_play.sh \"主节点已连接，{}\" >/dev/null 2>&1",
+                role_str
+            ))
+            .status()
+            .await;
+    });
+
+    // 9. 播放主循环
+    let mut pcm_buf = vec![0i16; config.frame_size];
+    let mut last_seq: Option<u32> = None;
+
+    loop {
+        // 检查 TCP 是否已断开
+        if let Ok(_) = disconnect_rx.try_recv() {
+            return Err(anyhow!("主节点已断开: {}", master_tcp_addr));
+        }
+
+        // 接收数据包
+        if let Ok(pkt) = audio_rx.try_recv() {
+            let now = now_us();
+            let current_server_time = clock.lock().await.to_server_time(now);
+
+            // 检查包是否迟到（目标时间已过）
+            if current_server_time > pkt.timestamp {
+                let late_ms = (current_server_time - pkt.timestamp) / 1000;
+                if late_ms > 50 {
+                    // 迟到超过 50ms，直接丢弃
+                    continue;
+                }
+                // 轻微迟到（<50ms），尝试播放
+            }
+
+            last_seq = Some(pkt.seq);
+
+            // 精确等待到播放时间
+            loop {
+                let now = now_us();
+                let current_server_time_precise = clock.lock().await.to_server_time(now);
+
+                if current_server_time_precise >= pkt.timestamp {
+                    break;
+                }
+
+                let wait_us = (pkt.timestamp - current_server_time_precise) as u64;
+
+                if wait_us > 1000 {
+                    tokio::time::sleep(Duration::from_micros(wait_us as u64)).await;
+                } else {
+                    // 小于 1ms，直接播放，让播放时机稍微早一点点
+                    break;
+                }
+            }
+
+            // 解码并播放
+            if let Ok(len) = codec.decode(&pkt.data, &mut pcm_buf) {
+                let _ = player.write(&pcm_buf[..len]);
+            }
+        } else {
+            tokio::time::sleep(Duration::from_micros(100)).await;
+        }
+    }
+}

packages/client-v2/src/audio/codec.rs

@@ -0,0 +1,70 @@
+use crate::audio::config::{AudioConfig, AudioScene};
+use anyhow::{Context, Result};
+use opus::{Application, Bitrate, Channels, Decoder, Encoder};
+
+pub struct OpusCodec {
+    encoder: Encoder,
+    decoder: Decoder,
+}
+
+impl OpusCodec {
+    pub fn new(config: &AudioConfig) -> Result<Self> {
+        let channels = match config.channels {
+            1 => Channels::Mono,
+            2 => Channels::Stereo,
+            _ => return Err(anyhow::anyhow!("Invalid channels: {}", config.channels)),
+        };
+        let mode = match config.audio_scene {
+            AudioScene::Music => Application::Audio,
+            AudioScene::Voice => Application::Voip,
+        };
+        let bitrate = match config.bitrate {
+            -1 => Bitrate::Max,
+            0 => Bitrate::Auto,
+            _ => Bitrate::Bits(config.bitrate),
+        };
+
+        let mut encoder = Encoder::new(config.sample_rate, channels, mode)
+            .context("Failed to create Opus encoder")?;
+
+        encoder.set_bitrate(bitrate)?;
+        if config.vbr {
+            encoder.set_vbr(true)?;
+        }
+        if config.fec {
+            encoder.set_inband_fec(true)?; // 内联前向纠错
+            encoder.set_packet_loss_perc(20)?; // 预期丢包率20%
+        }
+
+        let decoder =
+            Decoder::new(config.sample_rate, channels).context("Failed to create Opus decoder")?;
+
+        Ok(Self { encoder, decoder })
+    }
+
+    pub fn encode(&mut self, pcm: &[i16], out: &mut [u8]) -> Result<usize> {
+        self.encoder
+            .encode(pcm, out)
+            .context("Opus encoding failed")
+    }
+
+    pub fn decode(&mut self, opus: &[u8], out: &mut [i16]) -> Result<usize> {
+        self.decoder
+            .decode(opus, out, false)
+            .context("Opus decoding failed")
+    }
+
+    /// 前向纠错(FEC)
+    pub fn decode_fec(&mut self, opus: &[u8], out: &mut [i16]) -> Result<usize> {
+        self.decoder
+            .decode(opus, out, true)
+            .context("Opus FEC decoding failed")
+    }
+
+    /// 丢包补偿(PLC)
+    pub fn decode_loss(&mut self, out: &mut [i16]) -> Result<usize> {
+        self.decoder
+            .decode(&[], out, false)
+            .context("Opus PLC (decode_loss) failed")
+    }
+}

packages/client-v2/src/audio/config.rs

@@ -0,0 +1,61 @@
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum AudioScene {
+    Music,
+    Voice,
+}
+
+#[derive(Debug, Clone)]
+pub struct AudioConfig {
+    // ALSA 设备参数，用于录音和播放
+    pub capture_device: String,
+    pub playback_device: String,
+    pub sample_rate: u32,
+    pub channels: u16,
+    pub frame_size: usize, // 帧大小，单位为采样点
+
+    // Opus 编解码参数，用于音频传输
+    pub audio_scene: AudioScene,
+    pub bitrate: i32,
+    pub vbr: bool, // 是否启用 VBR（动态比特率）
+    pub fec: bool, // 是否启用 FEC（内联前向纠错）
+}
+
+impl AudioConfig {
+    pub fn music() -> Self {
+        Self {
+            audio_scene: AudioScene::Music,
+            sample_rate: 48_000, // 48kHz
+            channels: 2,
+            frame_size: 960,  // 20ms at 48kHz
+            bitrate: 320_000, // 320 kbps
+            ..Default::default()
+        }
+    }
+
+    pub fn voice() -> Self {
+        Self {
+            audio_scene: AudioScene::Voice,
+            sample_rate: 16_000, // 16kHz
+            channels: 1,
+            frame_size: 320, // 20ms at 16kHz
+            bitrate: 32_000, // 32 kbps
+            ..Default::default()
+        }
+    }
+}
+
+impl Default for AudioConfig {
+    fn default() -> Self {
+        Self {
+            audio_scene: AudioScene::Voice,
+            capture_device: "plug:Capture".to_string(),
+            playback_device: "default".to_string(),
+            sample_rate: 16_000,
+            channels: 1,
+            frame_size: 320, // 20ms at 16kHz
+            bitrate: 32_000,
+            vbr: false,
+            fec: false,
+        }
+    }
+}

packages/client-v2/src/audio/mod.rs

@@ -0,0 +1,4 @@
+pub mod codec;
+pub mod config;
+pub mod player;
+pub mod recorder;

packages/client-v2/src/audio/player.rs

@@ -0,0 +1,71 @@
+#![cfg(target_os = "linux")]
+
+use crate::audio::config::AudioConfig;
+use alsa::Direction;
+use alsa::pcm::{Access, Format, HwParams, PCM};
+use anyhow::{Context, Result};
+
+pub struct AudioPlayer {
+    pcm: PCM,
+}
+
+impl AudioPlayer {
+    pub fn new(config: &AudioConfig) -> Result<Self> {
+        let pcm = PCM::new(&config.playback_device, Direction::Playback, false)
+            .context("Failed to open playback PCM device")?;
+
+        setup_pcm(&pcm, config.sample_rate, config.channels)?;
+        Ok(Self { pcm })
+    }
+
+    pub fn write(&self, buffer: &[i16]) -> Result<usize> {
+        let res = self.pcm.io_i16()?.writei(buffer);
+
+        match res {
+            Ok(written) => Ok(written),
+            Err(e) => {
+                // Buffer Underrun，即播放缓冲区的数据被耗尽，导致音频流中断
+                if e.errno() == 32 {
+                    // 恢复音频流状态
+                    self.pcm.prepare()?;
+                    // 重新获取 IO 对象并尝试写入数据
+                    self.pcm
+                        .io_i16()?
+                        .writei(buffer)
+                        .context("Failed to write to playback device after recovery")
+                } else {
+                    Err(e).context("Failed to write to playback device")
+                }
+            }
+        }
+    }
+
+    pub fn prepare(&self) -> Result<()> {
+        self.pcm.prepare().context("Failed to prepare PCM")
+    }
+}
+
+fn setup_pcm(pcm: &PCM, sample_rate: u32, channels: u16) -> Result<()> {
+    let hwp = HwParams::any(pcm).context("Failed to get HwParams")?;
+    hwp.set_access(Access::RWInterleaved)?;
+    hwp.set_format(Format::s16())?;
+    hwp.set_rate(sample_rate, alsa::ValueOr::Nearest)?;
+    hwp.set_channels(channels as u32)?;
+
+    // 设置较大的缓冲区以减少由于调度抖动和设备重初始化导致的断音/卡顿
+    // 使用 100ms 缓冲区，既能防止 underrun，又不会引入过大延迟
+    let buffer_size = (sample_rate as f64 * 0.1) as u32; // 100ms 缓冲
+    let period_size = buffer_size / 4; // 25ms 周期
+    hwp.set_buffer_size_near(buffer_size as alsa::pcm::Frames)?;
+    hwp.set_period_size_near(period_size as alsa::pcm::Frames, alsa::ValueOr::Nearest)?;
+
+    pcm.hw_params(&hwp).context("Failed to set HwParams")?;
+
+    let swp = pcm.sw_params_current()?;
+    // 设置 start_threshold，当缓冲区有 1 个 period 数据时就开始播放
+    // 这样可以快速启动，同时保持足够的缓冲余量
+    swp.set_start_threshold(period_size as alsa::pcm::Frames)?;
+    pcm.sw_params(&swp)?;
+    pcm.prepare()?;
+    Ok(())
+}

packages/client-v2/src/audio/recorder.rs

@@ -0,0 +1,41 @@
+#![cfg(target_os = "linux")]
+
+use crate::audio::config::AudioConfig;
+use alsa::Direction;
+use alsa::pcm::{Access, Format, HwParams, PCM};
+use anyhow::{Context, Result};
+
+pub struct AudioRecorder {
+    pcm: PCM,
+}
+
+impl AudioRecorder {
+    pub fn new(config: &AudioConfig) -> Result<Self> {
+        let pcm = PCM::new(&config.capture_device, Direction::Capture, false)
+            .context("Failed to open capture PCM device")?;
+
+        setup_pcm(&pcm, config.sample_rate, config.channels)?;
+        Ok(Self { pcm })
+    }
+
+    pub fn read(&self, buffer: &mut [i16]) -> Result<usize> {
+        self.pcm
+            .io_i16()?
+            .readi(buffer)
+            .context("Failed to read from capture device")
+    }
+}
+
+fn setup_pcm(pcm: &PCM, sample_rate: u32, channels: u16) -> Result<()> {
+    let hwp = HwParams::any(pcm).context("Failed to get HwParams")?;
+    hwp.set_access(Access::RWInterleaved)?;
+    hwp.set_format(Format::s16())?;
+    hwp.set_rate(sample_rate, alsa::ValueOr::Nearest)?;
+    hwp.set_channels(channels as u32)?;
+    pcm.hw_params(&hwp).context("Failed to set HwParams")?;
+
+    let swp = pcm.sw_params_current()?;
+    pcm.sw_params(&swp)?;
+    pcm.prepare()?;
+    Ok(())
+}

packages/client-v2/src/bin/client.rs

@@ -0,0 +1,11 @@
+use anyhow::Result;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    #[cfg(target_os = "linux")]
+    {
+        xiao::app::entry::run_xiao().await.unwrap();
+    }
+    println!("Only support Linux");
+    Ok(())
+}

packages/client-v2/src/lib.rs

@@ -0,0 +1,4 @@
+pub mod app;
+pub mod audio;
+pub mod net;
+pub mod utils;

packages/client-v2/src/net/discovery.rs

@@ -0,0 +1,45 @@
+use crate::net::protocol::ControlPacket;
+use anyhow::Result;
+use std::net::{IpAddr, SocketAddr};
+use std::time::Duration;
+use tokio::net::UdpSocket;
+
+pub const DISCOVERY_PORT: u16 = 53530;
+
+/// 服务发现模块，用于主从节点的自动发现
+pub struct Discovery;
+
+impl Discovery {
+    /// 主节点：启动广播，告知从节点自己的 TCP 端口
+    pub async fn start_broadcast(tcp_port: u16) -> Result<()> {
+        let socket = UdpSocket::bind("0.0.0.0:0").await?;
+        socket.set_broadcast(true)?;
+
+        let target: SocketAddr = format!("255.255.255.255:{}", DISCOVERY_PORT).parse()?;
+        let msg = postcard::to_allocvec(&ControlPacket::ServerHello { udp_port: tcp_port })?;
+
+        tokio::spawn(async move {
+            loop {
+                let _ = socket.send_to(&msg, target).await;
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            }
+        });
+
+        Ok(())
+    }
+
+    /// 从节点：监听广播，发现主节点的 IP 和 TCP 端口
+    pub async fn discover_master() -> Result<(IpAddr, u16)> {
+        let socket = UdpSocket::bind(format!("0.0.0.0:{}", DISCOVERY_PORT)).await?;
+        let mut buf = [0u8; 1024];
+
+        loop {
+            let (len, addr) = socket.recv_from(&mut buf).await?;
+            if let Ok(ControlPacket::ServerHello { udp_port }) =
+                postcard::from_bytes::<ControlPacket>(&buf[..len])
+            {
+                return Ok((addr.ip(), udp_port));
+            }
+        }
+    }
+}

packages/client-v2/src/net/mod.rs

@@ -0,0 +1,3 @@
+pub mod discovery;
+pub mod network;
+pub mod protocol;

packages/client-v2/src/net/network.rs

@@ -0,0 +1,126 @@
+use crate::net::protocol::{AudioPacket, ControlPacket};
+use anyhow::{Context, Result};
+use std::net::SocketAddr;
+use std::sync::Arc;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream, UdpSocket};
+
+/// UDP 音频传输
+pub struct AudioSocket {
+    socket: Arc<UdpSocket>,
+}
+
+impl AudioSocket {
+    pub async fn bind() -> Result<Self> {
+        let socket = UdpSocket::bind("0.0.0.0:0").await?;
+        Ok(Self {
+            socket: Arc::new(socket),
+        })
+    }
+
+    pub fn local_port(&self) -> Result<u16> {
+        Ok(self.socket.local_addr()?.port())
+    }
+
+    pub async fn send_packet(&self, packet: &AudioPacket, target: SocketAddr) -> Result<()> {
+        let bytes = postcard::to_allocvec(packet)?;
+        self.socket.send_to(&bytes, target).await?;
+        Ok(())
+    }
+
+    pub async fn recv_packet(&self, buf: &mut [u8]) -> Result<(AudioPacket, SocketAddr)> {
+        let (len, addr) = self.socket.recv_from(buf).await?;
+        let packet = postcard::from_bytes(&buf[..len])?;
+        Ok((packet, addr))
+    }
+
+    pub async fn punch(&self, target: SocketAddr) -> Result<()> {
+        self.socket.send_to(&[0u8; 1], target).await?;
+        Ok(())
+    }
+
+    pub fn clone_inner(&self) -> Arc<UdpSocket> {
+        self.socket.clone()
+    }
+}
+
+/// TCP 控制连接
+pub struct ControlConnection {
+    stream: TcpStream,
+}
+
+impl ControlConnection {
+    pub fn new(stream: TcpStream) -> Self {
+        Self { stream }
+    }
+
+    pub async fn send_packet(&mut self, packet: &ControlPacket) -> Result<()> {
+        let bytes = postcard::to_allocvec(packet)?;
+        self.stream.write_all(&bytes).await?;
+        Ok(())
+    }
+
+    pub async fn recv_packet(&mut self, buf: &mut [u8]) -> Result<ControlPacket> {
+        let len = self.stream.read(buf).await?;
+        if len == 0 {
+            return Err(anyhow::anyhow!("连接已关闭"));
+        }
+        let packet = postcard::from_bytes(&buf[..len])?;
+        Ok(packet)
+    }
+
+    pub fn split(
+        self,
+    ) -> (
+        tokio::net::tcp::OwnedReadHalf,
+        tokio::net::tcp::OwnedWriteHalf,
+    ) {
+        self.stream.into_split()
+    }
+}
+
+/// 主节点网络管理器
+pub struct MasterNetwork {
+    listener: TcpListener,
+    audio: AudioSocket,
+}
+
+impl MasterNetwork {
+    pub async fn setup(port: u16) -> Result<Self> {
+        let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await?;
+        let audio = AudioSocket::bind().await?;
+        Ok(Self { listener, audio })
+    }
+
+    pub async fn accept(&self) -> Result<(ControlConnection, SocketAddr)> {
+        let (stream, addr) = self.listener.accept().await?;
+        Ok((ControlConnection::new(stream), addr))
+    }
+
+    pub fn audio_socket(&self) -> &AudioSocket {
+        &self.audio
+    }
+}
+
+/// 从节点网络管理器
+pub struct SlaveNetwork {
+    control: ControlConnection,
+    audio: AudioSocket,
+}
+
+impl SlaveNetwork {
+    pub async fn connect(master_addr: SocketAddr) -> Result<Self> {
+        let stream = TcpStream::connect(master_addr)
+            .await
+            .context(format!("无法连接到主节点 TCP 地址: {}", master_addr))?;
+        let audio = AudioSocket::bind().await?;
+        Ok(Self {
+            control: ControlConnection::new(stream),
+            audio,
+        })
+    }
+
+    pub fn split(self) -> (ControlConnection, AudioSocket) {
+        (self.control, self.audio)
+    }
+}

packages/client-v2/src/net/protocol.rs

@@ -0,0 +1,45 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Copy)]
+pub enum ChannelRole {
+    Left,
+    Right,
+}
+
+impl ChannelRole {
+    pub fn to_string(&self) -> String {
+        match self {
+            ChannelRole::Left => "左声道".to_string(),
+            ChannelRole::Right => "右声道".to_string(),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum ControlPacket {
+    // 发现协议
+    ServerHello {
+        udp_port: u16, // UDP 音频流端口
+    },
+    // 握手协议
+    ClientIdentify {
+        role: ChannelRole,
+    },
+    // 时间同步 (持续进行)
+    Ping {
+        client_ts: u128,
+        seq: u32,
+    },
+    Pong {
+        client_ts: u128,
+        server_ts: u128,
+        seq: u32,
+    },
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct AudioPacket {
+    pub seq: u32,        // 序列号，用于丢包检测
+    pub timestamp: u128, // 目标播放时间 (主节点时间)
+    pub data: Vec<u8>,   // Opus 编码数据
+}

packages/client-v2/src/utils/alsa.rs

@@ -0,0 +1,85 @@
+#![cfg(target_os = "linux")]
+
+use anyhow::{Context, Result};
+use std::fs;
+use std::process::Command;
+
+const FIFO_PATH: &str = "/tmp/xiao_out.fifo";
+const REAL_ASOUND_CONF: &str = "/etc/asound.conf";
+const TEMP_ASOUND_CONF: &str = "/tmp/asound.xiao.conf";
+
+/// ALSA 音频重定向器，用于拦截系统音频输出到 FIFO 管道
+pub struct AlsaRedirector;
+
+impl AlsaRedirector {
+    pub fn new() -> Result<Self> {
+        Self::cleanup(); // 确保环境干净
+
+        let original_conf = fs::read_to_string(REAL_ASOUND_CONF).unwrap_or_default();
+
+        if !original_conf.contains("pcm.original_default") {
+            // 重命名原有的 default 逻辑，插入拦截器
+            let mut new_conf = original_conf.replace("pcm.!default", "pcm.original_default");
+            new_conf.push_str(&format!(
+                "\npcm.!default {{ type plug slave {{ pcm \"xiao_interceptor\" format S16_LE rate 48000 channels 2 }} }}\n\
+                pcm.xiao_interceptor {{ type file slave.pcm \"null\" file \"{}\" format \"raw\" }}\n",
+                FIFO_PATH
+            ));
+
+            fs::write(TEMP_ASOUND_CONF, new_conf)?;
+
+            // 挂载覆盖 /etc/asound.conf
+            let status = Command::new("mount")
+                .arg("--bind")
+                .arg(TEMP_ASOUND_CONF)
+                .arg(REAL_ASOUND_CONF)
+                .status()
+                .context("执行 mount 命令失败")?;
+
+            if !status.success() {
+                return Err(anyhow::anyhow!("挂载 asound.conf 失败"));
+            }
+
+            Self::restart_applications();
+        }
+
+        // 创建 FIFO 管道
+        let _ = Command::new("mkfifo").arg(FIFO_PATH).status();
+        let _ = Command::new("chmod").arg("666").arg(FIFO_PATH).status();
+
+        Ok(Self)
+    }
+
+    pub fn cleanup() {
+        let _ = Command::new("sh")
+            .arg("-c")
+            .arg(format!("umount -l {} >/dev/null 2>&1", REAL_ASOUND_CONF))
+            .status();
+        let _ = fs::remove_file(TEMP_ASOUND_CONF);
+        let _ = fs::remove_file(FIFO_PATH);
+        Self::restart_applications();
+    }
+
+    pub fn fifo_path() -> &'static str {
+        FIFO_PATH
+    }
+
+    pub fn restart_applications() {
+        // 重启媒体播放器
+        let _ = Command::new("sh")
+            .arg("-c")
+            .arg("/etc/init.d/mediaplayer restart >/dev/null 2>&1")
+            .status();
+        // 重启蓝牙
+        let _ = Command::new("sh")
+            .arg("-c")
+            .arg("/etc/init.d/bluetooth restart >/dev/null 2>&1")
+            .status();
+    }
+}
+
+impl Drop for AlsaRedirector {
+    fn drop(&mut self) {
+        Self::cleanup();
+    }
+}

packages/client-v2/src/utils/mod.rs

@@ -0,0 +1,2 @@
+pub mod alsa;
+pub mod sync;

packages/client-v2/src/utils/sync.rs

@@ -0,0 +1,224 @@
+use std::collections::VecDeque;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+/// 获取当前微秒级时间戳
+pub fn now_us() -> u128 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("时间倒流")
+        .as_micros()
+}
+
+/// 时钟同步管理器，用于计算主从节点间的时钟偏移
+/// 采用改进的 NTP 算法 + Kalman 滤波思想
+pub struct ClockSync {
+    /// 偏移量样本窗口
+    offsets: VecDeque<OffsetSample>,
+    /// 当前估计的时钟偏移 (server_time - client_time)
+    pub current_offset: i128,
+    /// RTT 样本窗口
+    rtts: VecDeque<i128>,
+    /// 当前估计的最小 RTT
+    min_rtt: i128,
+    /// 窗口大小
+    window_size: usize,
+    /// 时钟漂移率 (ppm: parts per million)
+    /// 正值表示从节点时钟比主节点快
+    drift_rate: f64,
+    /// 上次更新时间
+    last_update_time: u128,
+    /// 漂移率估计窗口
+    drift_samples: VecDeque<DriftSample>,
+}
+
+#[derive(Clone, Copy)]
+struct OffsetSample {
+    offset: i128,
+    rtt: i128,
+    timestamp: u128,
+}
+
+#[derive(Clone, Copy)]
+struct DriftSample {
+    offset: i128,
+    timestamp: u128,
+}
+
+impl ClockSync {
+    pub fn new(window_size: usize) -> Self {
+        Self {
+            offsets: VecDeque::with_capacity(window_size),
+            current_offset: 0,
+            rtts: VecDeque::with_capacity(window_size),
+            min_rtt: i128::MAX,
+            window_size,
+            drift_rate: 0.0,
+            last_update_time: now_us(),
+            drift_samples: VecDeque::with_capacity(60), // 保留 60 秒的样本
+        }
+    }
+
+    /// 更新时钟偏移估计 (NTP 算法)
+    /// 
+    /// NTP 时间戳标记:
+    /// t1 = client_send_ts  : 客户端发送 Ping 的时间
+    /// t2 = server_ts       : 服务器接收 Ping 的时间  
+    /// t3 = server_ts       : 服务器发送 Pong 的时间 (假设处理时间忽略不计)
+    /// t4 = client_recv_ts  : 客户端接收 Pong 的时间
+    ///
+    /// RTT = (t4 - t1) - (t3 - t2) = (t4 - t1) (因为 t3 = t2)
+    /// Offset = ((t2 - t1) + (t3 - t4)) / 2 = ((t2 - t1) + (t2 - t4)) / 2
+    ///        = t2 - (t1 + t4) / 2
+    pub fn update(&mut self, client_send_ts: u128, server_ts: u128, client_recv_ts: u128) {
+        let t1 = client_send_ts as i128;
+        let t2 = server_ts as i128;
+        let t4 = client_recv_ts as i128;
+
+        let rtt = t4 - t1;
+
+        // 过滤异常 RTT (局域网内 > 100ms 视为异常)
+        if rtt < 0 || rtt > 100_000 {
+            return;
+        }
+
+        // 计算时钟偏移: offset = server_time - client_time
+        // offset = t2 - (t1 + t4) / 2
+        let offset = t2 - (t1 + t4) / 2;
+
+        // 更新 RTT 窗口
+        self.rtts.push_back(rtt);
+        if self.rtts.len() > self.window_size {
+            self.rtts.pop_front();
+        }
+        self.min_rtt = *self.rtts.iter().min().unwrap_or(&rtt);
+
+        // 更新偏移量窗口
+        let sample = OffsetSample {
+            offset,
+            rtt,
+            timestamp: client_recv_ts,
+        };
+        self.offsets.push_back(sample);
+        if self.offsets.len() > self.window_size {
+            self.offsets.pop_front();
+        }
+
+        // 偏移量估计: 使用低 RTT 样本的中位数
+        // 原理: RTT 较小的样本受网络抖动影响小，时间测量更准确
+        let mut low_rtt_offsets: Vec<i128> = self
+            .offsets
+            .iter()
+            .filter(|s| s.rtt <= self.min_rtt + 5000) // 5ms 容差
+            .map(|s| s.offset)
+            .collect();
+
+        if !low_rtt_offsets.is_empty() {
+            low_rtt_offsets.sort_unstable();
+            let new_offset = low_rtt_offsets[low_rtt_offsets.len() / 2];
+
+            // 漂移率估计
+            self.estimate_drift(new_offset, client_recv_ts);
+
+            // 平滑更新偏移量 (避免突变)
+            let alpha = 0.3; // 低通滤波系数
+            self.current_offset =
+                (alpha * new_offset as f64 + (1.0 - alpha) * self.current_offset as f64) as i128;
+        }
+
+        self.last_update_time = client_recv_ts;
+    }
+
+    /// 估计时钟漂移率
+    /// 时钟漂移率 = d(offset) / dt
+    fn estimate_drift(&mut self, offset: i128, timestamp: u128) {
+        self.drift_samples.push_back(DriftSample { offset, timestamp });
+        if self.drift_samples.len() > 60 {
+            self.drift_samples.pop_front();
+        }
+
+        // 至少需要 10 秒的数据才能估计漂移
+        if self.drift_samples.len() < 10 {
+            return;
+        }
+
+        // 使用线性回归估计漂移率
+        let first = self.drift_samples.front().unwrap();
+        let last = self.drift_samples.back().unwrap();
+
+        let dt = (last.timestamp - first.timestamp) as f64;
+        let d_offset = (last.offset - first.offset) as f64;
+
+        if dt > 10_000_000.0 {
+            // 超过 10 秒
+            // drift_rate 单位: 微秒/秒 = ppm
+            let new_drift = d_offset / (dt / 1_000_000.0);
+
+            // 平滑更新漂移率
+            let beta = 0.1;
+            self.drift_rate = beta * new_drift + (1.0 - beta) * self.drift_rate;
+        }
+    }
+
+    /// 将本地时间转换为服务器(主节点)时间
+    /// 考虑时钟漂移补偿
+    pub fn to_server_time(&self, client_time: u128) -> u128 {
+        let base_server_time = (client_time as i128 + self.current_offset) as u128;
+
+        // 漂移补偿: 根据距离上次同步的时间，补偿时钟漂移
+        let elapsed_since_update = client_time.saturating_sub(self.last_update_time) as f64;
+        let drift_correction = (self.drift_rate * elapsed_since_update / 1_000_000.0) as i128;
+
+        (base_server_time as i128 + drift_correction) as u128
+    }
+
+    /// 将服务器(主节点)时间转换为本地时间
+    pub fn to_client_time(&self, server_time: u128) -> u128 {
+        // 简化版本，不考虑漂移补偿 (播放时主要用 to_server_time)
+        (server_time as i128 - self.current_offset) as u128
+    }
+
+    /// 获取当前估计的 RTT (微秒)
+    pub fn get_rtt(&self) -> i128 {
+        self.min_rtt
+    }
+
+    /// 获取当前时钟漂移率 (ppm)
+    pub fn get_drift_rate(&self) -> f64 {
+        self.drift_rate
+    }
+
+    /// 获取同步质量评估 (0-100, 越高越好)
+    pub fn get_sync_quality(&self) -> u8 {
+        if self.offsets.is_empty() {
+            return 0;
+        }
+
+        // 基于 RTT 稳定性和偏移量方差评估
+        let rtt_variance = self.calculate_variance(&self.rtts.iter().copied().collect::<Vec<_>>());
+        let offset_variance = self.calculate_variance(
+            &self.offsets.iter().map(|s| s.offset).collect::<Vec<_>>(),
+        );
+
+        // RTT 越稳定，方差越小，质量越高
+        let rtt_score = ((100_000.0 - rtt_variance.min(100_000.0)) / 100_000.0 * 50.0) as u8;
+        let offset_score = ((50_000.0 - offset_variance.min(50_000.0)) / 50_000.0 * 50.0) as u8;
+
+        rtt_score + offset_score
+    }
+
+    fn calculate_variance(&self, samples: &[i128]) -> f64 {
+        if samples.is_empty() {
+            return 0.0;
+        }
+        let mean = samples.iter().sum::<i128>() as f64 / samples.len() as f64;
+        let variance = samples
+            .iter()
+            .map(|&x| {
+                let diff = x as f64 - mean;
+                diff * diff
+            })
+            .sum::<f64>()
+            / samples.len() as f64;
+        variance.sqrt()
+    }
+}