fix(M2): wire vc_client's real voice plane through the C ABI, not just raw sockets

test_m2_voice passed against raw BSD sockets, but vc_client::stream_start/stop, UDP binding, and capture/recv were still VC_ERR_NOT_IMPLEMENTED stubs -- meaning vccli and any GUI client still couldn't actually talk. Implements the real client-side UDP-binding handshake, media key derivation, capture->encode->seal-> send and recv->open->decode->playback paths, plus server-side StreamInfo broadcast so peers learn about each other's streams via sync_remote_streams(). Adds test_voice_client_abi (two real vc_client instances, not raw sockets) and vccli --voice/--mute/--text flags, manually verified live between two instances. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-16 02:12:50 +02:00
parent 694494a5be
commit c693cab35c
13 changed files with 1017 additions and 30 deletions
--- a/core/src/core/client.cpp
+++ b/core/src/core/client.cpp
@@ -22,7 +22,9 @@
   static void close_sock(sock_t s) { ::close(s); }
 #endif

+#include <algorithm>
 #include <chrono>
+#include <cstring>

 #include "protocol/protocol.h"

@@ -107,6 +109,8 @@ vc_result vc_client::disconnect() {
        io_fd_.store(-1, std::memory_order_release);
    }

+    teardown_voice();
+
    if (io_thread_.joinable()) io_thread_.join();
    return VC_OK;
 }
@@ -114,6 +118,7 @@ vc_result vc_client::disconnect() {
 // ── io_thread_ entry point ────────────────────────────────────────────────────

 void vc_client::run_io(std::string host, uint16_t port) {
+    udp_host_ = host;
    set_state(VC_STATE_CONNECTING);

    // ── TCP connect ──────────────────────────────────────────────────────────
@@ -220,6 +225,7 @@ void vc_client::run_io(std::string host, uint16_t port) {
            }
        }

+        teardown_voice();
        tls_.reset();
        close_sock(sock);
        io_fd_.store(-1);
@@ -234,6 +240,28 @@ cleanup:
    return;
 }

+void vc_client::teardown_voice() {
+    udp_stop_.store(true, std::memory_order_release);
+    int ufd = udp_fd_.load(std::memory_order_acquire);
+    if (ufd != -1) {
+        close_sock(static_cast<sock_t>(ufd));
+        udp_fd_.store(-1, std::memory_order_release);
+    }
+    if (udp_thread_.joinable()) udp_thread_.join();
+    udp_ready_.store(false, std::memory_order_release);
+
+    audio_engine_.stop();
+    local_stream_active_.store(false, std::memory_order_release);
+    local_stream_pending_ = false;
+    local_encoder_.destroy();
+
+    media_send_crypto_.reset();
+    media_recv_crypto_.reset();
+
+    std::lock_guard lk(remote_streams_mu_);
+    remote_streams_.clear();
+}
+
 void vc_client::drain_sends() {
    while (true) {
        std::vector<uint8_t> frame;
@@ -282,6 +310,12 @@ void vc_client::handle_envelope(const voicecat::v1::Envelope& env) {
        case voicecat::v1::Envelope::kDisconnect:
            handle_disconnect(env.disconnect());
            break;
+        case voicecat::v1::Envelope::kUdpBinding:
+            handle_udp_binding_ack(env.udp_binding());
+            break;
+        case voicecat::v1::Envelope::kStreamAnnounceResult:
+            handle_stream_announce_result(env.stream_announce_result());
+            break;
        case voicecat::v1::Envelope::kPong:
            break;  // ignore keepalive responses
        default:
@@ -290,6 +324,8 @@ void vc_client::handle_envelope(const voicecat::v1::Envelope& env) {
 }

 void vc_client::handle_server_hello(const voicecat::v1::ServerHello& msg, uint64_t /*req_id*/) {
+    server_udp_port_ = static_cast<uint16_t>(msg.udp_port());
+
    // Server acknowledged our ClientHello. Now send AuthRequest (or queue it).
    std::optional<PendingAuth> auth;
    {
@@ -321,6 +357,12 @@ void vc_client::handle_auth_result(const voicecat::v1::AuthResult& msg) {
        server_session_id_ = msg.session_id();
        ev.user_id        = self_user_id_;
        set_state(VC_STATE_CONNECTED);
+
+        const std::string& tok = msg.udp_token();
+        if (tok.size() == udp_token_.size()) {
+            std::memcpy(udp_token_.data(), tok.data(), udp_token_.size());
+            start_udp_binding();
+        }
    } else {
        ev.text = msg.error().c_str();
    }
@@ -329,6 +371,7 @@ void vc_client::handle_auth_result(const voicecat::v1::AuthResult& msg) {

 void vc_client::handle_server_state(const voicecat::v1::ServerStateSnapshot& snap) {
    session_model_.apply_snapshot(snap);
+    for (const auto& u : snap.users()) sync_remote_streams(u);
    vc_event ev{};
    ev.type = VC_EVENT_CHANNEL_LIST;
    emit(ev);
@@ -350,14 +393,26 @@ void vc_client::handle_user_event(const voicecat::v1::UserEvent& ue) {
            ev.type = VC_EVENT_USER_JOINED;
            ev.text = nick.c_str();
            emit(ev);
+            sync_remote_streams(user);
            break;
-        case voicecat::v1::UserEvent::LEFT:
+        case voicecat::v1::UserEvent::LEFT: {
            ev.type = VC_EVENT_USER_LEFT;
            emit(ev);
+            std::lock_guard lk(remote_streams_mu_);
+            for (auto it = remote_streams_.begin(); it != remote_streams_.end();) {
+                if (it->second.first == user.id()) {
+                    audio_engine_.remove_stream(it->first);
+                    it = remote_streams_.erase(it);
+                } else {
+                    ++it;
+                }
+            }
            break;
+        }
        case voicecat::v1::UserEvent::UPDATED:
            ev.type = VC_EVENT_USER_UPDATED;
            emit(ev);
+            sync_remote_streams(user);
            break;
        default:
            break;
@@ -454,19 +509,323 @@ vc_result vc_client::send_text(vc_text_scope scope, uint32_t target_id, const ch
    return VC_OK;
 }

-// ── Stubs for audio/device (M2) ───────────────────────────────────────────────
+// ── M2: UDP binding ───────────────────────────────────────────────────────────

-vc_result vc_client::stream_start(const vc_stream_desc&, uint32_t*) {
-    return VC_ERR_NOT_IMPLEMENTED;
+void vc_client::start_udp_binding() {
+    voicecat::v1::Envelope req;
+    req.set_request_id(next_req_id_++);
+    req.mutable_udp_binding()->set_udp_token(
+        reinterpret_cast<const char*>(udp_token_.data()), udp_token_.size());
+    queue_envelope(req);
 }
-vc_result vc_client::stream_stop(uint32_t) { return VC_ERR_NOT_IMPLEMENTED; }
+
+void vc_client::handle_udp_binding_ack(const voicecat::v1::UdpBinding& msg) {
+    if (!msg.ack()) return;
+    finish_udp_binding();
+}
+
+void vc_client::finish_udp_binding() {
+    if (udp_ready_.load(std::memory_order_acquire)) return;
+    if (!tls_ || server_udp_port_ == 0) return;
+
+    media_send_crypto_ = voicecat::crypto::SodiumMediaCrypto::derive_send(*tls_, true);
+    media_recv_crypto_ = voicecat::crypto::SodiumMediaCrypto::derive_recv(*tls_, true);
+    if (!media_send_crypto_ || !media_recv_crypto_) {
+        emit_error(VC_ERR_CRYPTO, "failed to derive media keys");
+        return;
+    }
+
+    sock_t s = ::socket(AF_INET, SOCK_DGRAM, 0);
+    if (s == kBadSock) {
+        emit_error(VC_ERR_IO, "udp socket() failed");
+        return;
+    }
+
+    struct addrinfo hints{};
+    hints.ai_family   = AF_INET;
+    hints.ai_socktype = SOCK_DGRAM;
+    struct addrinfo* res = nullptr;
+    std::string port_str = std::to_string(server_udp_port_);
+    if (getaddrinfo(udp_host_.c_str(), port_str.c_str(), &hints, &res) != 0 || !res) {
+        close_sock(s);
+        emit_error(VC_ERR_IO, "udp hostname resolution failed");
+        return;
+    }
+
+    auto* sin = reinterpret_cast<sockaddr_in*>(res->ai_addr);
+    udp_dest_addr_ = sin->sin_addr.s_addr;
+    udp_dest_port_ = sin->sin_port;
+
+    auto pkt = voicecat::net::make_udp_binding_packet(udp_token_.data(), udp_token_.size());
+    ::sendto(s, reinterpret_cast<const char*>(pkt.data()), static_cast<int>(pkt.size()), 0,
+             res->ai_addr, static_cast<int>(res->ai_addrlen));
+    freeaddrinfo(res);
+
+    udp_fd_.store(static_cast<int>(s), std::memory_order_release);
+    udp_stop_.store(false, std::memory_order_release);
+    udp_ready_.store(true, std::memory_order_release);
+    udp_thread_ = std::thread([this] { run_udp_recv(); });
+}
+
+void vc_client::run_udp_recv() {
+    int fd = udp_fd_.load(std::memory_order_acquire);
+    if (fd == -1) return;
+
+#ifdef _WIN32
+    DWORD tv = 200;
+    setsockopt(static_cast<sock_t>(fd), SOL_SOCKET, SO_RCVTIMEO,
+               reinterpret_cast<const char*>(&tv), sizeof(tv));
+#else
+    struct timeval tv{0, 200000};
+    setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
+#endif
+
+    std::vector<uint8_t> buf(2048);
+    while (!udp_stop_.load(std::memory_order_acquire)) {
+        int n = static_cast<int>(::recv(static_cast<sock_t>(fd), reinterpret_cast<char*>(buf.data()),
+                                         static_cast<int>(buf.size()), 0));
+        if (n < static_cast<int>(voicecat::net::kVoiceHeaderSize)) continue;
+
+        voicecat::net::VoiceFrame hdr{};
+        if (!voicecat::net::parse_header(buf.data(), static_cast<size_t>(n), hdr)) continue;
+        if (hdr.type != voicecat::net::kFrameVoice) continue;
+        if (!media_recv_crypto_) continue;
+
+        size_t sealed_len = static_cast<size_t>(n) - voicecat::net::kVoiceHeaderSize;
+        std::vector<uint8_t> plain(sealed_len);
+        long plain_len = media_recv_crypto_->open(
+            buf.data() + voicecat::net::kVoiceHeaderSize, sealed_len, buf.data(),
+            voicecat::net::kVoiceHeaderSize, plain.data(), plain.size());
+        if (plain_len < 0) continue;
+        plain.resize(static_cast<size_t>(plain_len));
+
+        voicecat::audio::JitterBuffer::Frame jf;
+        jf.seq         = hdr.seq;
+        jf.timestamp   = hdr.timestamp;
+        jf.fec_present = (hdr.flags & voicecat::net::kFlagFecPresent) != 0;
+        jf.payload     = std::move(plain);
+        audio_engine_.push_recv_frame(hdr.ssrc, std::move(jf));
+    }
+}
+
+void vc_client::on_capture_frame(const int16_t* pcm, int samples) {
+    if (!local_stream_active_.load(std::memory_order_acquire)) return;
+    if (self_mic_muted_.load(std::memory_order_acquire)) return;
+    if (!media_send_crypto_) return;
+    int fd = udp_fd_.load(std::memory_order_acquire);
+    if (fd == -1) return;
+
+    uint8_t opus_buf[1500];
+    int opus_len = local_encoder_.encode(pcm, samples, opus_buf, sizeof(opus_buf));
+    if (opus_len <= 0) return;
+
+    voicecat::net::VoiceFrame hdr;
+    hdr.ssrc      = local_ssrc_;
+    hdr.seq       = static_cast<uint16_t>(media_send_crypto_->peek_send_counter());
+    hdr.timestamp = local_timestamp_;
+    local_timestamp_ += static_cast<uint32_t>(samples);
+
+    uint8_t header_bytes[voicecat::net::kVoiceHeaderSize];
+    voicecat::net::serialize_header(hdr, header_bytes);
+
+    uint8_t sealed[1500];
+    long sealed_len = media_send_crypto_->seal(opus_buf, static_cast<size_t>(opus_len),
+                                                header_bytes, voicecat::net::kVoiceHeaderSize,
+                                                sealed, sizeof(sealed));
+    if (sealed_len < 0) return;
+
+    std::vector<uint8_t> pkt(voicecat::net::kVoiceHeaderSize + static_cast<size_t>(sealed_len));
+    std::memcpy(pkt.data(), header_bytes, voicecat::net::kVoiceHeaderSize);
+    std::memcpy(pkt.data() + voicecat::net::kVoiceHeaderSize, sealed,
+                static_cast<size_t>(sealed_len));
+
+    sockaddr_in dest{};
+    dest.sin_family      = AF_INET;
+    dest.sin_addr.s_addr = udp_dest_addr_;
+    dest.sin_port        = udp_dest_port_;
+    ::sendto(static_cast<sock_t>(fd), reinterpret_cast<const char*>(pkt.data()),
+             static_cast<int>(pkt.size()), 0, reinterpret_cast<sockaddr*>(&dest), sizeof(dest));
+}
+
+void vc_client::ensure_audio_running() {
+    if (audio_engine_.running()) return;
+    voicecat::audio::AudioParams p;
+    p.sample_rate = 48000;
+    p.channels    = 1;
+    p.frame_ms    = 20;
+    audio_engine_.start(p, [this](const int16_t* pcm, int samples) { on_capture_frame(pcm, samples); });
+}
+
+void vc_client::sync_remote_streams(const voicecat::v1::User& user) {
+    if (user.id() == self_user_id_) return;
+
+    std::vector<uint32_t> current_ssrcs;
+    for (const auto& si : user.streams()) current_ssrcs.push_back(si.ssrc());
+
+    std::vector<std::pair<uint32_t, uint32_t>> newly_added;  // {ssrc, stream_id}
+    {
+        std::lock_guard lk(remote_streams_mu_);
+        for (const auto& si : user.streams()) {
+            uint32_t ssrc = si.ssrc();
+            if (remote_streams_.count(ssrc)) continue;
+
+            remote_streams_[ssrc] = {user.id(), si.stream_id()};
+
+            voicecat::codec::OpusParams p;
+            p.sample_rate = si.audio().sample_rate() ? si.audio().sample_rate() : 48000;
+            p.frame_ms    = si.audio().frame_ms() ? si.audio().frame_ms() : 20;
+            audio_engine_.init_recv_stream(ssrc, p);
+            audio_engine_.set_stream_mute(ssrc, self_deafened_.load(std::memory_order_acquire));
+
+            newly_added.emplace_back(ssrc, si.stream_id());
+        }
+
+        for (auto it = remote_streams_.begin(); it != remote_streams_.end();) {
+            if (it->second.first == user.id() &&
+                std::find(current_ssrcs.begin(), current_ssrcs.end(), it->first) ==
+                    current_ssrcs.end()) {
+                uint32_t stream_id = it->second.second;
+                audio_engine_.remove_stream(it->first);
+                it = remote_streams_.erase(it);
+
+                vc_event ev{};
+                ev.type      = VC_EVENT_STREAM_STOPPED;
+                ev.user_id   = user.id();
+                ev.stream_id = stream_id;
+                emit(ev);
+            } else {
+                ++it;
+            }
+        }
+    }
+
+    if (!newly_added.empty()) ensure_audio_running();
+    for (auto& [ssrc, stream_id] : newly_added) {
+        (void)ssrc;
+        vc_event ev{};
+        ev.type      = VC_EVENT_STREAM_STARTED;
+        ev.user_id   = user.id();
+        ev.stream_id = stream_id;
+        emit(ev);
+    }
+}
+
+// ── M2: stream / device control ──────────────────────────────────────────────
+
+vc_result vc_client::stream_start(const vc_stream_desc& desc, uint32_t* out_stream_id) {
+    if (state_net_.load(std::memory_order_acquire) != VC_STATE_CONNECTED) return VC_ERR_NOT_CONNECTED;
+    if (local_stream_active_.load(std::memory_order_acquire) || local_stream_pending_)
+        return VC_ERR_ALREADY;
+
+    uint32_t sid = next_local_stream_id_++;
+    local_stream_id_     = sid;
+    local_stream_pending_ = true;
+    if (out_stream_id) *out_stream_id = sid;
+
+    voicecat::v1::Envelope req;
+    req.set_request_id(next_req_id_++);
+    auto* ann = req.mutable_stream_announce();
+    ann->set_kind(desc.kind == VC_STREAM_SCREEN_AUDIO ? voicecat::v1::STREAM_SCREEN_AUDIO
+                  : desc.kind == VC_STREAM_AUX_DEVICE  ? voicecat::v1::STREAM_AUX_DEVICE
+                                                        : voicecat::v1::STREAM_MIC);
+    if (desc.label) ann->set_label(desc.label);
+    auto* audio = ann->mutable_requested_audio();
+    audio->set_sample_rate(48000);
+    audio->set_bitrate_bps(24000);
+    audio->set_frame_ms(20);
+    audio->set_fec(true);
+    queue_envelope(req);
+    return VC_OK;
+}
+
+void vc_client::handle_stream_announce_result(const voicecat::v1::StreamAnnounceResult& msg) {
+    if (!local_stream_pending_) return;
+    local_stream_pending_ = false;
+
+    if (!msg.ok()) {
+        emit_error(VC_ERR_PROTOCOL, msg.error().c_str());
+        return;
+    }
+
+    local_ssrc_      = msg.ssrc();
+    local_timestamp_ = 0;
+
+    voicecat::codec::OpusParams p;
+    const auto& eff = msg.effective_audio();
+    p.sample_rate = eff.sample_rate() ? eff.sample_rate() : 48000;
+    p.bitrate_bps = eff.bitrate_bps() ? eff.bitrate_bps() : 24000;
+    p.frame_ms    = eff.frame_ms() ? eff.frame_ms() : 20;
+    p.fec         = eff.fec();
+    local_frame_samples_ = static_cast<uint16_t>(voicecat::codec::opus_frame_samples(p));
+
+    if (!local_encoder_.init(p)) {
+        emit_error(VC_ERR_AUDIO, "opus encoder init failed");
+        return;
+    }
+
+    ensure_audio_running();
+    local_stream_active_.store(true, std::memory_order_release);
+
+    vc_event ev{};
+    ev.type      = VC_EVENT_STREAM_STARTED;
+    ev.user_id   = self_user_id_;
+    ev.stream_id = local_stream_id_;
+    emit(ev);
+}
+
+vc_result vc_client::stream_stop(uint32_t stream_id) {
+    if (state_net_.load(std::memory_order_acquire) != VC_STATE_CONNECTED) return VC_ERR_NOT_CONNECTED;
+    if (!local_stream_active_.load(std::memory_order_acquire) || stream_id != local_stream_id_)
+        return VC_ERR_INVALID_ARG;
+
+    local_stream_active_.store(false, std::memory_order_release);
+    local_encoder_.destroy();
+
+    voicecat::v1::Envelope req;
+    req.set_request_id(next_req_id_++);
+    req.mutable_stream_stop()->set_stream_id(stream_id);
+    queue_envelope(req);
+
+    vc_event ev{};
+    ev.type      = VC_EVENT_STREAM_STOPPED;
+    ev.user_id   = self_user_id_;
+    ev.stream_id = stream_id;
+    emit(ev);
+    return VC_OK;
+}
+
 vc_result vc_client::set_input_device(uint32_t, const char*) { return VC_ERR_NOT_IMPLEMENTED; }
 vc_result vc_client::set_input_mode(vc_input_mode) { return VC_ERR_NOT_IMPLEMENTED; }
 vc_result vc_client::set_push_to_talk(bool) { return VC_ERR_NOT_IMPLEMENTED; }
-vc_result vc_client::set_self_mute(bool, bool) { return VC_ERR_NOT_IMPLEMENTED; }
-vc_result vc_client::set_remote_stream(uint32_t, uint32_t, float, bool, bool) {
-    return VC_ERR_NOT_IMPLEMENTED;
+
+vc_result vc_client::set_self_mute(bool mic_muted, bool deafened) {
+    if (state_net_.load(std::memory_order_acquire) != VC_STATE_CONNECTED) return VC_ERR_NOT_CONNECTED;
+    self_mic_muted_.store(mic_muted, std::memory_order_release);
+    self_deafened_.store(deafened, std::memory_order_release);
+
+    std::lock_guard lk(remote_streams_mu_);
+    for (auto& [ssrc, info] : remote_streams_) {
+        (void)info;
+        audio_engine_.set_stream_mute(ssrc, deafened);
+    }
+    return VC_OK;
 }
+
+vc_result vc_client::set_remote_stream(uint32_t user_id, uint32_t stream_id, float gain,
+                                       bool muted, bool /*noise_reduction*/) {
+    if (state_net_.load(std::memory_order_acquire) != VC_STATE_CONNECTED) return VC_ERR_NOT_CONNECTED;
+    const auto* user = session_model_.find_user(user_id);
+    if (!user) return VC_ERR_INVALID_ARG;
+    for (const auto& s : user->streams) {
+        if (s.stream_id == stream_id) {
+            audio_engine_.set_stream_gain(s.ssrc, gain);
+            audio_engine_.set_stream_mute(s.ssrc, muted);
+            return VC_OK;
+        }
+    }
+    return VC_ERR_INVALID_ARG;
+}
+
 vc_result vc_client::list_devices(vc_device_kind, vc_device_list* out) {
    out->items = nullptr;
    out->count = 0;
--- a/core/src/core/client.h
+++ b/core/src/core/client.h
@@ -15,9 +15,14 @@
 #include <optional>
 #include <string>
 #include <thread>
+#include <unordered_map>
+#include <utility>
 #include <vector>

+#include "audio/audio_engine.h"
+#include "codec/opus_codec.h"
 #include "crypto/crypto.h"
+#include "net/voice_frame.h"
 #include "protocol/envelope.h"
 #include "protocol/protocol.h"
 #include "session/session.h"
@@ -103,6 +108,41 @@ struct vc_client {
    // Client-side session model
    voicecat::session::SessionModel session_model_;

+    // ── M2: UDP / media plane ────────────────────────────────────────────────────
+    std::array<uint8_t, 16> udp_token_{};
+    uint16_t                server_udp_port_{0};
+    std::string              udp_host_;
+
+    std::atomic<int>   udp_fd_{-1};
+    std::thread        udp_thread_;
+    std::atomic<bool>  udp_stop_{false};
+    std::atomic<bool>  udp_ready_{false};
+
+    std::unique_ptr<voicecat::crypto::SodiumMediaCrypto> media_send_crypto_;
+    std::unique_ptr<voicecat::crypto::SodiumMediaCrypto> media_recv_crypto_;
+
+    voicecat::audio::AudioEngine audio_engine_;
+    voicecat::codec::OpusEncoder local_encoder_;
+
+    std::atomic<bool>     local_stream_active_{false};
+    bool                  local_stream_pending_{false};  // announced, waiting for result
+    uint32_t              local_stream_id_{0};
+    uint32_t              local_ssrc_{0};
+    uint32_t              next_local_stream_id_{1};
+    uint32_t              local_timestamp_{0};
+    uint16_t              local_frame_samples_{960};
+
+    // ssrc → (user_id, stream_id) for remote streams already wired into audio_engine_.
+    mutable std::mutex                                            remote_streams_mu_;
+    std::unordered_map<uint32_t, std::pair<uint32_t, uint32_t>>   remote_streams_;
+
+    // Local UDP destination (server media endpoint), resolved once during binding.
+    uint32_t udp_dest_addr_{0};  // network byte order
+    uint16_t udp_dest_port_{0};  // network byte order
+
+    std::atomic<bool> self_mic_muted_{false};
+    std::atomic<bool> self_deafened_{false};
+
    // ── io_thread_ entry point ──────────────────────────────────────────────────
    void run_io(std::string host, uint16_t port);

@@ -114,6 +154,26 @@ struct vc_client {
    void handle_user_event(const voicecat::v1::UserEvent& ue);
    void handle_text_message(const voicecat::v1::TextMessage& msg);
    void handle_disconnect(const voicecat::v1::Disconnect& msg);
+    void handle_udp_binding_ack(const voicecat::v1::UdpBinding& msg);
+    void handle_stream_announce_result(const voicecat::v1::StreamAnnounceResult& msg);
+
+    // ── M2: UDP / media helpers ──────────────────────────────────────────────────
+    // Kicks off TCP UdpBinding request; called once after a successful AuthResult.
+    void start_udp_binding();
+    // Opens the UDP socket, sends the plaintext bootstrap packet, starts udp_thread_.
+    void finish_udp_binding();
+    // udp_thread_ entry point: recv loop, AEAD-open, decode, push to audio_engine_.
+    void run_udp_recv();
+    // capture_cb passed to audio_engine_.start(): encode + seal + send one frame.
+    void on_capture_frame(const int16_t* pcm, int samples);
+    // Inspect a User proto's streams and wire up any new remote ssrc into audio_engine_,
+    // emitting VC_EVENT_STREAM_STARTED/STOPPED as streams appear/disappear.
+    void sync_remote_streams(const voicecat::v1::User& user);
+    // Starts audio_engine_ (capture+playback) if not already running.
+    void ensure_audio_running();
+    // Joins udp_thread_, stops audio_engine_, clears media crypto/remote-stream state.
+    // Safe to call multiple times. Called both from run_io()'s cleanup and disconnect().
+    void teardown_voice();

    // ── Helpers (io_thread_ and caller threads) ─────────────────────────────────
    // Queue an encoded envelope to be sent on io_thread_.
--- a/core/src/session/session.cpp
+++ b/core/src/session/session.cpp
@@ -14,8 +14,36 @@ const User* SessionModel::find_user(uint32_t id) const {
    return nullptr;
 }

+std::pair<const User*, const Stream*> SessionModel::find_user_by_ssrc(uint32_t ssrc) const {
+    for (auto& u : users_) {
+        for (auto& s : u.streams) {
+            if (s.ssrc == ssrc) return {&u, &s};
+        }
+    }
+    return {nullptr, nullptr};
+}
+
 #ifdef VOICECAT_HAS_NET

+namespace {
+std::vector<Stream> copy_streams(
+    const google::protobuf::RepeatedPtrField<voicecat::v1::StreamInfo>& src) {
+    std::vector<Stream> out;
+    out.reserve(src.size());
+    for (const auto& pb : src) {
+        Stream s;
+        s.stream_id   = pb.stream_id();
+        s.ssrc        = pb.ssrc();
+        s.kind        = static_cast<int>(pb.kind());
+        s.label       = pb.label();
+        s.sample_rate = pb.audio().sample_rate() ? pb.audio().sample_rate() : 48000;
+        s.frame_ms    = pb.audio().frame_ms() ? pb.audio().frame_ms() : 20;
+        out.push_back(std::move(s));
+    }
+    return out;
+}
+}  // namespace
+
 void SessionModel::apply_snapshot(const voicecat::v1::ServerStateSnapshot& snap) {
    channels_.clear();
    for (const auto& pb : snap.channels()) {
@@ -32,6 +60,7 @@ void SessionModel::apply_snapshot(const voicecat::v1::ServerStateSnapshot& snap)
        u.nickname   = pb.nickname();
        u.is_guest   = pb.is_guest();
        u.channel_id = pb.channel_id();
+        u.streams    = copy_streams(pb.streams());
        users_.push_back(std::move(u));
    }
 }
@@ -46,6 +75,7 @@ void SessionModel::apply_user_event(const voicecat::v1::UserEvent& ev) {
        u.nickname   = pb.nickname();
        u.is_guest   = pb.is_guest();
        u.channel_id = pb.channel_id();
+        u.streams    = copy_streams(pb.streams());

        auto it = std::find_if(users_.begin(), users_.end(),
                               [&](const User& x) { return x.id == u.id; });
--- a/core/src/session/session.h
+++ b/core/src/session/session.h
@@ -9,6 +9,7 @@

 #include <cstdint>
 #include <string>
+#include <utility>
 #include <vector>

 #ifdef VOICECAT_HAS_NET
@@ -30,6 +31,8 @@ struct Stream {
    uint32_t    ssrc{0};
    int         kind{0};
    std::string label;
+    uint32_t    sample_rate{48000};
+    uint32_t    frame_ms{20};
 };

 struct User {
@@ -48,6 +51,10 @@ class SessionModel {
    const Channel* find_channel(uint32_t id) const;
    const User*    find_user(uint32_t id) const;

+    // Find the user that owns a given media ssrc, and the matching Stream entry.
+    // Returns {nullptr, nullptr} if not found.
+    std::pair<const User*, const Stream*> find_user_by_ssrc(uint32_t ssrc) const;
+
 #ifdef VOICECAT_HAS_NET
    void apply_snapshot(const voicecat::v1::ServerStateSnapshot& snap);
    void apply_user_event(const voicecat::v1::UserEvent& ev);