// Status: minimal — connection table + REQUEST→RESPONSE/RST routing
// + TSI control DGRAM (port 1024..1031) handling.
//
// What we DON'T have yet (later commits):
//   - tsi_stream::TsiStreamProxy (TCP backend) — TSI_LISTEN
//     creates a stub Proxy that always replies success but doesn't
//     actually bind a host socket
//   - epoll-driven proxy event handling — without TCP backend, no
//     events to drive
//   - reaper thread — proxies just live until VM shutdown
//   - DGRAM (UDP) proxies via tsi_dgram

#![allow(dead_code)]

use std::collections::{HashMap, VecDeque};
use std::net::{SocketAddr, TcpStream, UdpSocket};
use std::os::fd::FromRawFd;
use std::os::unix::net::UnixStream;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{mpsc, Arc, Mutex};
use std::time::Instant;

/// Bind a dual-stack IPv6+IPv4 ephemeral UDP socket on the host.
///
/// We need a single socket that can `send_to` both IPv4 destinations
/// (e.g. `1.1.1.1:53` for DNS, Cloudflare's IPv4 QUIC edge) and
/// native IPv6 destinations (e.g. `2606:4700:4700::1111:53`, IPv6
/// QUIC edges) on behalf of guests that opened a dual-stack
/// `socket(AF_INET6, SOCK_DGRAM)` with IPV6_V6ONLY=0 — Go's
/// `net.ListenUDP("udp", ...)`, Node's `dgram.createSocket('udp')`,
/// quic-go (cloudflared), and the modern libc idiom for "I want one
/// UDP socket for both address families."
///
/// Linux defaults `IPV6_V6ONLY=0`, but macOS — which is where the
/// muxer actually runs — defaults to `=1`. Explicitly setsockopt to
/// 0 so v4-mapped-v6 routing actually works.
/// Hard cap on concurrent host-side UDP flows (each = one host UDP fd + one mio
/// registration). When exceeded, the least-recently-used flow is evicted +
/// torn down. Sized far above any realistic in-flight DNS concurrency (npm's
/// resolver fan-out is tens, not hundreds) so completed lookups are what get
/// recycled — never an in-flight one — while bounding fd use absolutely.
const UDP_FLOW_CAP: usize = 256;

/// Desired soft `RLIMIT_NOFILE` for a process that runs the muxer. Must clear
/// `UDP_FLOW_CAP` host UDP fds (each DNS `getaddrinfo` opens a fresh ephemeral
/// flow) PLUS the recv-clone per flow, the vsock/blk/kqueue/log fds, and a
/// margin. macOS's `kern.maxfilesperproc` is ~10k, so 8192 is safely
/// raisable.
const TARGET_NOFILE: u64 = 8192;

/// The soft fd limit to set, given the current `(soft, hard)`: raise toward
/// [`TARGET_NOFILE`] but never above the hard cap and never *lower* an
/// already-generous inherited limit. Pure (no syscalls) so the policy is
/// unit-testable.
fn target_nofile(soft: u64, hard: u64) -> u64 {
    TARGET_NOFILE.min(hard).max(soft)
}

/// Raise this process's `RLIMIT_NOFILE` soft limit toward [`TARGET_NOFILE`] so a
/// DNS-heavy workload doesn't exhaust host fds before the muxer's
/// [`UDP_FLOW_CAP`] LRU bound recycles flows.
///
/// Why this matters: when the worker is launched from a process with macOS's
/// default 256-fd soft limit (e.g. a Node/launchd parent), `UDP_FLOW_CAP=256`
/// host UDP flows plus the worker's other fds blow past 256 — so a build's apt
/// fan-out resolves fine early, then `getaddrinfo` for a *later* host
/// (github.com) fails to `bind` a fresh UDP socket and the guest sees
/// `EAI_AGAIN`. Raising our own limit removes that dependency on whatever the
/// parent happened to set. Idempotent + best-effort (a failure just keeps the
/// inherited limit). Called from every muxer constructor — the choke point for
/// egress — so it covers the worker subprocess and any in-process VM alike.
// `rlim_t` happens to be u64 on macOS and 64-bit Linux (where the casts
// are "unnecessary"), but it is platform-defined — keep the explicit
// widening so any future target with a narrower rlim_t still compiles.
#[allow(clippy::unnecessary_cast)]
pub fn raise_nofile_limit() {
    #[cfg(unix)]
    // SAFETY: plain getrlimit/setrlimit on a zeroed `rlimit`; we only ever
    // RAISE the soft limit (clamped to the hard limit), never lower it.
    unsafe {
        let mut rl: libc::rlimit = std::mem::zeroed();
        if libc::getrlimit(libc::RLIMIT_NOFILE, &mut rl) == 0 {
            let want = target_nofile(rl.rlim_cur as u64, rl.rlim_max as u64);
            if want > rl.rlim_cur as u64 {
                rl.rlim_cur = want as libc::rlim_t;
                let _ = libc::setrlimit(libc::RLIMIT_NOFILE, &rl);
            }
        }
    }
}

fn bind_dual_stack_udp() -> std::io::Result<UdpSocket> {
    // SAFETY: standard libc dance. The fd is unconditionally
    // adopted via FromRawFd at the end; on early-error paths we
    // close it explicitly.
    unsafe {
        let fd = libc::socket(libc::AF_INET6, libc::SOCK_DGRAM, 0);
        if fd < 0 {
            return Err(std::io::Error::last_os_error());
        }
        let zero: libc::c_int = 0;
        if libc::setsockopt(
            fd,
            libc::IPPROTO_IPV6,
            libc::IPV6_V6ONLY,
            &zero as *const _ as *const libc::c_void,
            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
        ) != 0
        {
            let e = std::io::Error::last_os_error();
            libc::close(fd);
            return Err(e);
        }
        // Bind to `[::]:0` — kernel picks an ephemeral port.
        let mut sa: libc::sockaddr_in6 = std::mem::zeroed();
        sa.sin6_family = libc::AF_INET6 as _;
        // sin6_addr is `::` (in6addr_any) from the zeroed init.
        // sin6_port = 0 (ephemeral) likewise.
        if libc::bind(
            fd,
            &sa as *const _ as *const libc::sockaddr,
            std::mem::size_of::<libc::sockaddr_in6>() as libc::socklen_t,
        ) != 0
        {
            let e = std::io::Error::last_os_error();
            libc::close(fd);
            return Err(e);
        }
        Ok(UdpSocket::from_raw_fd(fd))
    }
}

use super::mux_profile::{self, Stage};
use super::muxer_rxq::{MuxerRxQ, MUXER_RXQ_CONTROL_RESERVE};
use super::muxer_thread::{self, MuxerCmd, MuxerStream};
use super::packet::{
    Header, RxPacket, VSOCK_HOST_CID, VSOCK_OP_CREDIT_REQUEST, VSOCK_OP_CREDIT_UPDATE,
    VSOCK_OP_REQUEST, VSOCK_OP_RESPONSE, VSOCK_OP_RST, VSOCK_OP_RW, VSOCK_OP_SHUTDOWN,
    VSOCK_TYPE_DGRAM, VSOCK_TYPE_STREAM,
};
use super::proxy::{proxy_key, Proxy};
use super::tsi_stream::TsiListener;

/// Guest-Linux ABI address-family constants. The TSI wire protocol
/// carries the *guest's* `AF_INET` / `AF_INET6` value verbatim in
/// every TSI_LISTEN / TSI_CONNECT / TSI_SENDTO_ADDR control message
/// and in PROXY_CREATE's `family` field. We have to compare against
/// the Linux values, not the host's `libc::AF_INET6` — that constant
/// is 30 on macOS while Linux uses 10. Using the host value made
/// every AF_INET6 dual-stack DGRAM (Node `dgram('udp')`, Go's
/// `ListenUDP("udp")`, quic-go / cloudflared) and every AF_INET6
/// TCP `connect()` fail silently on macOS hosts.
pub(crate) const LINUX_AF_INET: u16 = 2;
pub(crate) const LINUX_AF_INET6: u16 = 10;

/// AF_TSI control op opcodes, encoded as the dst_port of a DGRAM
/// packet sent by the guest kernel's TSI patches. See the patch
/// series under `kernel-build/patches/af-tsi/`.
pub const TSI_PROXY_PORT: u32 = 620;
pub const VSOCK_ENV_PORT: u32 = 1026;
pub const TSI_PROXY_CREATE: u32 = 1024;
pub const TSI_CONNECT: u32 = 1025;
pub const TSI_GETNAME: u32 = 1026;
pub const TSI_SENDTO_ADDR: u32 = 1027;
pub const TSI_SENDTO_DATA: u32 = 1028;
pub const TSI_LISTEN: u32 = 1029;
pub const TSI_ACCEPT: u32 = 1030;
pub const TSI_PROXY_RELEASE: u32 = 1031;

/// Snapshot record for one TSI listener: the (cid, peer_port, vm_port,
/// family, socktype) tuple needed to re-bind a host TcpListener on
/// restore. The host listener itself can't be serialized (OS socket
/// fd) — it gets rebuilt on restore via `restore_tsi_listeners`.
#[derive(Clone, Debug)]
pub struct TsiListenerSnapshot {
    pub cid: u64,
    pub peer_port: u32,
    pub vm_port: u32,
    pub family: u16,
    pub socktype: u16,
    /// AF_INET userland port; preserved across snapshot/restore so
    /// `Vm::expose_tcp(host_port, guest_port)` keeps working after
    /// resume.
    pub inet_port: Option<u16>,
}

impl TsiListenerSnapshot {
    /// Fixed wire size of one record (see [`write_to`](Self::write_to)).
    pub const WIRE_LEN: usize = 24;

    /// The single canonical codec for a TSI listener record — lives with the
    /// type so every snapshot pipeline serializes it identically (Phase-3
    /// backend unification). Fixed 24-byte little-endian layout; `inet_port`
    /// rides as a `u32` with `u32::MAX` as the `None` sentinel (real ports are
    /// u16, so no collision). Fixed width keeps the bounds-checked reader's
    /// per-record `cap_count` exact.
    ///
    /// ```text
    ///   cid u64 · peer_port u32 · vm_port u32 · family u16 · socktype u16
    ///   inet_port u32 (u32::MAX = None)
    /// ```
    pub fn write_to<W: std::io::Write>(&self, w: &mut W) -> std::io::Result<()> {
        w.write_all(&self.cid.to_le_bytes())?;
        w.write_all(&self.peer_port.to_le_bytes())?;
        w.write_all(&self.vm_port.to_le_bytes())?;
        w.write_all(&self.family.to_le_bytes())?;
        w.write_all(&self.socktype.to_le_bytes())?;
        w.write_all(&self.inet_port.map_or(u32::MAX, u32::from).to_le_bytes())?;
        Ok(())
    }

    /// Inverse of [`write_to`](Self::write_to). Reads exactly [`WIRE_LEN`](Self::WIRE_LEN) bytes.
    pub fn read_from<R: std::io::Read>(r: &mut R) -> std::io::Result<TsiListenerSnapshot> {
        let mut e = [0u8; Self::WIRE_LEN];
        r.read_exact(&mut e)?;
        let inet_raw = u32::from_le_bytes([e[20], e[21], e[22], e[23]]);
        Ok(TsiListenerSnapshot {
            cid: u64::from_le_bytes([e[0], e[1], e[2], e[3], e[4], e[5], e[6], e[7]]),
            peer_port: u32::from_le_bytes([e[8], e[9], e[10], e[11]]),
            vm_port: u32::from_le_bytes([e[12], e[13], e[14], e[15]]),
            family: u16::from_le_bytes([e[16], e[17]]),
            socktype: u16::from_le_bytes([e[18], e[19]]),
            inet_port: (inet_raw != u32::MAX).then_some(inet_raw as u16),
        })
    }
}

/// State for one TSI listener proxy. Created on TSI_PROXY_CREATE,
/// activated on TSI_LISTEN. We don't yet bind a host socket (no
/// tsi_stream port); we just remember the (local_port, peer_port,
/// vm_port) so subsequent PROXY_CREATE/LISTEN/ACCEPT calls can be
/// answered.
struct TsiState {
    family: u16,
    socktype: u16,
    /// vm_port from TSI_LISTEN — the underlying *VSOCK* port the
    /// guest assigned to its inner proxy socket (scrambled-looking
    /// value, not the AF_INET port). Used to route the host →
    /// guest leg of the connection.
    vm_port: Option<u32>,
    /// The AF_INET port the guest userspace passed to `bind(2)`
    /// (e.g. 80 for nginx, 9222 for Chrome DevTools). Extracted
    /// from the `tsi_listen_req::addr` sockaddr_in payload of
    /// TSI_LISTEN. Used by `Vm::expose_tcp(host_port, guest_port)`
    /// to pin the host TCP forwarder to a specific guest service
    /// when the guest has multiple AF_INET listeners.
    inet_port: Option<u16>,
    /// Host TCP listener bound on TSI_LISTEN. Drops on
    /// PROXY_RELEASE (or VM shutdown).
    listener: Option<TsiListener>,
}

/// In-flight inbound TCP conn that's been accepted on the host
/// side but the guest hasn't sent RESPONSE yet.
pub struct PendingInbound {
    pub cid: u64,
    pub host_src_port: u32,
    pub vm_port: u32,
    pub stream: Option<MuxerStream>,
}

/// In-flight outbound: guest sent TSI_CONNECT(peer_port, target);
/// we opened TCP to target and now park it until the guest's
/// follow-up VSOCK_OP_REQUEST(src=peer_port, dst=TSI_PROXY_PORT)
/// arrives. Keyed by (cid, peer_port).
pub struct PendingOutbound {
    pub cid: u64,
    pub peer_port: u32,
    pub tcp: Option<TcpStream>,
}

/// Per-conn flow-control accounting. Two directions:
///
/// 1. Guest → host (the bytes the guest writes to us, that we
///    forward to host TCP). The vsock peer sees us as having
///    `OUR_BUF_ALLOC` of receive buffer; every byte they send to
///    us must be ACKed via `our_fwd_cnt` or they stop sending
///    once they've used (peer_buf_alloc - 0) = OUR_BUF_ALLOC
///    bytes. We bump `our_fwd_cnt` as we consume (which is
///    immediate, since we forward straight to TCP), and emit
///    CREDIT_UPDATE when half of the buffer has been consumed
///    since last update.
///
/// 2. Host → guest (the bytes we read from host TCP, that we
///    push as RW packets into the guest's RX queue). The guest
///    advertises its RX buffer via `buf_alloc` on every header
///    and ACKs consumption via `fwd_cnt`. We must respect
///    `available = peer_buf_alloc - (tx_cnt - peer_fwd_cnt)` —
///    when that drops to zero, we MUST pause reading from host
///    TCP, or the guest's vsock RX socket buffer fills and
///    bytes are silently dropped (see issue: 4 MiB GET
///    truncating at ~1.5 MiB). When peer_fwd_cnt advances via
///    a subsequent RW or CREDIT_UPDATE from the guest, we
///    resume.
///
/// Tracked under `Arc<Mutex<...>>` so both the dispatcher path
/// (handle_rw / handle_credit) and the on_data closure invoked
/// from the muxer-io thread share a single view of credit.
pub struct ProxyConnState {
    /// Bytes WE'VE consumed from the guest's TX. Advances on
    /// each handle_rw call as we forward to host TCP.
    pub our_fwd_cnt: u32,
    /// Snapshot of our_fwd_cnt at the last CREDIT_UPDATE we sent
    /// to the guest. Used to decide when to emit the next one.
    pub last_credit_fwd_cnt: u32,
    /// Guest socket's local port — the `dst_port` of every RW
    /// we push toward the guest.
    pub guest_dst_port: u32,
    /// Last buf_alloc the guest advertised. Updated on EVERY
    /// inbound packet from the guest (RW + CREDIT_UPDATE +
    /// CREDIT_REQUEST all carry credit fields in their header).
    pub peer_buf_alloc: u32,
    /// Last fwd_cnt the guest reported — how many bytes of what
    /// we sent it has consumed.
    pub peer_fwd_cnt: u32,
    /// Cumulative bytes we've sent toward the guest on this
    /// conn. Wraps mod 2^32 (vsock uses u32 wrapping arithmetic).
    pub tx_cnt: u32,
    /// host_src_port — the key the muxer-io thread uses to
    /// route Write/Close/PauseReadable/ResumeReadable commands
    /// for this conn.
    pub host_src_port: u32,
    /// True when we asked muxer-io to pause READABLE on the
    /// host TCP because we ran out of credit. Cleared by
    /// handle_rw / handle_credit when credit returns and we
    /// send ResumeReadable.
    pub paused_readable: bool,
    /// Shared flag with the muxer-io thread's `Conn` so the
    /// per-conn read loop can break out of `loop { read }`
    /// inline (without round-tripping a PauseReadable command,
    /// which would be processed only on the next poll wakeup
    /// — by which time many more bytes might have been read).
    /// Set by `try_reserve_credit` when credit hits the low
    /// watermark; cleared when ResumeReadable is processed.
    pub pause_flag: Arc<AtomicBool>,
    /// Host→guest RX packets that could not be enqueued because the
    /// shared rxq was at its cap. Stashed here (NOT dropped) and
    /// re-pushed in order by handle_rw / handle_credit once the guest
    /// drains the queue — zero byte loss. Bounded: the io thread pauses
    /// after the overflowing read, so this holds at most the in-flight
    /// chunk(s) for this conn until resume.
    pub pending_rx: VecDeque<RxPacket>,
}

/// Low watermark below which the on_data path stops pushing
/// new bytes and trips the pause flag. Set generously above
/// zero so the in-flight packet that triggers the pause still
/// fits within `peer_buf_alloc`. Linux's vsock_stream tends to
/// advertise 256 KiB; 4 KiB is the smallest individual packet
/// we'd ever push (MAX_PKT_PAYLOAD in muxer_thread), so 4 KiB
/// is the absolute floor — any packet pushed when remaining
/// credit ≥ 4 KiB still fits.
pub const CREDIT_LOW_WATERMARK: u32 = 4096;

impl ProxyConnState {
    pub fn new(guest_dst_port: u32, host_src_port: u32) -> Self {
        Self {
            our_fwd_cnt: 0,
            last_credit_fwd_cnt: 0,
            guest_dst_port,
            peer_buf_alloc: 0,
            peer_fwd_cnt: 0,
            tx_cnt: 0,
            host_src_port,
            paused_readable: false,
            pause_flag: Arc::new(AtomicBool::new(false)),
            pending_rx: VecDeque::new(),
        }
    }

    /// Available credit toward the guest: how many more bytes
    /// we can push before the guest's RX buffer is conceptually
    /// full. Uses u32 wrapping arithmetic (vsock spec) — if
    /// peer_buf_alloc is zero the result is zero.
    pub fn available_credit(&self) -> u32 {
        let in_flight = self.tx_cnt.wrapping_sub(self.peer_fwd_cnt);
        self.peer_buf_alloc.saturating_sub(in_flight)
    }

    /// Update peer credit fields from an incoming header. Called
    /// on every RW / CREDIT_UPDATE / CREDIT_REQUEST. Returns true
    /// if we were paused AND can now resume — i.e. credit is above
    /// the watermark AND any rxq back-pressure stash has been fully
    /// flushed (callers drain `pending_rx` *before* calling this, so
    /// a non-empty stash here means the shared rxq is still full and
    /// we must stay paused). On resume the caller sends ResumeReadable.
    pub fn observe_peer(&mut self, peer_buf_alloc: u32, peer_fwd_cnt: u32) -> bool {
        self.peer_buf_alloc = peer_buf_alloc;
        self.peer_fwd_cnt = peer_fwd_cnt;
        if self.paused_readable
            && self.pending_rx.is_empty()
            && self.available_credit() > CREDIT_LOW_WATERMARK
        {
            self.paused_readable = false;
            self.pause_flag.store(false, Ordering::Release);
            true
        } else {
            false
        }
    }
}

/// Buffer size we advertise to the guest — the cap on guest→host bytes
/// in flight (sent but not yet `fwd_cnt`-acked).
///
/// History: this was `u32::MAX / 2` ("effectively unlimited") on the
/// theory that we forward to the host synchronously. That was false
/// under load: bytes are QUEUED on the muxer-io thread until the host
/// socket drains, and with credit granted at receipt the guest could
/// pump an entire multi-GiB stream (a rootfs tar over exec) into host
/// RAM (builder #6: 7.3 GiB worker footprint, quadratic drain, a
/// "hung" bake). Now that credit is granted on DELIVERY (`on_drained`),
/// this constant is the real end-to-end backpressure window: big
/// enough to keep the relay at line rate (the io thread drains a local
/// socket at GiB/s; 8 MiB ≫ bandwidth×latency here), small enough to
/// bound per-conn host buffering.
pub const OUR_BUF_ALLOC: u32 = 8 * 1024 * 1024;
/// Send a CREDIT_UPDATE to the guest each time this many delivered
/// bytes have accumulated since the last one. Half the window keeps
/// the guest's view fresh long before it could stall on a full window.
const CREDIT_UPDATE_THRESHOLD: u32 = OUR_BUF_ALLOC / 2;

fn body_le_u16(body: &[u8], offset: usize) -> Option<u16> {
    let bytes = body.get(offset..offset + 2)?.try_into().ok()?;
    Some(u16::from_le_bytes(bytes))
}

fn body_be_u16(body: &[u8], offset: usize) -> Option<u16> {
    let bytes = body.get(offset..offset + 2)?.try_into().ok()?;
    Some(u16::from_be_bytes(bytes))
}

fn body_le_u32(body: &[u8], offset: usize) -> Option<u32> {
    let bytes = body.get(offset..offset + 4)?.try_into().ok()?;
    Some(u32::from_le_bytes(bytes))
}

/// Guest connects to (HOST_CID, VSOCK_ENV_PORT) via plain AF_VSOCK to
/// fetch its env JSON. Set at startup by `--env` / `--env-file`. The
/// blob is delivered as a single VSOCK_OP_RW followed by SHUTDOWN.
/// Format: `{"env":{"K":"V",...},"secrets":{"K":"V",...}}`.
pub static VSOCK_ENV_JSON: Mutex<Option<String>> = Mutex::new(None);

pub fn set_env_json(json: String) {
    *VSOCK_ENV_JSON.lock().unwrap() = Some(json);
}

/// Cached `$SUPERMACHINE_VSOCK_TRACE` flag. Calling
/// `std::env::var_os(...)` per packet path is a real perf hit:
/// `getenv` takes a libc global lock + linear scan of `environ`.
/// At nginx c=64 it dominated muxer-io non-syscall CPU. Read
/// once on first use, cache. Toggling at runtime no longer
/// supported — set the env before the worker starts.
#[inline]
pub fn vsock_trace_enabled() -> bool {
    use std::sync::atomic::{AtomicU8, Ordering};
    static CACHED: AtomicU8 = AtomicU8::new(0);
    let v = CACHED.load(Ordering::Relaxed);
    if v != 0 {
        return v == 2;
    }
    // The actual env lookup. Inside the cached helper we call the
    // libc path directly; the sed-rewrite that replaced sites in
    // this file would otherwise recurse here.
    let on = crate::trace::enabled("vsock");
    CACHED.store(if on { 2 } else { 1 }, Ordering::Relaxed);
    on
}

pub struct VsockMuxer {
    cid: u64,
    /// (local_port, peer_port) → Proxy.
    proxies: Mutex<HashMap<u64, Box<dyn Proxy>>>,
    /// (cid, host_src_port) → PendingInbound waiting for guest's
    /// RESPONSE to a host-initiated REQUEST.
    pending_inbound: Arc<Mutex<HashMap<(u64, u32), PendingInbound>>>,
    /// (cid, peer_port) → PendingOutbound. Set by TSI_CONNECT,
    /// consumed by VSOCK_OP_REQUEST(dst=TSI_PROXY_PORT) from guest.
    pending_outbound: Mutex<HashMap<(u64, u32), PendingOutbound>>,
    /// (cid, peer_port) → staged dst SocketAddr from TSI_SENDTO_ADDR,
    /// consumed by the next TSI_SENDTO_DATA.
    udp_dst: Mutex<HashMap<(u64, u32), SocketAddr>>,
    /// (cid, peer_port) → persistent UdpSocket. Send fd shared with
    /// the muxer-io thread (which holds a try_clone()'d recv fd).
    ///
    /// Bounded by `UDP_FLOW_CAP` with LRU eviction (see `send_udp_payload`):
    /// glibc `getaddrinfo` uses a FRESH ephemeral source port per DNS lookup,
    /// so a burst (e.g. `npm install`'s thousands of tarball-host resolutions)
    /// would otherwise leak one host UDP fd + mio registration per lookup until
    /// the host runs out of fds — after which `bind_dual_stack_udp` fails and
    /// every subsequent resolution returns `EAI_AGAIN` for the VM's lifetime.
    udp_sockets: Mutex<HashMap<(u64, u32), Arc<UdpSocket>>>,
    /// (cid, peer_port) → last guest→host send time, for LRU eviction of idle
    /// UDP flows once `udp_sockets` reaches `UDP_FLOW_CAP`.
    udp_last_used: Mutex<HashMap<(u64, u32), std::time::Instant>>,
    /// (cid, host_src_port) → ProxyConnState. The TCP stream
    /// lives in the muxer-io thread; we keep flow-control
    /// accounting (both directions) here so handle_rw can emit
    /// CREDIT_UPDATE packets and the on_data closure (called
    /// from the muxer-io thread) can decrement available credit
    /// and trip the pause flag. The inner Arc<Mutex<...>> is
    /// cloned into the on_data closure on conn establishment.
    inbound_conns: Mutex<HashMap<(u64, u32), Arc<Mutex<ProxyConnState>>>>,
    /// Sender to the single muxer I/O thread. Owns ALL inbound
    /// TcpStreams, replaces per-conn reader+writer threads.
    io_tx: mpsc::Sender<MuxerCmd>,
    /// Waker to nudge the I/O thread after sending a command.
    io_waker: Arc<mio::Waker>,
    /// Set by [`shutdown`](Self::shutdown) to make the I/O thread exit so it can
    /// be joined before the VM frees guest RAM it drains into.
    io_stop: Arc<AtomicBool>,
    /// The I/O thread's join handle (taken by `shutdown`).
    io_join: Mutex<Option<std::thread::JoinHandle<()>>>,
    /// (peer_cid, peer_port) → TsiState. Indexed by the guest CID
    /// (which is always our `cid` for in-guest TSI) and the
    /// guest's chosen `peer_port` from the TSI control DGRAMs.
    tsi: Mutex<HashMap<(u64, u32), TsiState>>,
    /// Outbound RX queue (intents waiting for guest descriptors).
    /// The vsock device drains this on every notify() and pushes
    /// the packets into RX virtq. Anyone (TsiListener accept
    /// thread, future TsiConn pump threads) can push to it.
    pub rxq: Arc<Mutex<MuxerRxQ>>,
    /// Callback the muxer invokes after pushing to rxq, to wake
    /// the vsock device so it drains. Set by the device after
    /// construction (it has the IRQ-raise closure).
    kick_device: Mutex<Option<Arc<dyn Fn() + Send + Sync>>>,
    /// Expected TSI control-channel auth token (32 bytes). When
    /// `Some`, every TSI control message dispatched in
    /// `handle_tsi_control` must be prefixed with these bytes —
    /// mismatches are silently dropped. The kernel patch 0014
    /// produces matching prefixes; guest userspace can't read the
    /// token (it's static module BSS, scrubbed from /proc/cmdline)
    /// so it can't forge authenticated messages.
    ///
    /// `None` means legacy mode — accept all messages. Used by
    /// pre-0.6.0 snapshots that don't have a baked-in token, and
    /// by unit tests that exercise the dispatcher directly.
    pub(crate) tsi_token: Option<[u8; TSI_TOKEN_LEN]>,
}

/// Length of the TSI control-channel auth token in bytes. Matches
/// the kernel-side `TSI_TOKEN_LEN` in `af_tsi.c`. See
/// `kernel-build/patches/af-tsi/0014-*.patch` for the threat model.
pub const TSI_TOKEN_LEN: usize = 32;

/// Constant-time byte-slice compare. Returns true iff both slices
/// have the same length AND every byte is equal. Visits every byte
/// of `b` so the runtime is independent of where the first mismatch
/// is — closing the timing oracle that a naïve `==` would open. We
/// roll our own instead of pulling in `subtle` because it's three
/// lines and we want zero new deps in the auth path.
#[inline]
fn ct_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff: u8 = 0;
    for i in 0..a.len() {
        diff |= a[i] ^ b[i];
    }
    diff == 0
}

impl VsockMuxer {
    pub fn new(cid: u64) -> Result<Self, muxer_thread::StartError> {
        Self::with_tsi_token(cid, None)
    }

    /// Construct a muxer that enforces TSI control-channel auth.
    /// `token` is the 32-byte secret baked into the booting kernel's
    /// `supermachine.tsi_token=<hex>` cmdline arg. Pass `None` to
    /// disable enforcement (legacy snapshots, tests).
    pub fn with_tsi_token(
        cid: u64,
        token: Option<[u8; TSI_TOKEN_LEN]>,
    ) -> Result<Self, muxer_thread::StartError> {
        // Ensure the fd budget clears UDP_FLOW_CAP DNS flows + the worker's
        // other fds, regardless of the inherited soft limit (fixes the
        // build-VM github.com EAI_AGAIN under DNS load on 256-fd parents).
        raise_nofile_limit();
        let (io_tx, io_waker, io_stop, io_join) = muxer_thread::spawn()?;
        Ok(Self {
            cid,
            proxies: Mutex::new(HashMap::new()),
            pending_inbound: Arc::new(Mutex::new(HashMap::new())),
            pending_outbound: Mutex::new(HashMap::new()),
            udp_dst: Mutex::new(HashMap::new()),
            udp_sockets: Mutex::new(HashMap::new()),
            udp_last_used: Mutex::new(HashMap::new()),
            inbound_conns: Mutex::new(HashMap::new()),
            tsi: Mutex::new(HashMap::new()),
            rxq: Arc::new(Mutex::new(MuxerRxQ::new())),
            kick_device: Mutex::new(None),
            io_tx,
            io_waker,
            io_stop,
            io_join: Mutex::new(Some(io_join)),
            tsi_token: token,
        })
    }

    /// Stop the I/O thread and join it. MUST be called before the VM frees guest
    /// RAM: the I/O thread drains inbound packets into the guest's RX descriptors
    /// (`kick` → `try_drain_rx`), so a still-running thread would write freed
    /// memory after munmap. Idempotent; clearing the kick callback first stops
    /// any further guest-RAM drains the moment the thread wakes.
    pub fn shutdown(&self) {
        *self.kick_device.lock().unwrap() = None;
        self.io_stop.store(true, Ordering::SeqCst);
        let _ = self.io_waker.wake();
        if let Some(handle) = self.io_join.lock().unwrap().take() {
            let _ = handle.join();
        }
    }

    pub fn set_kick(&self, kick: Arc<dyn Fn() + Send + Sync>) {
        *self.kick_device.lock().unwrap() = Some(kick);
    }

    /// True when no per-request transport state is still in flight.
    ///
    /// Persistent TSI listeners and cached UDP sockets are intentionally
    /// allowed to remain alive here: the pool can snapshot an idle listening
    /// service, but must not interrupt active TCP streams, queued packets, or
    /// pending outbound connects.
    pub fn is_transport_idle(&self) -> bool {
        self.pending_inbound.lock().unwrap().is_empty()
            && self.pending_outbound.lock().unwrap().is_empty()
            && self.inbound_conns.lock().unwrap().is_empty()
            && self.rxq.lock().unwrap().is_empty()
    }

    /// Drain all per-dispatch state so the muxer is fresh for the
    /// next RESTORE. Pool-worker mode calls this between dispatches.
    /// Concrete steps:
    /// - Drop every TsiState (their TsiListener::Drop closes the
    ///   listener fd + joins the accept thread → no leaked threads).
    /// - Clear all in-flight bookkeeping (proxies, pending_*, udp_*,
    ///   inbound_conns, rxq).
    /// - Send MuxerCmd::Reset to the io thread to drop its tcp/udp
    ///   registrations.
    /// - Keep the io thread + waker + io_tx alive (re-used).
    pub fn reset(&self) {
        // Order matters: drop TSI listeners FIRST so accept threads
        // can't push fresh connections while we're clearing maps.
        self.tsi.lock().unwrap().clear();
        // Reset io thread state and wait for the boundary before
        // draining queues below. Without the ack, a rapid restore can
        // register a fresh stream before the previous async Reset is
        // processed, then have that stale Reset drop the new stream.
        // Also, the io thread can enqueue EOF/SHUTDOWN callbacks while
        // dropping old streams, so rxq must be drained after this ack.
        let (done_tx, done_rx) = mpsc::channel();
        if self.io_tx.send(MuxerCmd::Reset { done: done_tx }).is_ok() {
            let _ = self.io_waker.wake();
            // Wait for the io thread to confirm it processed the Reset
            // (dropped every conn + emitted any closing callbacks)
            // before we drain the rxq below. The ack normally lands in
            // microseconds and `recv_timeout` returns the instant it
            // does, so this costs nothing on the common path; the cap
            // is only a safety valve against a wedged io thread (field
            // report 0.7.32). 500 ms (was 50 ms) makes the ack-received
            // case — where the post-ack drain catches every late
            // callback — the norm even when the io thread is mid-read,
            // closing the window where a stale packet from the prior
            // clone could slip into the rxq after the drain.
            let _ = done_rx.recv_timeout(std::time::Duration::from_millis(500));
        }
        self.proxies.lock().unwrap().clear();
        self.pending_inbound.lock().unwrap().clear();
        self.pending_outbound.lock().unwrap().clear();
        self.udp_dst.lock().unwrap().clear();
        self.udp_sockets.lock().unwrap().clear();
        self.udp_last_used.lock().unwrap().clear();
        self.inbound_conns.lock().unwrap().clear();
        // Drain any stale packets, including callbacks the io thread
        // emitted while closing old streams during Reset.
        self.rxq.lock().unwrap().drain();
    }

    /// Look up the host-side TCP port currently bound for the
    /// listener serving guest `vm_port`. Returns None if no listener
    /// is registered yet (e.g. guest hasn't done TSI_LISTEN).
    ///
    /// Caveat: the guest's TSI driver assigns RANDOM vm_ports per
    /// bind — they don't match the AF_INET port the guest userspace
    /// asked for (e.g. http-echo's bind(8000) becomes vm_port=
    /// 0x72b58b71 or similar). So this is rarely what callers want;
    /// `first_host_port` is usually what you reach for.
    pub fn host_port_for_vm_port(&self, vm_port: u32) -> Option<u16> {
        let s = self.tsi.lock().unwrap();
        s.values().find_map(|st| {
            if st.vm_port == Some(vm_port) {
                st.listener.as_ref().map(|l| l.host_addr.port())
            } else {
                None
            }
        })
    }

    /// Return the host TCP port of the first TSI listener registered.
    /// Used by the TLS terminator: typical guest workloads have one
    /// HTTP server, so "the first listener" is unambiguous. Returns
    /// None if no listener has been bound yet.
    pub fn first_host_port(&self) -> Option<u16> {
        let s = self.tsi.lock().unwrap();
        // Prefer the AF_INET TSI listener if one exists. Many
        // production images bind both 0.0.0.0 and ::; selecting
        // the AF_INET listener (when present) keeps the host TCP
        // path entirely V4 end-to-end, which avoids any V4-mapped-
        // V6 acrobatics in tools that introspect the address.
        // `st.family` carries the *guest Linux* ABI value (filled
        // from `tsk->family` in the guest kernel's PROXY_CREATE),
        // so we compare to the Linux constant — `libc::AF_INET6`
        // differs across host platforms (`=30` on macOS) and
        // would mis-classify. AF_TSI6 fallback is fully byte-flow
        // capable since the TSI_GETNAME family fix in 0.7.45.
        s.values()
            .find_map(|st| {
                if st.family == LINUX_AF_INET {
                    st.listener.as_ref().map(|l| l.host_addr.port())
                } else {
                    None
                }
            })
            .or_else(|| {
                s.values()
                    .find_map(|st| st.listener.as_ref().map(|l| l.host_addr.port()))
            })
    }

    /// Resolve a `(host_port → guest TSI listener)` route.
    ///
    /// `want_inet_port`: if `Some`, pin to the TSI listener the guest
    /// bound on that AF_INET port (e.g. Chrome on 9222 vs nginx on
    /// 80). If `None`, fall back to "the first AF_INET TSI listener"
    /// — the legacy behavior used by `vm.connect()`, docker-router,
    /// and any unix-mux client that doesn't send the
    /// `SMUX-PORT-V1` routing header.
    fn first_listener_endpoint(&self, want_inet_port: Option<u16>) -> Option<(u64, u32)> {
        let s = self.tsi.lock().unwrap();
        let endpoint = |cid: u64, st: &TsiState| {
            let listener_vm_port = st.vm_port?;
            if let Some(want) = want_inet_port {
                if st.inet_port != Some(want) {
                    return None;
                }
            }
            Some((cid, listener_vm_port))
        };
        s.iter()
            .find_map(|((cid, _), st)| {
                if st.family == LINUX_AF_INET {
                    endpoint(*cid, st)
                } else {
                    None
                }
            })
            .or_else(|| s.iter().find_map(|((cid, _), st)| endpoint(*cid, st)))
    }

    /// Returns true once a TSI listener bound to AF_INET
    /// `want_inet_port` is registered. `vsock_mux::wait_for_host_port`
    /// uses this to retry while the guest's listener is still coming
    /// up.
    pub fn host_port_for_inet_port(&self, want_inet_port: u16) -> Option<u16> {
        let s = self.tsi.lock().unwrap();
        s.values().find_map(|st| {
            if st.inet_port == Some(want_inet_port) {
                st.listener.as_ref().map(|l| l.host_addr.port())
            } else {
                None
            }
        })
    }

    fn open_stream_to_guest(
        &self,
        stream: MuxerStream,
        want_inet_port: Option<u32>,
    ) -> std::io::Result<()> {
        // The plumbing argument is `Option<u32>` for historical reasons
        // (legacy callers expected the guest *vsock* port). Today it
        // carries the *AF_INET* user port written by
        // `Vm::expose_tcp`'s `SMUX-PORT-V1` header; anything above
        // u16::MAX can't be an AF_INET port, so narrow lossily and
        // fall back to "first AF_INET listener" otherwise.
        let want = want_inet_port.and_then(|p| u16::try_from(p).ok());
        let Some((cid, vm_port)) = self.first_listener_endpoint(want) else {
            return Err(std::io::Error::new(
                std::io::ErrorKind::NotConnected,
                "no TSI listener",
            ));
        };
        let host_src_port = crate::devices::virtio::vsock::tsi_stream::alloc_host_src_port();
        self.pending_inbound.lock().unwrap().insert(
            (cid, host_src_port),
            PendingInbound {
                cid,
                host_src_port,
                vm_port,
                stream: Some(stream),
            },
        );
        let req = RxPacket {
            hdr: Header {
                src_cid: VSOCK_HOST_CID,
                dst_cid: cid,
                src_port: host_src_port,
                dst_port: vm_port,
                len: 0,
                type_: VSOCK_TYPE_STREAM,
                op: VSOCK_OP_REQUEST,
                flags: 0,
                buf_alloc: OUR_BUF_ALLOC,
                fwd_cnt: 0,
            },
            data: Vec::new(),
        };
        let kick = self.kick_device.lock().unwrap().clone();
        // Connection-setup REQUEST (host→guest). Pushed through the
        // control reserve so it survives a saturated data queue — a setup
        // REQUEST has no ProxyConnState stash to back-pressure into, and
        // the guest can't re-drive a connection it never saw.
        push_control_and_kick(&self.rxq, &kick, req);
        Ok(())
    }

    pub fn open_unix_to_guest(
        &self,
        unix: UnixStream,
        vm_port: Option<u32>,
    ) -> std::io::Result<()> {
        self.open_stream_to_guest(MuxerStream::Unix(unix), vm_port)
    }

    /// Host→guest *native* AF_VSOCK connect. Used by the
    /// `<vsock_mux>-exec.sock` frontend to reach the in-guest exec
    /// agent without going through TSI's TCP-emulation path.
    ///
    /// Differences vs [`open_unix_to_guest`] / [`open_tcp_to_guest`]:
    ///
    /// - No TSI listener registry lookup. The destination is just
    ///   `(self.cid, guest_port)` — caller-specified guest port,
    ///   muxer-owned guest CID.
    /// - The guest kernel routes the resulting `OP_REQUEST` to
    ///   whichever vsock socket is bound to `guest_port` (the
    ///   `socket(AF_VSOCK, ...) + bind(...) + listen()` path on the
    ///   guest), independent of TSI.
    /// - `handle_response` matches by `(cid, host_src_port)` from
    ///   `pending_inbound` so the byte bridge is identical to TSI.
    ///
    /// Return value: Ok if the request was queued. The actual
    /// connect outcome is observed on `OP_RESPONSE` (success) or
    /// `OP_RST` (no listener / kernel refused) over the muxer's
    /// rxq. Callers that need a "did it connect" handshake on the
    /// stream should expect to see EOF immediately if the guest
    /// rejected the request.
    pub fn open_native_to_guest(
        &self,
        stream: MuxerStream,
        guest_port: u32,
    ) -> std::io::Result<()> {
        let cid = self.cid;
        let host_src_port = crate::devices::virtio::vsock::tsi_stream::alloc_host_src_port();
        self.pending_inbound.lock().unwrap().insert(
            (cid, host_src_port),
            PendingInbound {
                cid,
                host_src_port,
                vm_port: guest_port,
                stream: Some(stream),
            },
        );
        let req = RxPacket {
            hdr: Header {
                src_cid: VSOCK_HOST_CID,
                dst_cid: cid,
                src_port: host_src_port,
                dst_port: guest_port,
                len: 0,
                type_: VSOCK_TYPE_STREAM,
                op: VSOCK_OP_REQUEST,
                flags: 0,
                buf_alloc: OUR_BUF_ALLOC,
                fwd_cnt: 0,
            },
            data: Vec::new(),
        };
        let kick = self.kick_device.lock().unwrap().clone();
        // Connection-setup REQUEST (host→guest). Pushed through the
        // control reserve so it survives a saturated data queue — a setup
        // REQUEST has no ProxyConnState stash to back-pressure into, and
        // the guest can't re-drive a connection it never saw.
        push_control_and_kick(&self.rxq, &kick, req);
        Ok(())
    }

    pub fn open_tcp_to_guest(&self, tcp: TcpStream, vm_port: Option<u32>) -> std::io::Result<()> {
        self.open_stream_to_guest(MuxerStream::Tcp(tcp), vm_port)
    }

    /// Same as `open_tcp_to_guest` but injects `prefix` bytes into the
    /// guest as if they had been read from the TCP socket first. Used
    /// for SCM_RIGHTS handoff, where the router has already consumed
    /// some bytes from the client's TCP buffer to make a routing
    /// decision and now needs the guest to see them.
    pub fn open_tcp_to_guest_with_prefix(
        &self,
        tcp: TcpStream,
        prefix: Vec<u8>,
        vm_port: Option<u32>,
    ) -> std::io::Result<()> {
        self.open_stream_to_guest(MuxerStream::TcpWithPrefix(tcp, prefix), vm_port)
    }

    /// Snapshot just the TSI listeners (the only host-side state worth
    /// surviving restore). In-flight TCP/UDP conns and pending hand-
    /// shakes are intentionally dropped — peer TCPs see RST, retried
    /// requests come back.
    pub fn capture_tsi_listeners(&self) -> Vec<TsiListenerSnapshot> {
        let s = self.tsi.lock().unwrap();
        s.iter()
            .filter_map(|((cid, peer_port), st)| {
                // Only listeners — entries past TSI_LISTEN have vm_port set.
                st.vm_port.map(|vm_port| TsiListenerSnapshot {
                    cid: *cid,
                    peer_port: *peer_port,
                    vm_port,
                    family: st.family,
                    socktype: st.socktype,
                    inet_port: st.inet_port,
                })
            })
            .collect()
    }

    pub fn listener_count(&self) -> usize {
        let s = self.tsi.lock().unwrap();
        s.values().filter(|st| st.vm_port.is_some()).count()
    }

    /// Re-bind every captured TSI listener: spawn a fresh accept
    /// thread on a new ephemeral host port and re-insert the
    /// TsiState. The guest only knows its `vm_port`; the host port
    /// changes across restore (transparent — guest never sees it).
    pub fn restore_tsi_listeners(self: &Arc<Self>, snaps: &[TsiListenerSnapshot]) {
        for s in snaps {
            // Build the on_accept closure exactly as TSI_LISTEN does.
            let pending = self.pending_inbound.clone();
            let rxq = self.rxq.clone();
            let kick = self.kick_device.lock().unwrap().clone();
            let on_accept: Arc<dyn Fn(u64, u32, u32, std::net::TcpStream) + Send + Sync> =
                Arc::new(move |cid, host_src_port, vm_port, tcp| {
                    pending.lock().unwrap().insert(
                        (cid, host_src_port),
                        PendingInbound {
                            cid,
                            host_src_port,
                            vm_port,
                            stream: Some(MuxerStream::Tcp(tcp)),
                        },
                    );
                    let req = RxPacket {
                        hdr: Header {
                            src_cid: VSOCK_HOST_CID,
                            dst_cid: cid,
                            src_port: host_src_port,
                            dst_port: vm_port,
                            len: 0,
                            type_: VSOCK_TYPE_STREAM,
                            op: VSOCK_OP_REQUEST,
                            flags: 0,
                            buf_alloc: OUR_BUF_ALLOC,
                            fwd_cnt: 0,
                        },
                        data: Vec::new(),
                    };
                    // Setup REQUEST — pushed through the control reserve so
                    // it survives a saturated data queue.
                    push_control_and_kick(&rxq, &kick, req);
                });
            match TsiListener::bind(s.cid, s.vm_port, on_accept) {
                Ok(listener) => {
                    if vsock_trace_enabled() {
                        eprintln!(
                            "[muxer] restored TSI listener cid={} vm_port={} -> host {}",
                            s.cid, s.vm_port, listener.host_addr
                        );
                    }
                    self.tsi.lock().unwrap().insert(
                        (s.cid, s.peer_port),
                        TsiState {
                            family: s.family,
                            socktype: s.socktype,
                            vm_port: Some(s.vm_port),
                            inet_port: s.inet_port,
                            listener: Some(listener),
                        },
                    );
                }
                Err(e) => {
                    eprintln!(
                        "[muxer] restore listener cid={} vm_port={} ERR: {e}",
                        s.cid, s.vm_port
                    );
                }
            }
        }
    }

    fn pending_inbound_arc(&self) -> Arc<Mutex<HashMap<(u64, u32), PendingInbound>>> {
        self.pending_inbound.clone()
    }

    /// Push a single packet to the muxer RX queue + wake the device.
    /// Routed through the control reserve (not the back-pressured data
    /// path): this is a generic one-shot submit with no ProxyConnState to
    /// stash into, so reserve-backed delivery is the safe default.
    pub fn submit(&self, pkt: RxPacket) {
        let kick = self.kick_device.lock().unwrap().clone();
        push_control_and_kick(&self.rxq, &kick, pkt);
    }

    /// Process one TX packet from the guest. Returns RxPackets to
    /// be queued back. This is the muxer's main entry point;
    /// vsock::device calls it for every TX-queue head.
    pub fn handle_tx(&self, hdr: &Header, payload: &[u8]) -> Vec<RxPacket> {
        // TSI control DGRAMs: dst_port in [1024, 1031], type=DGRAM.
        if hdr.type_ == VSOCK_TYPE_DGRAM
            && hdr.dst_port >= TSI_PROXY_CREATE
            && hdr.dst_port <= TSI_PROXY_RELEASE
        {
            return self.handle_tsi_control(hdr, payload);
        }
        if hdr.type_ == VSOCK_TYPE_DGRAM && hdr.op == VSOCK_OP_RW {
            // Connected UDP sockets send payloads as regular vsock
            // DGRAMs to their per-socket peer port after TSI_CONNECT.
            // sendto(2) with an explicit address still uses the
            // TSI_SENDTO_DATA control port handled below.
            self.send_udp_payload(hdr.src_cid, hdr.dst_port, hdr.src_port, payload);
            return Vec::new();
        }

        // Stream connection ops.
        match hdr.op {
            VSOCK_OP_REQUEST => self.handle_request(hdr),
            VSOCK_OP_RESPONSE => self.handle_response(hdr),
            VSOCK_OP_RW => self.handle_rw(hdr, payload),
            VSOCK_OP_SHUTDOWN | VSOCK_OP_RST => self.handle_close(hdr),
            VSOCK_OP_CREDIT_UPDATE | VSOCK_OP_CREDIT_REQUEST => self.handle_credit(hdr),
            _ => Vec::new(),
        }
    }

    fn handle_request(&self, hdr: &Header) -> Vec<RxPacket> {
        // Env service: guest opens AF_VSOCK to (HOST, VSOCK_ENV_PORT)
        // and reads JSON. Reply RESPONSE + RW(json) + SHUTDOWN. No
        // muxer state lives past the request.
        if hdr.dst_port == VSOCK_ENV_PORT {
            let json = VSOCK_ENV_JSON
                .lock()
                .unwrap()
                .clone()
                .unwrap_or_else(|| r#"{"env":{},"secrets":{}}"#.into());
            let mk = |op: u16, flags: u32, data: Vec<u8>| RxPacket {
                hdr: Header {
                    src_cid: hdr.dst_cid,
                    dst_cid: hdr.src_cid,
                    src_port: hdr.dst_port,
                    dst_port: hdr.src_port,
                    len: data.len() as u32,
                    type_: VSOCK_TYPE_STREAM,
                    op,
                    flags,
                    buf_alloc: OUR_BUF_ALLOC,
                    fwd_cnt: 0,
                },
                data,
            };
            return vec![
                mk(VSOCK_OP_RESPONSE, 0, Vec::new()),
                mk(VSOCK_OP_RW, 0, json.into_bytes()),
                mk(VSOCK_OP_SHUTDOWN, 1u32 | 2u32, Vec::new()),
            ];
        }

        // Outbound CONNECT phase 2: guest opens vsock to
        // (HOST, TSI_PROXY_PORT=620) carrying src_port=peer_port
        // (the same peer_port the guest sent in the prior
        // TSI_CONNECT). Pair it with the parked TcpStream.
        if hdr.dst_port == TSI_PROXY_PORT {
            let key = (hdr.src_cid, hdr.src_port);
            let mut pending = self.pending_outbound.lock().unwrap();
            if let Some(mut po) = pending.remove(&key) {
                let tcp = match po.tcp.take() {
                    Some(t) => t,
                    None => return vec![RxPacket::rst_for(hdr)],
                };
                drop(pending);
                // For outbound, we use guest's peer_port as the
                // host-side port too (so RW packets we send back
                // have src=hdr.dst_port=TSI_PROXY_PORT and the
                // guest's RW to us has dst=our_host_port=peer_port).
                // Track via inbound_conns keyed the same way; the
                // Write/Close routing in handle_rw matches on
                // (cid, hdr.dst_port = our side port).
                let host_src_port = hdr.src_port; // = peer_port
                let guest_dst_port = hdr.src_port;
                let rxq = self.rxq.clone();
                let kick = self.kick_device.lock().unwrap().clone();
                let cid = po.cid;
                let state = Arc::new(Mutex::new(ProxyConnState::new(
                    guest_dst_port,
                    host_src_port,
                )));
                // Seed peer_buf_alloc/peer_fwd_cnt from the
                // REQUEST header — outbound conns start with
                // whatever the guest advertised on its REQUEST.
                {
                    let mut st = state.lock().unwrap();
                    st.peer_buf_alloc = hdr.buf_alloc;
                    st.peer_fwd_cnt = hdr.fwd_cnt;
                }
                let pause_flag = state.lock().unwrap().pause_flag.clone();
                let state_for_cb = state.clone();
                let on_data: Arc<dyn Fn(Vec<u8>) + Send + Sync> = Arc::new(move |data| {
                    let (op, flags) = if data.is_empty() {
                        (VSOCK_OP_SHUTDOWN, 1u32 | 2u32)
                    } else {
                        (VSOCK_OP_RW, 0u32)
                    };
                    let (our_fwd_cnt, _avail_before) = {
                        let mut st = state_for_cb.lock().unwrap();
                        let n = data.len() as u32;
                        st.tx_cnt = st.tx_cnt.wrapping_add(n);
                        let avail = st.available_credit();
                        if !data.is_empty() && avail <= CREDIT_LOW_WATERMARK {
                            st.paused_readable = true;
                            st.pause_flag.store(true, Ordering::Release);
                            if vsock_trace_enabled() {
                                eprintln!(
                                    "[muxer-ondata] pause host_src_port={} tx_cnt={} peer_fwd_cnt={} peer_buf_alloc={} avail={}",
                                    st.host_src_port,
                                    st.tx_cnt,
                                    st.peer_fwd_cnt,
                                    st.peer_buf_alloc,
                                    avail,
                                );
                            }
                        }
                        (st.our_fwd_cnt, avail)
                    };
                    let pkt = RxPacket {
                        hdr: Header {
                            src_cid: VSOCK_HOST_CID,
                            dst_cid: cid,
                            src_port: TSI_PROXY_PORT,
                            dst_port: guest_dst_port,
                            len: data.len() as u32,
                            type_: VSOCK_TYPE_STREAM,
                            op,
                            flags,
                            buf_alloc: OUR_BUF_ALLOC,
                            fwd_cnt: our_fwd_cnt,
                        },
                        data,
                    };
                    if let Some(pkt) = push_rxq_and_kick(&rxq, &kick, pkt) {
                        // Shared rxq at cap — stash + pause instead of
                        // dropping (zero byte loss). handle_rw / handle_credit
                        // re-push the stash and resume once the guest drains.
                        let mut st = state_for_cb.lock().unwrap();
                        st.pending_rx.push_back(pkt);
                        st.paused_readable = true;
                        st.pause_flag.store(true, Ordering::Release);
                    }
                });
                // Outbound conns: the guest expects packets from
                // TSI_PROXY_PORT (see the on_data header above).
                let on_drained = self.make_on_drained(cid, TSI_PROXY_PORT, state.clone());
                let _ = self.io_tx.send(MuxerCmd::Register {
                    host_src_port,
                    stream: MuxerStream::Tcp(tcp),
                    on_data,
                    pause_flag,
                    on_drained,
                });
                let _ = self.io_waker.wake();
                self.inbound_conns
                    .lock()
                    .unwrap()
                    .insert((cid, host_src_port), state);
                if vsock_trace_enabled() {
                    eprintln!(
                        "[muxer] outbound conn established cid={cid} peer_port={host_src_port}"
                    );
                }
                // Send RESPONSE to acknowledge the vsock REQUEST.
                return vec![RxPacket {
                    hdr: Header {
                        src_cid: VSOCK_HOST_CID,
                        dst_cid: cid,
                        src_port: TSI_PROXY_PORT,
                        dst_port: hdr.src_port,
                        len: 0,
                        type_: VSOCK_TYPE_STREAM,
                        op: VSOCK_OP_RESPONSE,
                        flags: 0,
                        buf_alloc: OUR_BUF_ALLOC,
                        fwd_cnt: 0,
                    },
                    data: Vec::new(),
                }];
            }
            return vec![RxPacket::rst_for(hdr)];
        }
        // No matching listener for any other dst_port → RST.
        vec![RxPacket::rst_for(hdr)]
    }

    fn handle_response(&self, hdr: &Header) -> Vec<RxPacket> {
        // Guest accepted our host-initiated REQUEST. Match by
        // (cid, our_host_src_port = hdr.dst_port). Spawn the
        // TsiConn bridge: TCP→on_data closure pushes RW packets
        // to the muxer's rxq + kicks the device.
        let key = (hdr.src_cid, hdr.dst_port);
        let pending = self.pending_inbound.lock().unwrap().remove(&key);
        if let Some(mut pi) = pending {
            let stream = match pi.stream.take() {
                Some(s) => s,
                None => return vec![RxPacket::rst_for(hdr)],
            };
            let rxq = self.rxq.clone();
            let kick = self.kick_device.lock().unwrap().clone();
            let cid = pi.cid;
            let host_src_port = pi.host_src_port;
            // guest_dst_port = hdr.src_port — the guest's accepted
            // socket's source port, used as the dst for our RW.
            let guest_dst_port = hdr.src_port;
            let state = Arc::new(Mutex::new(ProxyConnState::new(
                guest_dst_port,
                host_src_port,
            )));
            // Seed peer credit from the RESPONSE header.
            {
                let mut st = state.lock().unwrap();
                st.peer_buf_alloc = hdr.buf_alloc;
                st.peer_fwd_cnt = hdr.fwd_cnt;
            }
            let pause_flag = state.lock().unwrap().pause_flag.clone();
            let state_for_cb = state.clone();
            // on_data: empty Vec means "host closed, send SHUTDOWN".
            let on_data: Arc<dyn Fn(Vec<u8>) + Send + Sync> = Arc::new(move |data| {
                let (op, flags) = if data.is_empty() {
                    // VSOCK_FLAGS_SHUTDOWN_RCV | VSOCK_FLAGS_SHUTDOWN_SEND
                    (VSOCK_OP_SHUTDOWN, 1u32 | 2u32)
                } else {
                    (VSOCK_OP_RW, 0u32)
                };
                let our_fwd_cnt = {
                    let mut st = state_for_cb.lock().unwrap();
                    let n = data.len() as u32;
                    st.tx_cnt = st.tx_cnt.wrapping_add(n);
                    let avail = st.available_credit();
                    if !data.is_empty() && avail <= CREDIT_LOW_WATERMARK {
                        st.paused_readable = true;
                        st.pause_flag.store(true, Ordering::Release);
                        if vsock_trace_enabled() {
                            eprintln!(
                                "[muxer-ondata] pause host_src_port={} tx_cnt={} peer_fwd_cnt={} peer_buf_alloc={} avail={}",
                                st.host_src_port,
                                st.tx_cnt,
                                st.peer_fwd_cnt,
                                st.peer_buf_alloc,
                                avail,
                            );
                        }
                    }
                    st.our_fwd_cnt
                };
                let pkt = RxPacket {
                    hdr: Header {
                        src_cid: VSOCK_HOST_CID,
                        dst_cid: cid,
                        src_port: host_src_port,
                        dst_port: guest_dst_port,
                        len: data.len() as u32,
                        type_: VSOCK_TYPE_STREAM,
                        op,
                        flags,
                        buf_alloc: OUR_BUF_ALLOC,
                        fwd_cnt: our_fwd_cnt,
                    },
                    data,
                };
                if let Some(pkt) = push_rxq_and_kick(&rxq, &kick, pkt) {
                    // Shared rxq at cap — stash + pause instead of
                    // dropping (zero byte loss). handle_rw / handle_credit
                    // re-push the stash and resume once the guest drains.
                    let mut st = state_for_cb.lock().unwrap();
                    st.pending_rx.push_back(pkt);
                    st.paused_readable = true;
                    st.pause_flag.store(true, Ordering::Release);
                }
            });
            // Hand the TCP stream to the single muxer-io thread.
            // No per-conn pthread_create — see the single-epoll-
            // thread design in muxer_thread.rs.
            // Inbound/native conns: the guest expects packets from
            // host_src_port (see the on_data header above).
            let on_drained = self.make_on_drained(cid, host_src_port, state.clone());
            let _ = self.io_tx.send(MuxerCmd::Register {
                host_src_port,
                stream,
                on_data,
                pause_flag,
                on_drained,
            });
            let _ = self.io_waker.wake();
            self.inbound_conns.lock().unwrap().insert(key, state);
            if vsock_trace_enabled() {
                eprintln!("[muxer] inbound conn established cid={cid} host_src={host_src_port} guest_dst={guest_dst_port}");
            }
        }
        Vec::new()
    }

    /// Build the delivery-coupled credit callback for a stream conn. The
    /// muxer-io thread invokes it with each byte count ACTUALLY WRITTEN to
    /// the host socket (inline or from the WRITABLE drain). It advances
    /// `our_fwd_cnt` — per the vsock spec, fwd_cnt counts bytes freed from
    /// our receive buffer, and our "receive buffer" for guest→host bytes is
    /// the io thread's pending queue — and emits a CREDIT_UPDATE to the
    /// guest every [`CREDIT_UPDATE_THRESHOLD`] delivered bytes, so a guest
    /// blocked on a full [`OUR_BUF_ALLOC`] window resumes as the host
    /// drains. Granting credit at RECEIPT instead (the pre-#6 behaviour)
    /// let the guest pump an entire multi-GiB stream into host RAM.
    ///
    /// `credit_src_port` is the src the guest expects on packets from this
    /// conn: `TSI_PROXY_PORT` for guest-initiated (outbound) conns,
    /// `host_src_port` for host-initiated (inbound/native) conns.
    fn make_on_drained(
        &self,
        cid: u64,
        credit_src_port: u32,
        state: Arc<Mutex<ProxyConnState>>,
    ) -> Arc<dyn Fn(usize) + Send + Sync> {
        let rxq = self.rxq.clone();
        let kick = self.kick_device.lock().unwrap().clone();
        Arc::new(move |n: usize| {
            let update = {
                let mut st = state.lock().unwrap();
                st.our_fwd_cnt = st.our_fwd_cnt.wrapping_add(n as u32);
                let consumed = st.our_fwd_cnt.wrapping_sub(st.last_credit_fwd_cnt);
                if consumed >= CREDIT_UPDATE_THRESHOLD {
                    st.last_credit_fwd_cnt = st.our_fwd_cnt;
                    Some((st.our_fwd_cnt, st.guest_dst_port))
                } else {
                    None
                }
            };
            if let Some((fwd_cnt, guest_dst_port)) = update {
                // Control reserve: a lost CREDIT_UPDATE can wedge a
                // window-blocked guest forever, so never drop it on a
                // saturated data rxq.
                push_control_and_kick(
                    &rxq,
                    &kick,
                    RxPacket {
                        hdr: Header {
                            src_cid: VSOCK_HOST_CID,
                            dst_cid: cid,
                            src_port: credit_src_port,
                            dst_port: guest_dst_port,
                            len: 0,
                            type_: VSOCK_TYPE_STREAM,
                            op: VSOCK_OP_CREDIT_UPDATE,
                            flags: 0,
                            buf_alloc: OUR_BUF_ALLOC,
                            fwd_cnt,
                        },
                        data: Vec::new(),
                    },
                );
            }
        })
    }

    fn handle_rw(&self, hdr: &Header, payload: &[u8]) -> Vec<RxPacket> {
        // Inbound vs outbound routing:
        //   inbound  RW: dst_port = our_host_src_port (we registered under it)
        //   outbound RW: dst_port = TSI_PROXY_PORT, src_port = peer_port
        //                (we registered under peer_port)
        let conn_key = if hdr.dst_port == TSI_PROXY_PORT {
            (hdr.src_cid, hdr.src_port)
        } else {
            (hdr.src_cid, hdr.dst_port)
        };
        // Look up the state Arc, then drop the outer map lock
        // before we lock the per-conn Mutex — the on_data path
        // also takes that inner lock, and holding both at once
        // (outer then inner) while another thread might take
        // them inner-only would be fine, but keeping locks
        // narrow is cheaper anyway.
        let state_arc = {
            let conns = self.inbound_conns.lock().unwrap();
            conns.get(&conn_key).cloned()
        };
        let Some(state_arc) = state_arc else {
            // Not an inbound conn — try fallback proxy.
            let key = proxy_key(hdr.dst_port, hdr.src_port);
            let mut proxies = self.proxies.lock().unwrap();
            if let Some(p) = proxies.get_mut(&key) {
                return p.handle_packet(hdr, payload);
            }
            return vec![RxPacket::rst_for(hdr)];
        };

        let kick = self.kick_device.lock().unwrap().clone();
        let (needs_resume, host_src_port) = {
            let mut st = state_arc.lock().unwrap();
            // Re-push any RX packets stashed while the shared rxq was
            // full. The guest consuming its RX queue — which is what
            // produced this very packet — frees rxq slots, so there's
            // likely room now. Preserve order; stop at the first that
            // still doesn't fit (the conn stays paused until it clears:
            // observe_peer won't resume while the stash is non-empty).
            // This is the zero-loss other half of the on_data stash.
            while let Some(pkt) = st.pending_rx.pop_front() {
                if let Some(pkt) = push_rxq_and_kick(&self.rxq, &kick, pkt) {
                    st.pending_rx.push_front(pkt);
                    break;
                }
            }
            // EVERY vsock packet carries buf_alloc + fwd_cnt; we
            // refresh peer credit on every RW, not just on
            // dedicated CREDIT_UPDATE messages. This is the
            // critical flow-control fix — without it,
            // peer_fwd_cnt stays at 0 and we infinitely accrete
            // tx_cnt against a wall that never moves.
            //
            // NOTE: `our_fwd_cnt` does NOT advance here. Credit for
            // guest→host bytes is granted on DELIVERY to the host
            // socket — the io thread's `on_drained` callback (see
            // `make_on_drained`) — so a slow host consumer
            // backpressures the guest at OUR_BUF_ALLOC instead of
            // letting it pump an unbounded backlog into host RAM
            // (builder #6).
            let resume = st.observe_peer(hdr.buf_alloc, hdr.fwd_cnt);
            (resume, st.host_src_port)
        };

        if !payload.is_empty() {
            let t0 = Instant::now();
            let _ = self.io_tx.send(MuxerCmd::Write {
                host_src_port: conn_key.1,
                bytes: payload.to_vec(),
            });
            let _ = self.io_waker.wake();
            mux_profile::record(
                Stage::GuestToTcpSend,
                payload.len(),
                t0.elapsed().as_micros() as u64,
            );
        }

        if needs_resume {
            let _ = self.io_tx.send(MuxerCmd::ResumeReadable { host_src_port });
            let _ = self.io_waker.wake();
            if vsock_trace_enabled() {
                eprintln!("[muxer-rw-fc] resume host_src_port={host_src_port} (credit returned)");
            }
        }
        // CREDIT_UPDATE emission moved to the io thread's `on_drained`
        // callback (delivery-coupled credit — see `make_on_drained`).
        Vec::new()
    }

    fn handle_close(&self, hdr: &Header) -> Vec<RxPacket> {
        let key = proxy_key(hdr.dst_port, hdr.src_port);
        self.proxies.lock().unwrap().remove(&key);
        let conn_key = if hdr.dst_port == TSI_PROXY_PORT {
            (hdr.src_cid, hdr.src_port)
        } else {
            (hdr.src_cid, hdr.dst_port)
        };
        if self
            .inbound_conns
            .lock()
            .unwrap()
            .remove(&conn_key)
            .is_some()
        {
            let _ = self.io_tx.send(MuxerCmd::Close {
                host_src_port: conn_key.1,
            });
            let _ = self.io_waker.wake();
        }
        Vec::new()
    }

    fn handle_credit(&self, hdr: &Header) -> Vec<RxPacket> {
        // CREDIT_UPDATE / CREDIT_REQUEST carry the same
        // buf_alloc + fwd_cnt fields every header does. Refresh
        // peer credit on the matching conn and, if we were
        // paused, send ResumeReadable.
        let conn_key = if hdr.dst_port == TSI_PROXY_PORT {
            (hdr.src_cid, hdr.src_port)
        } else {
            (hdr.src_cid, hdr.dst_port)
        };
        let state_arc = {
            let conns = self.inbound_conns.lock().unwrap();
            conns.get(&conn_key).cloned()
        };
        let Some(state_arc) = state_arc else {
            return Vec::new();
        };
        let kick = self.kick_device.lock().unwrap().clone();
        let (resume, host_src_port, our_fwd_cnt, guest_dst_port) = {
            let mut st = state_arc.lock().unwrap();
            // Flush any rxq back-pressure stash before deciding to
            // resume (same as handle_rw) — a CREDIT_UPDATE means the
            // guest just consumed RX, freeing rxq slots. Zero byte loss.
            while let Some(pkt) = st.pending_rx.pop_front() {
                if let Some(pkt) = push_rxq_and_kick(&self.rxq, &kick, pkt) {
                    st.pending_rx.push_front(pkt);
                    break;
                }
            }
            let resume = st.observe_peer(hdr.buf_alloc, hdr.fwd_cnt);
            (resume, st.host_src_port, st.our_fwd_cnt, st.guest_dst_port)
        };
        if resume {
            let _ = self.io_tx.send(MuxerCmd::ResumeReadable { host_src_port });
            let _ = self.io_waker.wake();
            if vsock_trace_enabled() {
                eprintln!("[muxer-credit] resume host_src_port={host_src_port} (credit returned)");
            }
        }
        // On CREDIT_REQUEST the guest is explicitly asking us to
        // tell it our fwd_cnt — reply with a CREDIT_UPDATE so it
        // can compute its own send credit. (We always have room
        // since we forward straight to host TCP.)
        if hdr.op == VSOCK_OP_CREDIT_REQUEST {
            return vec![RxPacket {
                hdr: Header {
                    src_cid: VSOCK_HOST_CID,
                    dst_cid: hdr.src_cid,
                    src_port: hdr.dst_port,
                    dst_port: guest_dst_port,
                    len: 0,
                    type_: VSOCK_TYPE_STREAM,
                    op: VSOCK_OP_CREDIT_UPDATE,
                    flags: 0,
                    buf_alloc: OUR_BUF_ALLOC,
                    fwd_cnt: our_fwd_cnt,
                },
                data: Vec::new(),
            }];
        }
        Vec::new()
    }

    /// TSI control DGRAM dispatcher. Each control op is a separate
    /// DGRAM packet whose dst_port IS the opcode (1024..1031).
    /// Replies are DGRAM RW packets pushed back via the same
    /// (local_port, peer_port) — guest reads the result from a
    /// recvmsg on the AF_TSI control socket.
    fn handle_tsi_control(&self, hdr: &Header, body: &[u8]) -> Vec<RxPacket> {
        // Authentication: if this muxer was constructed with an
        // expected token (i.e. a snapshot baked under 0.6.0+),
        // every control DGRAM must be prefixed with that token.
        // Anything else — short body, wrong prefix — gets silently
        // dropped. We do NOT reply with an error code: an error
        // reply would itself confirm to a probing attacker that
        // they're talking to a real TSI muxer. The legitimate
        // kernel sender never sees mismatches, so the silent drop
        // is invisible to the only consumer that matters.
        //
        // Constant-time compare (see `ct_eq`) closes the obvious
        // timing oracle a naïve byte-by-byte `==` would open.
        //
        // EXEMPTION: TSI_SENDTO_DATA (port 1028) is a DATA packet,
        // not a control opcode. The kernel patch series 0014
        // explicitly documents in its commit message:
        //
        //     "Data path. TSI_SENDTO_DATA (1028) is a payload
        //      packet routed through normal vsock RW, not through
        //      tsi_control_sendmsg. Authenticating bulk data is a
        //      separate decision (perf cost of per-packet HMAC vs.
        //      dropping connection setup) and lands in a follow-up."
        //
        // The kernel does NOT prepend the token on SENDTO_DATA
        // packets. If we apply the token check to them, every
        // outbound UDP datagram from the guest is silently
        // dropped — the symptom is musl getaddrinfo() failing
        // with EAI_AGAIN (manifesting as `apk add` / `wget` /
        // `getent ahosts` all returning "DNS: transient error"
        // even though `nslookup` works because nslookup uses
        // unconnected UDP via TSI_SENDTO_ADDR which IS auth'd).
        //
        // The integrity story for SENDTO_DATA: the guest already
        // has its own kernel-owned (cid, peer_port) state from
        // PROXY_CREATE/TSI_SENDTO_ADDR (both auth'd); a malicious
        // userland process can't fabricate a SENDTO_DATA against a
        // peer_port it didn't legitimately register. So the
        // exemption doesn't widen the attack surface — the
        // entry-point gate already happened on the control op
        // that set up udp_dst.
        let body: &[u8] = if hdr.dst_port == TSI_SENDTO_DATA {
            body
        } else if let Some(expected) = self.tsi_token.as_ref() {
            if body.len() < TSI_TOKEN_LEN || !ct_eq(&body[..TSI_TOKEN_LEN], &expected[..]) {
                return Vec::new();
            }
            &body[TSI_TOKEN_LEN..]
        } else {
            body
        };
        let mk_resp = |result: i32| -> RxPacket {
            // TSI replies ride on RW DGRAMs.
            let bytes = result.to_le_bytes().to_vec();
            RxPacket {
                hdr: Header {
                    src_cid: hdr.dst_cid,
                    dst_cid: hdr.src_cid,
                    src_port: hdr.dst_port,
                    dst_port: hdr.src_port,
                    len: bytes.len() as u32,
                    type_: VSOCK_TYPE_DGRAM,
                    op: VSOCK_OP_RW,
                    flags: 0,
                    buf_alloc: OUR_BUF_ALLOC,
                    fwd_cnt: 0,
                },
                data: bytes,
            }
        };

        match hdr.dst_port {
            TSI_PROXY_CREATE if body.len() >= 8 => {
                // struct { u32 peer_port; u16 family; u16 type; }
                let (Some(peer_port), Some(family), Some(socktype)) = (
                    body_le_u32(body, 0),
                    body_le_u16(body, 4),
                    body_le_u16(body, 6),
                ) else {
                    return Vec::new();
                };
                eprintln!(
                    "[muxer] PROXY_CREATE peer_port={peer_port} family={family} type={socktype}"
                );
                self.tsi.lock().unwrap().insert(
                    (hdr.src_cid, peer_port),
                    TsiState {
                        family,
                        socktype,
                        vm_port: None,
                        inet_port: None,
                        listener: None,
                    },
                );
                // No reply: the AF_TSI control op doesn't emit a
                // recvmsg on the guest side for this op.
                Vec::new()
            }
            TSI_LISTEN if body.len() >= 8 => {
                // struct tsi_listen_req {
                //     u32 svm_port;   // offset 0 (=peer_port slot)
                //     u32 vm_port;    // offset 4 (= guest's *vsock* port)
                //     u32 backlog;    // offset 8
                //     u32 addr_len;   // offset 12
                //     char addr[];    // offset 16 — a sockaddr_in/_in6
                // };
                let (Some(peer_port), Some(vm_port)) = (body_le_u32(body, 0), body_le_u32(body, 4))
                else {
                    return vec![mk_resp(-22)];
                };
                // Extract the userspace AF_INET port from the addr
                // payload so `Vm::expose_tcp(_, guest_port)` can route
                // to it. sockaddr_in: family at offset 0..2 (LE), port
                // at offset 2..4 (BIG-endian, like network order).
                // sockaddr_in6 puts port at the same offset 2..4.
                //
                let inet_port: Option<u16> = if body.len() >= 16 + 4 {
                    let fam_lo = body[16];
                    let fam_hi = body[17];
                    let family_addr = ((fam_hi as u16) << 8) | fam_lo as u16;
                    if family_addr == LINUX_AF_INET || family_addr == LINUX_AF_INET6 {
                        Some(u16::from_be_bytes([body[18], body[19]]))
                    } else {
                        None
                    }
                } else {
                    None
                };
                // Accept-thread closure must hold weak refs to our
                // own state. We can't get Arc<Self> from inside an
                // &self method, so we capture the inner Arc-able
                // pieces directly.
                let pending = self.pending_inbound_arc();
                let rxq = self.rxq.clone();
                let kick = self.kick_device.lock().unwrap().clone();
                let on_accept: Arc<dyn Fn(u64, u32, u32, TcpStream) + Send + Sync> =
                    Arc::new(move |cid, host_src_port, vm_port, tcp| {
                        // Stash the TcpStream until guest sends RESPONSE.
                        pending.lock().unwrap().insert(
                            (cid, host_src_port),
                            PendingInbound {
                                cid,
                                host_src_port,
                                vm_port,
                                stream: Some(MuxerStream::Tcp(tcp)),
                            },
                        );
                        // Push vsock REQUEST to the guest.
                        let req = RxPacket {
                            hdr: Header {
                                src_cid: VSOCK_HOST_CID,
                                dst_cid: cid,
                                src_port: host_src_port,
                                dst_port: vm_port,
                                len: 0,
                                type_: VSOCK_TYPE_STREAM,
                                op: VSOCK_OP_REQUEST,
                                flags: 0,
                                buf_alloc: OUR_BUF_ALLOC,
                                fwd_cnt: 0,
                            },
                            data: Vec::new(),
                        };
                        // Setup REQUEST — pushed through the control reserve
                        // so it survives a saturated data queue.
                        push_control_and_kick(&rxq, &kick, req);
                    });
                let listener = TsiListener::bind(hdr.src_cid, vm_port, on_accept).ok();
                let host = listener.as_ref().map(|l| l.host_addr);
                eprintln!(
                    "[muxer] LISTEN peer_port={peer_port} vm_port={vm_port} inet_port={inet_port:?} -> host {host:?}"
                );
                if let Some(s) = self.tsi.lock().unwrap().get_mut(&(hdr.src_cid, peer_port)) {
                    s.vm_port = Some(vm_port);
                    s.inet_port = inet_port;
                    s.listener = listener;
                }
                vec![mk_resp(if host.is_some() { 0 } else { -22 })]
            }
            TSI_ACCEPT if body.len() >= 4 => {
                let Some(peer_port) = body_le_u32(body, 0) else {
                    return vec![mk_resp(-22)];
                };
                let s = self.tsi.lock().unwrap();
                let r = if s.contains_key(&(hdr.src_cid, peer_port)) {
                    0
                } else {
                    -22
                };
                vec![mk_resp(r)]
            }
            TSI_GETNAME if body.len() >= 12 => {
                // struct tsi_getname_req {
                //   u32 svm_port;       // tsk->svm_port (proxy port for
                //                       //   listener/outbound; listener's
                //                       //   vsock port for accepted child
                //                       //   via vsock-accept inheritance)
                //   u32 svm_peer_port;  // tsk->svm_peer_port
                //                       //   (=TSI_DEFAULT_PORT for outbound
                //                       //   connect; =host_src_port for
                //                       //   accepted child)
                //   u32 peer;           // 0 = getsockname, 1 = getpeername
                // }
                //
                // **The hard contract** is in the guest kernel's
                // `tsi_check_addr_len`: AF_TSI6 (PF_INET6) sockets
                // require sockaddr_in6 (28 B); AF_TSI requires
                // sockaddr_in (16 B). Returning the wrong family makes
                // post-accept getname() fail with -EINVAL, which the
                // kernel's `do_accept` propagates by releasing the
                // freshly-accepted child — TSI_PROXY_RELEASE + vsock
                // SHUTDOWN, no bytes ever flow. This is the
                // "AF_INET6 dual-stack TSI forwarding byte-flow gap":
                // Remix/Express/Node `net.createServer().listen(port)`
                // bind `[::]:port` with V6ONLY=0; host→guest forwards
                // would die at this exact getname step until we made
                // the response family-aware.
                //
                // The cases we handle:
                //   * peer=1 + outbound connect: keyed by tsk->svm_port
                //     (body[0..4]) in pending_outbound — the real
                //     remote we dialed.
                //   * peer=1 + accepted child: keyed by host_src_port
                //     (body[4..8]) in inbound_conns. The host
                //     TcpForwarder binds 127.0.0.1 so the peer is
                //     always 127.0.0.1; we synthesize a port from
                //     the host_src_port token. For AF_INET6 listeners
                //     this gets wrapped to ::ffff:127.0.0.1 below.
                //   * peer=0 on a direct tsi_sock (listener/outbound):
                //     keyed by svm_port in self.tsi → listener.host_addr
                //     or pending_outbound's tcp.local_addr().
                //   * peer=0 on an accepted child: not in self.tsi
                //     directly; find the listener via
                //     `st.vm_port == svm_port` (vsock accept
                //     inheritance makes the child's svm_port match the
                //     listener's).
                let own_port = body_le_u32(body, 0);
                let peer_port = body_le_u32(body, 4);
                let peer_flag = body_le_u32(body, 8).unwrap_or(0);

                // Resolve the calling socket's family. AF_TSI6's
                // tsi_check_addr_len enforces 28-byte addrs;
                // returning a 16-byte sockaddr_in tears down accepts.
                let af: Option<u16> = own_port.and_then(|sp| {
                    let tsi = self.tsi.lock().unwrap();
                    // Direct tsi_sock (listener / outbound connect).
                    if let Some(st) = tsi.get(&(hdr.src_cid, sp)) {
                        return Some(st.family);
                    }
                    // Accepted child: inherits the listener's vsock
                    // port via vsock-accept; find the listener.
                    tsi.values().find_map(|st| {
                        if st.vm_port == Some(sp) {
                            Some(st.family)
                        } else {
                            None
                        }
                    })
                });

                let resolved_addr: Option<SocketAddr> = (|| {
                    let cid = hdr.src_cid;
                    if peer_flag != 0 {
                        // getpeername
                        if let Some(sp) = own_port {
                            if let Some(po) = self.pending_outbound.lock().unwrap().get(&(cid, sp))
                            {
                                if let Some(tcp) = po.tcp.as_ref() {
                                    if let Ok(a) = tcp.peer_addr() {
                                        return Some(a);
                                    }
                                }
                            }
                            if let Some(a) = self.udp_dst.lock().unwrap().get(&(cid, sp)) {
                                return Some(*a);
                            }
                        }
                        // Accepted child: peer is the host TCP client.
                        // The host's TcpForwarder binds 127.0.0.1 only,
                        // so the peer addr is always 127.0.0.1:<port>.
                        // We don't track the actual ephemeral port —
                        // use the host_src_port token as a synthetic
                        // port so logs/audit show a stable distinct
                        // value per connection.
                        if let Some(pp) = peer_port {
                            if self.inbound_conns.lock().unwrap().contains_key(&(cid, pp)) {
                                return Some(SocketAddr::from((
                                    std::net::Ipv4Addr::LOCALHOST,
                                    (pp & 0xffff) as u16,
                                )));
                            }
                        }
                        None
                    } else {
                        // getsockname
                        if let Some(sp) = own_port {
                            // Direct tsi_sock (listener/outbound).
                            let l_host_addr = {
                                let tsi = self.tsi.lock().unwrap();
                                tsi.get(&(cid, sp))
                                    .and_then(|st| st.listener.as_ref().map(|l| l.host_addr))
                            };
                            if let Some(a) = l_host_addr {
                                return Some(a);
                            }
                            if let Some(po) = self.pending_outbound.lock().unwrap().get(&(cid, sp))
                            {
                                if let Some(tcp) = po.tcp.as_ref() {
                                    if let Ok(a) = tcp.local_addr() {
                                        return Some(a);
                                    }
                                }
                            }
                            // Accepted child: local addr = listener's
                            // host_addr (find by vm_port).
                            let child_host_addr = {
                                let tsi = self.tsi.lock().unwrap();
                                tsi.values().find_map(|st| {
                                    if st.vm_port == Some(sp) {
                                        st.listener.as_ref().map(|l| l.host_addr)
                                    } else {
                                        None
                                    }
                                })
                            };
                            if let Some(a) = child_host_addr {
                                return Some(a);
                            }
                        }
                        None
                    }
                })();

                // AF_TSI6 sockets require sockaddr_in6 (28 B). If the
                // resolved address is V4 but the guest socket is V6,
                // wrap as IPv4-mapped IPv6 — what Linux's own
                // dual-stack stack returns for V4 clients connecting
                // to a `[::]:port` listener with V6ONLY=0.
                let want_v6 = af == Some(LINUX_AF_INET6);
                let final_addr = resolved_addr.map(|addr| match (addr, want_v6) {
                    (SocketAddr::V4(v4), true) => {
                        SocketAddr::from((v4.ip().to_ipv6_mapped(), v4.port()))
                    }
                    (other, _) => other,
                });

                let mut buf = vec![0u8; 4 + 4 + 128];
                buf[0..4].copy_from_slice(&0i32.to_le_bytes()); // result=0
                match final_addr {
                    Some(SocketAddr::V4(v4)) => {
                        buf[4..8].copy_from_slice(&16u32.to_le_bytes()); // addr_len=16
                        buf[8..10].copy_from_slice(&LINUX_AF_INET.to_le_bytes()); // AF_INET (guest=2)
                        buf[10..12].copy_from_slice(&v4.port().to_be_bytes()); // port, network order
                        buf[12..16].copy_from_slice(&v4.ip().octets());
                    }
                    Some(SocketAddr::V6(v6)) => {
                        buf[4..8].copy_from_slice(&28u32.to_le_bytes()); // sockaddr_in6 = 28
                        buf[8..10].copy_from_slice(&LINUX_AF_INET6.to_le_bytes()); // AF_INET6 (guest=10)
                        buf[10..12].copy_from_slice(&v6.port().to_be_bytes()); // port
                                                                               // [12..16] flowinfo (zero)
                        buf[16..32].copy_from_slice(&v6.ip().octets()); // addr
                                                                        // [32..36] scope_id (zero)
                    }
                    None if want_v6 => {
                        // V6 fallback: ::1 (loopback IPv6). Mirrors
                        // the V4 fallback shape but in the right
                        // family so AF_TSI6 passes its addr_len check.
                        buf[4..8].copy_from_slice(&28u32.to_le_bytes());
                        buf[8..10].copy_from_slice(&LINUX_AF_INET6.to_le_bytes());
                        buf[16..32].copy_from_slice(&std::net::Ipv6Addr::LOCALHOST.octets());
                    }
                    None => {
                        // Last-resort fallback (pre-0.7.17 behavior):
                        // AF_INET 127.0.0.1:0. Still better than no
                        // response — the guest kernel won't crash and
                        // the caller gets *something*.
                        buf[4..8].copy_from_slice(&16u32.to_le_bytes());
                        buf[8..10].copy_from_slice(&LINUX_AF_INET.to_le_bytes());
                        buf[12..16].copy_from_slice(&[127, 0, 0, 1]);
                    }
                }
                vec![RxPacket {
                    hdr: Header {
                        src_cid: hdr.dst_cid,
                        dst_cid: hdr.src_cid,
                        src_port: hdr.dst_port,
                        dst_port: hdr.src_port,
                        len: buf.len() as u32,
                        type_: VSOCK_TYPE_DGRAM,
                        op: VSOCK_OP_RW,
                        flags: 0,
                        buf_alloc: OUR_BUF_ALLOC,
                        fwd_cnt: 0,
                    },
                    data: buf,
                }]
            }
            TSI_CONNECT if body.len() >= 8 => {
                // struct tsi_connect_req {
                //   u32 peer_port; u32 addr_len; sockaddr addr;
                // }
                // sockaddr_in: u16 family; u16 port_be; u32 addr_be4;
                let (Some(peer_port), Some(addr_len)) =
                    (body_le_u32(body, 0), body_le_u32(body, 4))
                else {
                    return vec![mk_resp(-22)];
                };
                let addr_len = addr_len as usize;
                if body.len() < 8 + addr_len.min(128) || addr_len < 16 {
                    return vec![mk_resp(-22)]; // -EINVAL
                }
                let Some(family) = body_le_u16(body, 8) else {
                    return vec![mk_resp(-22)];
                };
                // Mirror the dual-stack handling we wired into
                // TSI_SENDTO_ADDR: AF_INET6 sockets (Go's
                // `net.Dial("tcp", …)`, Node's connect with
                // `family: 6`, anything with IPV6_V6ONLY=0)
                // address IPv4 hosts via v4-mapped-v6 — unwrap so
                // the egress policy + downstream TCP connect see a
                // real V4 IP. Native IPv6 destinations pass
                // through as V6; the host's `TcpStream::connect`
                // will use a v6 outbound interface if available.
                //
                let target: SocketAddr = if family == LINUX_AF_INET {
                    if addr_len < 8 || body.len() < 8 + 8 {
                        return vec![mk_resp(-22)];
                    }
                    let Some(port) = body_be_u16(body, 10) else {
                        return vec![mk_resp(-22)];
                    };
                    let ip = std::net::Ipv4Addr::new(body[12], body[13], body[14], body[15]);
                    SocketAddr::from((ip, port))
                } else if family == LINUX_AF_INET6 {
                    if addr_len < 28 || body.len() < 8 + 28 {
                        return vec![mk_resp(-22)];
                    }
                    let Some(port) = body_be_u16(body, 10) else {
                        return vec![mk_resp(-22)];
                    };
                    let mut octets = [0u8; 16];
                    octets.copy_from_slice(&body[16..32]);
                    let v6 = std::net::Ipv6Addr::from(octets);
                    if let Some(v4) = v6.to_ipv4_mapped() {
                        SocketAddr::from((v4, port))
                    } else {
                        SocketAddr::from((v6, port))
                    }
                } else {
                    return vec![mk_resp(-97)]; // -EAFNOSUPPORT
                };
                let socktype = self
                    .tsi
                    .lock()
                    .unwrap()
                    .get(&(hdr.src_cid, peer_port))
                    .map(|s| s.socktype);
                if socktype == Some(libc::SOCK_DGRAM as u16) {
                    if let Err(why) = crate::vmm::egress_policy::check_addr(target) {
                        eprintln!("[muxer-udp] CONNECT cid={} peer_port={peer_port} -> {target} BLOCKED: {why}",
                            hdr.src_cid);
                        return vec![mk_resp(-13)]; // -EACCES
                    }
                    self.udp_dst
                        .lock()
                        .unwrap()
                        .insert((hdr.src_cid, peer_port), target);
                    eprintln!(
                        "[muxer-udp] CONNECT cid={} peer_port={peer_port} -> {target} OK",
                        hdr.src_cid
                    );
                    return vec![mk_resp(0)];
                }
                if let Err(why) = crate::vmm::egress_policy::check_addr(target) {
                    eprintln!(
                        "[muxer] CONNECT cid={} peer_port={peer_port} -> {target} BLOCKED: {why}",
                        hdr.src_cid
                    );
                    return vec![mk_resp(-13)]; // -EACCES
                }
                let res = std::net::TcpStream::connect_timeout(
                    &target,
                    std::time::Duration::from_secs(2),
                );
                match res {
                    Ok(tcp) => {
                        let _ = tcp.set_nodelay(true);
                        let _ = tcp.set_nonblocking(true);
                        eprintln!(
                            "[muxer] CONNECT cid={} peer_port={peer_port} -> {target} OK",
                            hdr.src_cid
                        );
                        self.pending_outbound.lock().unwrap().insert(
                            (hdr.src_cid, peer_port),
                            PendingOutbound {
                                cid: hdr.src_cid,
                                peer_port,
                                tcp: Some(tcp),
                            },
                        );
                        vec![mk_resp(0)]
                    }
                    Err(e) => {
                        // Map host errno → guest Linux errno. The
                        // guest expects Linux constants for the
                        // dual-stack address-fallback logic to work
                        // (musl wget, glibc getaddrinfo's
                        // `connect-then-try-next` loop, Go's
                        // happy-eyeballs).
                        //
                        // The subtle case: on a host with no IPv6
                        // route, a V6 destination's connect fails
                        // with macOS ENETUNREACH(51) / EHOSTUNREACH
                        // (65). We re-map those to **EAFNOSUPPORT
                        // (-97)** when the target was V6, *not* to
                        // -ENETUNREACH (-101) or -EHOSTUNREACH
                        // (-113). Reason: musl/busybox-wget only
                        // skips to the next address from the
                        // dual-stack getaddrinfo result when the
                        // first attempt returns EAFNOSUPPORT. Real
                        // network-unreachable / host-unreachable
                        // errors are treated as definitive failures.
                        // Pre-AF_INET6-fix, the muxer rejected V6
                        // SENDTO/CONNECT outright with EAFNOSUPPORT
                        // and clients fell back to V4 transparently;
                        // we have to preserve that behavior when v6
                        // genuinely doesn't work end-to-end.
                        let target_is_v6 = matches!(target, SocketAddr::V6(_));
                        let errno_guest = match e.kind() {
                            std::io::ErrorKind::NetworkUnreachable
                            | std::io::ErrorKind::HostUnreachable
                                if target_is_v6 =>
                            {
                                -97 // EAFNOSUPPORT → trigger v4 fallback
                            }
                            std::io::ErrorKind::NetworkUnreachable => -101,
                            std::io::ErrorKind::HostUnreachable => -113,
                            std::io::ErrorKind::TimedOut => -110,
                            std::io::ErrorKind::ConnectionRefused => -111,
                            _ => match e.raw_os_error() {
                                Some(51) | Some(65) if target_is_v6 => -97,
                                Some(51) => -101,
                                Some(65) => -113,
                                Some(60) => -110,
                                Some(61) => -111,
                                _ => -111,
                            },
                        };
                        eprintln!(
                            "[muxer] CONNECT cid={} peer_port={peer_port} -> {target} ERR {errno_guest}: {e}",
                            hdr.src_cid
                        );
                        vec![mk_resp(errno_guest)]
                    }
                }
            }
            TSI_PROXY_RELEASE => {
                // No-op: vsock accepts inherit the listener's port,
                // so releasing on conn close kills the listener
                // permanently. Listener stays alive for the VM.
                Vec::new()
            }
            TSI_SENDTO_ADDR if body.len() >= 8 => {
                // struct tsi_sendto_addr { u32 peer_port; u32 addr_len;
                //                          sockaddr_in addr; }
                let (Some(peer_port), Some(addr_len)) =
                    (body_le_u32(body, 0), body_le_u32(body, 4))
                else {
                    return Vec::new();
                };
                let addr_len = addr_len as usize;
                if body.len() < 8 + addr_len.min(128) || addr_len < 16 {
                    return Vec::new();
                }
                let Some(family) = body_le_u16(body, 8) else {
                    return Vec::new();
                };
                // Parse the destination sockaddr. AF_INET is the
                // common path. AF_INET6 also has to be handled —
                // Go's `net.ListenUDP("udp", ...)` and Node's
                // `dgram.createSocket('udp')` both create dual-
                // stack AF_INET6 sockets, then send to IPv4 hosts
                // via IPv4-mapped IPv6 (`::ffff:a.b.c.d`). Without
                // this branch the address was silently dropped,
                // `udp_dst` stayed `None`, and TSI_SENDTO_DATA
                // packets were discarded — which broke cloudflared
                // QUIC, Node dgram on Node v18+ udp-default-mode,
                // and anything using the modern "AF_INET6 dual
                // stack" Linux idiom.
                let parsed_addr: Option<SocketAddr> = (|| {
                    if family == LINUX_AF_INET {
                        // sockaddr_in: family (2) + port (2 BE) + ipv4 (4)
                        if addr_len < 8 || body.len() < 8 + 8 {
                            return None;
                        }
                        let port = body_be_u16(body, 10)?;
                        let ip = std::net::Ipv4Addr::new(body[12], body[13], body[14], body[15]);
                        Some(SocketAddr::from((ip, port)))
                    } else if family == LINUX_AF_INET6 {
                        // sockaddr_in6: family (2) + port (2 BE) + flowinfo (4) + addr (16) + scope (4)
                        if addr_len < 28 || body.len() < 8 + 28 {
                            return None;
                        }
                        let port = body_be_u16(body, 10)?;
                        let mut octets = [0u8; 16];
                        octets.copy_from_slice(&body[16..32]);
                        let v6 = std::net::Ipv6Addr::from(octets);
                        // Unwrap IPv4-mapped IPv6 (`::ffff:a.b.c.d`)
                        // to a plain V4 SocketAddr so the egress
                        // policy applies its IPv4 rules. The host
                        // outbound socket is now dual-stack
                        // AF_INET6+IPV6_V6ONLY=0 so it can serve
                        // either family; we re-wrap to v4-mapped-
                        // v6 in `send_udp_payload` if needed.
                        let addr = if let Some(v4) = v6.to_ipv4_mapped() {
                            SocketAddr::from((v4, port))
                        } else {
                            SocketAddr::from((v6, port))
                        };
                        Some(addr)
                    } else {
                        None
                    }
                })();
                if let Some(addr) = parsed_addr {
                    if let Err(why) = crate::vmm::egress_policy::check_addr(addr) {
                        eprintln!(
                            "[muxer] SENDTO_ADDR cid={} -> {addr} BLOCKED: {why}",
                            hdr.src_cid
                        );
                    } else {
                        self.udp_dst
                            .lock()
                            .unwrap()
                            .insert((hdr.src_cid, peer_port), addr);
                    }
                }
                Vec::new()
            }
            TSI_SENDTO_DATA => {
                // The DGRAM socket sends data here directly; peer_port
                // is in hdr.src_port (the AF_INET socket's local TSI
                // port). Body is just the datagram payload.
                let peer_port = hdr.src_port;
                self.send_udp_payload(hdr.src_cid, peer_port, hdr.src_port, body);
                Vec::new()
            }
            _ => Vec::new(),
        }
    }

    fn send_udp_payload(&self, cid: u64, peer_port: u32, guest_dst_port: u32, data: &[u8]) {
        if data.is_empty() {
            return;
        }
        let key = (cid, peer_port);
        let dst = match self.udp_dst.lock().unwrap().get(&key).cloned() {
            Some(a) => a,
            None => {
                if vsock_trace_enabled() {
                    eprintln!(
                        "[muxer-udp] DROP cid={cid} peer_port={peer_port} guest_dst={guest_dst_port} len={} (no udp_dst)",
                        data.len()
                    );
                }
                return;
            }
        };
        let sock_arc = {
            let mut s = self.udp_sockets.lock().unwrap();
            if let Some(existing) = s.get(&key) {
                let existing = existing.clone();
                drop(s);
                // Touch LRU so an active flow isn't evicted out from under itself.
                self.udp_last_used
                    .lock()
                    .unwrap()
                    .insert(key, std::time::Instant::now());
                existing
            } else {
                // Bound the table: a DNS burst (one ephemeral source port per
                // `getaddrinfo`) would otherwise leak a host UDP fd + mio
                // registration per lookup until fd exhaustion → permanent
                // EAI_AGAIN. Evict the least-recently-used flow when at capacity.
                if s.len() >= UDP_FLOW_CAP {
                    let victim = {
                        let lu = self.udp_last_used.lock().unwrap();
                        s.keys()
                            .min_by_key(|k| {
                                lu.get(*k)
                                    .copied()
                                    // No timestamp ⇒ treat as oldest (evict first).
                                    .unwrap_or_else(|| {
                                        std::time::Instant::now()
                                            - std::time::Duration::from_secs(3600)
                                    })
                            })
                            .copied()
                    };
                    if let Some(vk) = victim {
                        s.remove(&vk);
                        self.udp_dst.lock().unwrap().remove(&vk);
                        self.udp_last_used.lock().unwrap().remove(&vk);
                        // Tear down the host UDP fd + mio registration on the
                        // io thread (vk.1 == peer_port, the io-thread key).
                        let _ = self.io_tx.send(MuxerCmd::DeregisterUdp { key: vk.1 });
                        let _ = self.io_waker.wake();
                    }
                }
                let udp = match bind_dual_stack_udp() {
                    Ok(u) => u,
                    Err(e) => {
                        eprintln!("[muxer] UDP bind cid={cid} peer_port={peer_port}: {e}");
                        return;
                    }
                };
                let recv_fd = match udp.try_clone() {
                    Ok(fd) => fd,
                    Err(e) => {
                        eprintln!("[muxer] UDP clone cid={cid} peer_port={peer_port}: {e}");
                        return;
                    }
                };
                let arc = Arc::new(udp);
                s.insert(key, arc.clone());
                self.udp_last_used
                    .lock()
                    .unwrap()
                    .insert(key, std::time::Instant::now());

                let rxq = self.rxq.clone();
                let kick = self.kick_device.lock().unwrap().clone();
                // DNS flow on a v6-less host: suppress AAAA answers so
                // non-happy-eyeballs guests (alpine apk, busybox wget —
                // musl ignores the gai.conf the guest-side mitigation
                // writes) fall through to the A records they can
                // actually connect to. Evaluated once per flow: stub
                // resolvers open a fresh ephemeral port per lookup, so
                // a flow's destination port is stable in practice.
                let strip_aaaa = dst.port() == 53 && !crate::utils::net::host_ipv6_route();
                let on_data: Arc<dyn Fn(Vec<u8>) + Send + Sync> = Arc::new(move |data| {
                    let data = if strip_aaaa {
                        super::dns_filter::rewrite_aaaa_response_to_nodata(&data).unwrap_or(data)
                    } else {
                        data
                    };
                    let pkt = RxPacket {
                        hdr: Header {
                            src_cid: VSOCK_HOST_CID,
                            dst_cid: cid,
                            src_port: peer_port,
                            dst_port: guest_dst_port,
                            len: data.len() as u32,
                            type_: VSOCK_TYPE_DGRAM,
                            op: VSOCK_OP_RW,
                            flags: 0,
                            buf_alloc: OUR_BUF_ALLOC,
                            fwd_cnt: 0,
                        },
                        data,
                    };
                    // DGRAM is connectionless: a full rxq means this
                    // datagram is dropped (expected best-effort loss).
                    let _ = push_rxq_and_kick(&rxq, &kick, pkt);
                });
                let _ = self.io_tx.send(MuxerCmd::RegisterUdp {
                    key: peer_port,
                    udp: recv_fd,
                    on_data,
                });
                let _ = self.io_waker.wake();
                arc
            }
        };
        if vsock_trace_enabled() {
            eprintln!(
                "[muxer] UDP DATA cid={cid} peer_port={peer_port} guest_dst={guest_dst_port} -> {dst} {}B",
                data.len()
            );
        }
        // The host socket is dual-stack AF_INET6 (IPV6_V6ONLY=0).
        // macOS's `sendto` insists the destination sockaddr is
        // AF_INET6 for an AF_INET6 socket — passing a sockaddr_in
        // (V4 dst) returns EINVAL. Wrap V4 destinations as v4-
        // mapped v6 (`::ffff:a.b.c.d`) so the kernel handles the
        // routing back to v4 on the wire transparently.
        let send_dst = match dst {
            SocketAddr::V4(v4) => {
                SocketAddr::new(std::net::IpAddr::V6(v4.ip().to_ipv6_mapped()), v4.port())
            }
            SocketAddr::V6(_) => dst,
        };
        let _ = sock_arc.send_to(data, send_dst);
    }
}

/// Enqueue `pkt` for delivery to the guest and kick the device on the
/// empty → non-empty transition. Returns `None` on success, or
/// `Some(pkt)` — handing the packet back, NOT dropping it — when the
/// shared RX queue is at its cap.
///
/// A `Some` return is back-pressure, not loss. A STREAM caller MUST
/// stash the returned packet in `ProxyConnState::pending_rx` and pause
/// the connection's reads; handle_rw / handle_credit re-push it and
/// resume once the guest drains the queue (zero byte loss). DGRAM
/// callers may drop it: datagram loss is expected and best-effort.
/// Control callers (connection REQUEST / RST / setup) must instead use
/// [`push_control_and_kick`], which draws from the rxq control reserve
/// so a host-initiated setup/teardown frame isn't lost on a saturated
/// data queue (the guest can't re-drive a connection it never saw).
#[must_use = "a returned packet means the RX queue is full; stash it (back-pressure) or explicitly drop it"]
fn push_rxq_and_kick(
    rxq: &Arc<Mutex<MuxerRxQ>>,
    kick: &Option<Arc<dyn Fn() + Send + Sync>>,
    pkt: RxPacket,
) -> Option<RxPacket> {
    let bytes = pkt.data.len();
    let t0 = Instant::now();
    // Coalesce kicks: only raise the SPI on the empty → non-empty
    // transition. A follow-up push during the in-flight drain is
    // already visible to the vCPU RX loop. Saves one `hv_gic_set_spi`
    // syscall (~3.6% of muxer-io non-syscall CPU under load) per
    // coalesced packet during bursts. See
    // docs/design/concurrency-floor-2026-05-04.md.
    let was_empty = match rxq.lock().unwrap().push_was_empty(pkt) {
        Ok(was_empty) => was_empty,
        Err(pkt) => {
            // Queue at cap — hand the packet back so the caller can
            // back-pressure (stash + pause) instead of dropping it.
            mux_profile::record(Stage::RxqKick, bytes, t0.elapsed().as_micros() as u64);
            return Some(pkt);
        }
    };
    if was_empty {
        if let Some(k) = kick.as_ref() {
            k();
        }
    }
    mux_profile::record(Stage::RxqKick, bytes, t0.elapsed().as_micros() as u64);
    None
}

/// Push a control packet (connection REQUEST / RST / setup) and wake the
/// device. Unlike [`push_rxq_and_kick`] there is nothing to hand back:
/// a host-initiated control frame has no `ProxyConnState` stash, and the
/// guest can't re-drive a connection it never saw. So this draws from the
/// rxq's dedicated control reserve (above the data cap) and only drops —
/// loudly — if even that reserve is exhausted, which means the host is
/// badly oversubscribed.
fn push_control_and_kick(
    rxq: &Arc<Mutex<MuxerRxQ>>,
    kick: &Option<Arc<dyn Fn() + Send + Sync>>,
    pkt: RxPacket,
) {
    let bytes = pkt.data.len();
    let t0 = Instant::now();
    // Take the queue lock, push, and RELEASE it BEFORE kicking. The kick
    // closure re-enters the device's RX path, which locks the rxq to
    // drain — so holding the lock across `k()` self-deadlocks (std Mutex
    // is non-reentrant). Binding the result to a local drops the guard at
    // this `;`, before the match, exactly as `push_rxq_and_kick` does via
    // its `let was_empty = match …;`. Calling `k()` inside
    // `match rxq.lock()…{ … }` instead would extend the guard's lifetime
    // across the arms — the bug this comment guards against.
    let push_result = rxq.lock().unwrap().push_control(pkt);
    match push_result {
        Some(was_empty) => {
            // Same kick-coalescing as the data path: raise the SPI only on
            // the empty → non-empty transition.
            if was_empty {
                if let Some(k) = kick.as_ref() {
                    k();
                }
            }
        }
        None => {
            eprintln!(
                "[vsock] muxer RX queue and its {MUXER_RXQ_CONTROL_RESERVE}-slot control \
                 reserve are both full — dropped a connection setup/teardown frame; a \
                 connection may fail to open or close (host badly oversubscribed)"
            );
        }
    }
    mux_profile::record(Stage::RxqKick, bytes, t0.elapsed().as_micros() as u64);
}

// Use HOST_CID so the unused-import warning goes away.
const _: u64 = VSOCK_HOST_CID;

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn target_nofile_policy() {
        // Raise a stingy inherited limit toward the target…
        assert_eq!(target_nofile(256, 1 << 20), TARGET_NOFILE);
        // …but never above the hard cap…
        assert_eq!(target_nofile(256, 4096), 4096);
        assert_eq!(target_nofile(256, 1024), 1024);
        // …and never LOWER an already-generous soft limit.
        assert_eq!(target_nofile(100_000, 1 << 20), 100_000);
        assert_eq!(target_nofile(TARGET_NOFILE, 1 << 20), TARGET_NOFILE);
    }

    #[test]
    // See raise_nofile_limit: rlim_t is platform-defined; the widening
    // cast is deliberate even where it's a no-op.
    #[allow(clippy::unnecessary_cast)]
    fn raise_nofile_limit_clears_the_udp_flow_cap() {
        // After the raise, the soft fd limit must exceed UDP_FLOW_CAP with
        // headroom (this is what keeps a DNS-heavy build from EAI_AGAIN on a
        // 256-fd parent). Only ever raises, so it's safe to run in-process.
        raise_nofile_limit();
        // SAFETY: read-only getrlimit into a zeroed rlimit.
        let soft = unsafe {
            let mut rl: libc::rlimit = std::mem::zeroed();
            assert_eq!(libc::getrlimit(libc::RLIMIT_NOFILE, &mut rl), 0);
            rl.rlim_cur as u64
        };
        assert!(
            soft >= UDP_FLOW_CAP as u64 + 64,
            "soft fd limit {soft} doesn't clear UDP_FLOW_CAP {UDP_FLOW_CAP} + headroom"
        );
    }

    /// The UDP flow table must stay bounded under a burst of distinct
    /// ephemeral source ports (the DNS-lookup-per-`getaddrinfo` pattern that
    /// used to leak a host UDP fd + mio registration each, exhausting fds →
    /// permanent `EAI_AGAIN`). LRU eviction keeps the table at `UDP_FLOW_CAP`,
    /// retaining the most-recent flows and dropping the oldest.
    #[test]
    fn udp_flow_table_capped_with_lru_eviction() {
        // Binding > cap sockets transiently (muxer side + io-thread recv clones)
        // needs headroom over macOS's default 256-fd soft limit.
        unsafe {
            let mut rl: libc::rlimit = std::mem::zeroed();
            if libc::getrlimit(libc::RLIMIT_NOFILE, &mut rl) == 0 {
                rl.rlim_cur = rl.rlim_max.min(4096);
                let _ = libc::setrlimit(libc::RLIMIT_NOFILE, &rl);
            }
        }
        let muxer = VsockMuxer::new(3).expect("muxer");
        let dst: std::net::SocketAddr = "127.0.0.1:9".parse().unwrap();
        let base = 10_000u32;
        let total = UDP_FLOW_CAP as u32 + 32;
        for i in 0..total {
            let port = base + i;
            // CONNECT normally sets udp_dst; do it directly so the send isn't
            // dropped as "no udp_dst".
            muxer.udp_dst.lock().unwrap().insert((3, port), dst);
            muxer.send_udp_payload(3, port, 53, b"q");
        }
        let sockets = muxer.udp_sockets.lock().unwrap();
        assert!(
            sockets.len() <= UDP_FLOW_CAP,
            "udp flow table {} exceeds cap {UDP_FLOW_CAP}",
            sockets.len()
        );
        // LRU: the most-recent flow survives; the very first was evicted.
        assert!(
            sockets.contains_key(&(3, base + total - 1)),
            "most-recently-used flow must be retained"
        );
        assert!(
            !sockets.contains_key(&(3, base)),
            "oldest flow must have been evicted"
        );
    }

    /// Restore the previous value of an env var on drop, so a test
    /// that overrides `SUPERMACHINE_HOST_IPV6` can't leak its setting
    /// into tests that start later in the same process.
    struct EnvGuard {
        key: &'static str,
        prev: Option<std::ffi::OsString>,
    }
    impl EnvGuard {
        fn set(key: &'static str, val: &str) -> Self {
            let prev = std::env::var_os(key);
            std::env::set_var(key, val);
            EnvGuard { key, prev }
        }
    }
    impl Drop for EnvGuard {
        fn drop(&mut self) {
            match self.prev.take() {
                Some(v) => std::env::set_var(self.key, v),
                None => std::env::remove_var(self.key),
            }
        }
    }

    /// End-to-end wiring of the AAAA→NODATA gate through a REAL UDP
    /// round-trip: guest-side `send_udp_payload` → host socket → fake
    /// DNS server → muxer-io thread → per-flow `on_data` → rxq. The
    /// dns_filter unit tests prove the rewrite; this proves the GATE —
    /// the exact thing the original bug got wrong (a mitigation that
    /// existed but never engaged for the traffic it was written for).
    ///
    /// Three scenarios, serialized in one test because they flip the
    /// `SUPERMACHINE_HOST_IPV6` override:
    ///   1. v6-less host + non-53 flow   → identical bytes pass through
    ///      (the filter must not creep onto non-DNS UDP).
    ///   2. "has v6" host + port-53 flow → AAAA passes through (the
    ///      filter must not strip records v6-capable hosts can use).
    ///   3. v6-less host + port-53 flow  → AAAA response rewritten.
    ///
    /// Scenarios 2-3 need a server ON port 53 (the gate keys on the
    /// destination port): unprivileged on macOS 10.14+, root on the
    /// box; skipped gracefully where the bind is denied or taken.
    #[test]
    fn udp_dns_relay_gate_matrix() {
        use super::super::dns_filter::testutil::{aaaa_rr, dns_msg};

        let aaaa_query = dns_msg(false, 28, 0, 0, 0, &[]);
        let aaaa_response = dns_msg(true, 28, 1, 0, 0, &aaaa_rr());

        // One canned-reply DNS server.
        fn serve_one(sock: std::net::UdpSocket, reply: Vec<u8>) -> std::thread::JoinHandle<()> {
            std::thread::spawn(move || {
                let mut buf = [0u8; 1500];
                sock.set_read_timeout(Some(std::time::Duration::from_secs(10)))
                    .unwrap();
                if let Ok((_, from)) = sock.recv_from(&mut buf) {
                    let _ = sock.send_to(&reply, from);
                }
            })
        }

        // Drive one flow and wait (≤5 s) for the relayed response.
        fn round_trip(
            muxer: &VsockMuxer,
            peer_port: u32,
            dst: SocketAddr,
            query: &[u8],
        ) -> Vec<u8> {
            muxer.udp_dst.lock().unwrap().insert((3, peer_port), dst);
            muxer.send_udp_payload(3, peer_port, 4242, query);
            let deadline = Instant::now() + std::time::Duration::from_secs(5);
            loop {
                if let Some(pkt) = muxer.rxq.lock().unwrap().pop() {
                    assert_eq!(pkt.hdr.src_port, peer_port, "reply routed to its flow");
                    assert_eq!(
                        pkt.hdr.len as usize,
                        pkt.data.len(),
                        "header len must match (possibly rewritten) payload"
                    );
                    return pkt.data;
                }
                assert!(Instant::now() < deadline, "no UDP reply relayed within 5s");
                std::thread::sleep(std::time::Duration::from_millis(5));
            }
        }

        // 1. Non-53 flow on a v6-less host: ephemeral server port.
        {
            let _env = EnvGuard::set("SUPERMACHINE_HOST_IPV6", "0");
            let muxer = VsockMuxer::new(3).expect("muxer");
            let server = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
            let dst = server.local_addr().unwrap();
            let h = serve_one(server, aaaa_response.clone());
            let got = round_trip(&muxer, 41_000, dst, &aaaa_query);
            assert_eq!(
                got, aaaa_response,
                "non-53 flow on a v6-less host must pass AAAA through untouched"
            );
            h.join().unwrap();
        }
        // 2. Port-53 flow, host claims v6: gate must stay open.
        {
            let _env = EnvGuard::set("SUPERMACHINE_HOST_IPV6", "1");
            let muxer = VsockMuxer::new(3).expect("muxer");
            if let Ok(server) = std::net::UdpSocket::bind("127.0.0.1:53") {
                let dst = server.local_addr().unwrap();
                let h = serve_one(server, aaaa_response.clone());
                let got = round_trip(&muxer, 41_001, dst, &aaaa_query);
                assert_eq!(
                    got, aaaa_response,
                    "v6-capable host must receive AAAA records unfiltered"
                );
                h.join().unwrap();
            } else {
                eprintln!("[skip] gate-matrix scenario 2: cannot bind 127.0.0.1:53");
            }
        }
        // 3. Port-53 flow, v6-less host: the rewrite, through the relay.
        {
            let _env = EnvGuard::set("SUPERMACHINE_HOST_IPV6", "0");
            let muxer = VsockMuxer::new(3).expect("muxer");
            let server = match std::net::UdpSocket::bind("127.0.0.1:53") {
                Ok(s) => s,
                Err(e) => {
                    eprintln!("[skip] gate-matrix scenario 3: cannot bind 127.0.0.1:53: {e}");
                    return;
                }
            };
            let dst = server.local_addr().unwrap();
            let h = serve_one(server, aaaa_response.clone());
            let got = round_trip(&muxer, 41_002, dst, &aaaa_query);
            let expected =
                super::super::dns_filter::rewrite_aaaa_response_to_nodata(&aaaa_response)
                    .expect("canned response is rewritable");
            assert_eq!(
                got, expected,
                "port-53 flow on a v6-less host must deliver the NODATA rewrite"
            );
            assert_ne!(got, aaaa_response, "must differ from the raw response");
            h.join().unwrap();
        }
    }

    #[test]
    fn tsi_listener_snapshot_codec_round_trips_fixed_width() {
        for inet in [None, Some(80u16), Some(9222u16)] {
            let l = TsiListenerSnapshot {
                cid: 3,
                peer_port: 4_242_906_079,
                vm_port: 1234,
                family: 2,
                socktype: 1,
                inet_port: inet,
            };
            let mut buf = Vec::new();
            l.write_to(&mut buf).unwrap();
            assert_eq!(
                buf.len(),
                TsiListenerSnapshot::WIRE_LEN,
                "fixed 24-byte record"
            );
            let got = TsiListenerSnapshot::read_from(&mut &buf[..]).unwrap();
            assert_eq!(got.cid, l.cid);
            assert_eq!(got.peer_port, l.peer_port);
            assert_eq!(got.vm_port, l.vm_port);
            assert_eq!(got.family, l.family);
            assert_eq!(got.socktype, l.socktype);
            assert_eq!(got.inet_port, inet, "u32::MAX-sentinel None survives");
        }
    }

    fn control_header(dst_port: u32) -> Header {
        Header {
            src_cid: 3,
            dst_cid: VSOCK_HOST_CID,
            src_port: 40_000,
            dst_port,
            len: 0,
            type_: VSOCK_TYPE_DGRAM,
            op: VSOCK_OP_RW,
            flags: 0,
            buf_alloc: OUR_BUF_ALLOC,
            fwd_cnt: 0,
        }
    }

    fn response_i32(resp: &[RxPacket]) -> Option<i32> {
        let bytes = resp.first()?.data.get(0..4)?.try_into().ok()?;
        Some(i32::from_le_bytes(bytes))
    }

    #[test]
    fn shutdown_stops_io_thread_and_is_idempotent() -> Result<(), Box<dyn std::error::Error>> {
        let muxer = VsockMuxer::new(3)?;
        // Sanity: the I/O thread's join handle is present before shutdown.
        assert!(muxer.io_join.lock().unwrap().is_some());
        // First shutdown sets stop, wakes, and joins. If join deadlocked or the
        // thread ignored `stop`, this test would hang (and CI would time out),
        // so its mere completion proves the thread exits promptly.
        muxer.shutdown();
        assert!(
            muxer.io_stop.load(Ordering::SeqCst),
            "stop flag set after shutdown"
        );
        assert!(
            muxer.io_join.lock().unwrap().is_none(),
            "join handle consumed by shutdown"
        );
        // Idempotent: a second shutdown is a no-op (no double-join panic/hang).
        muxer.shutdown();
        Ok(())
    }

    #[test]
    fn malformed_tsi_connect_returns_einval() -> Result<(), Box<dyn std::error::Error>> {
        let muxer = VsockMuxer::new(3)?;
        let hdr = control_header(TSI_CONNECT);

        let mut body = Vec::new();
        body.extend_from_slice(&1234u32.to_le_bytes());
        body.extend_from_slice(&16u32.to_le_bytes());

        let resp = muxer.handle_tsi_control(&hdr, &body);
        assert_eq!(response_i32(&resp), Some(-22));
        Ok(())
    }

    #[test]
    fn malformed_tsi_sendto_addr_is_ignored() -> Result<(), Box<dyn std::error::Error>> {
        let muxer = VsockMuxer::new(3)?;
        let hdr = control_header(TSI_SENDTO_ADDR);

        let mut body = Vec::new();
        body.extend_from_slice(&1234u32.to_le_bytes());
        body.extend_from_slice(&16u32.to_le_bytes());

        let resp = muxer.handle_tsi_control(&hdr, &body);
        assert!(resp.is_empty());
        Ok(())
    }

    /// Build a body for a `TSI_PROXY_CREATE` op: u32 peer_port +
    /// u16 family + u16 socktype = 8 bytes minimum. Used by the
    /// auth tests below — these don't care about the meaning,
    /// only that a valid body shape gets either dispatched or
    /// dropped depending on the auth prefix.
    fn proxy_create_body(peer_port: u32) -> Vec<u8> {
        let mut b = Vec::new();
        b.extend_from_slice(&peer_port.to_le_bytes());
        // family=AF_INET=2, socktype=SOCK_STREAM=1
        b.extend_from_slice(&2u16.to_le_bytes());
        b.extend_from_slice(&1u16.to_le_bytes());
        b
    }

    #[test]
    fn tsi_control_rejects_missing_token_prefix() -> Result<(), Box<dyn std::error::Error>> {
        // Muxer armed with auth; body has NO prefix — must be
        // silently dropped (no response, no state change).
        let token = [0x11u8; TSI_TOKEN_LEN];
        let muxer = VsockMuxer::with_tsi_token(3, Some(token))?;
        let hdr = control_header(TSI_PROXY_CREATE);
        let body = proxy_create_body(1234);

        let resp = muxer.handle_tsi_control(&hdr, &body);
        assert!(
            resp.is_empty(),
            "unprefixed TSI control DGRAM must be dropped, got {} pkts",
            resp.len()
        );
        // Bookkeeping side-effect: PROXY_CREATE inserts into the
        // `tsi` map. The drop must prevent that insertion.
        assert!(
            muxer.tsi.lock().unwrap().is_empty(),
            "dropped DGRAM still mutated tsi state"
        );
        Ok(())
    }

    #[test]
    fn tsi_control_rejects_wrong_token_prefix() -> Result<(), Box<dyn std::error::Error>> {
        let token = [0x11u8; TSI_TOKEN_LEN];
        let muxer = VsockMuxer::with_tsi_token(3, Some(token))?;
        let hdr = control_header(TSI_PROXY_CREATE);

        // Wrong prefix (all 0x22) followed by the real body.
        let mut body = vec![0x22u8; TSI_TOKEN_LEN];
        body.extend_from_slice(&proxy_create_body(1234));

        let resp = muxer.handle_tsi_control(&hdr, &body);
        assert!(resp.is_empty(), "wrong-token DGRAM must be dropped");
        assert!(muxer.tsi.lock().unwrap().is_empty());
        Ok(())
    }

    #[test]
    fn tsi_control_accepts_correct_token_prefix() -> Result<(), Box<dyn std::error::Error>> {
        let token = [0x33u8; TSI_TOKEN_LEN];
        let muxer = VsockMuxer::with_tsi_token(3, Some(token))?;
        let hdr = control_header(TSI_PROXY_CREATE);

        // Real token prefix + valid body. PROXY_CREATE doesn't
        // emit a response, but it does insert into the `tsi`
        // map — we assert on the side-effect, not the (empty)
        // response.
        let mut body = token.to_vec();
        body.extend_from_slice(&proxy_create_body(1234));

        let resp = muxer.handle_tsi_control(&hdr, &body);
        // PROXY_CREATE legitimately emits no reply (see the
        // existing dispatcher), so an empty resp here is the
        // CORRECT signal that the message was dispatched.
        assert!(resp.is_empty(), "PROXY_CREATE should not reply");
        assert_eq!(
            muxer.tsi.lock().unwrap().len(),
            1,
            "authenticated PROXY_CREATE should have inserted exactly one state"
        );
        Ok(())
    }

    #[test]
    fn tsi_control_disabled_token_accepts_all() -> Result<(), Box<dyn std::error::Error>> {
        // No token armed → legacy mode. Existing kernels that
        // boot without `supermachine.tsi_token=` send bare bodies;
        // the muxer must still dispatch them.
        let muxer = VsockMuxer::new(3)?;
        let hdr = control_header(TSI_PROXY_CREATE);

        let body = proxy_create_body(1234);
        let resp = muxer.handle_tsi_control(&hdr, &body);
        assert!(resp.is_empty(), "PROXY_CREATE doesn't reply");
        assert_eq!(muxer.tsi.lock().unwrap().len(), 1);
        Ok(())
    }

    #[test]
    fn ct_eq_matches_pairs_and_short_inputs() {
        let a = [0x55u8; TSI_TOKEN_LEN];
        let b = [0x55u8; TSI_TOKEN_LEN];
        assert!(ct_eq(&a, &b));

        // One byte different anywhere — equal-length case.
        let mut c = a;
        c[0] ^= 1;
        assert!(!ct_eq(&a, &c));
        c = a;
        c[TSI_TOKEN_LEN - 1] ^= 1;
        assert!(!ct_eq(&a, &c));

        // Length mismatch: short input must NEVER compare equal,
        // even if its bytes are a prefix of `a`. This is the
        // short-input case from the test plan.
        assert!(!ct_eq(&a[..16], &a[..]));
        assert!(!ct_eq(&[], &a[..]));
        assert!(!ct_eq(&a[..], &[]));
    }

    // -----------------------------------------------------------
    // Credit / flow-control regression suite.
    //
    // Pre-fix bug: handle_credit was a no-op, peer_buf_alloc and
    // peer_fwd_cnt were never tracked, on_data closures pushed
    // RW packets with fwd_cnt:0 unconditionally. Result: large
    // host→guest streams (e.g. 4 MiB HTTP GET via TSI host proxy)
    // overflowed the guest's vsock RX socket buffer and bytes
    // were silently dropped — the integrator's npm install died
    // with `ossl_gcm_stream_update: cipher operation failed`
    // because the TLS stream truncated mid-block.
    //
    // These tests pin the per-conn ProxyConnState invariants:
    // credit observed on every inbound packet, tx_cnt advances
    // on every outbound, pause flag trips when available credit
    // drops to the low watermark, ResumeReadable issued when
    // credit returns.
    // -----------------------------------------------------------

    fn stream_header(src_cid: u64, src_port: u32, dst_port: u32, op: u16) -> Header {
        Header {
            src_cid,
            dst_cid: VSOCK_HOST_CID,
            src_port,
            dst_port,
            len: 0,
            type_: VSOCK_TYPE_STREAM,
            op,
            flags: 0,
            buf_alloc: 0,
            fwd_cnt: 0,
        }
    }

    /// Insert a synthetic ProxyConnState into the muxer's
    /// `inbound_conns` map so handle_rw / handle_credit can run
    /// without a real TCP stream. Returns the Arc so the test
    /// can inspect state after dispatch.
    fn inject_state(
        muxer: &VsockMuxer,
        cid: u64,
        host_src_port: u32,
        guest_dst_port: u32,
    ) -> Arc<Mutex<ProxyConnState>> {
        let st = Arc::new(Mutex::new(ProxyConnState::new(
            guest_dst_port,
            host_src_port,
        )));
        muxer
            .inbound_conns
            .lock()
            .unwrap()
            .insert((cid, host_src_port), st.clone());
        st
    }

    #[test]
    fn credit_initial_peer_buf_alloc_is_zero() -> Result<(), Box<dyn std::error::Error>> {
        // T1: a freshly-constructed conn state advertises no
        // peer credit yet — we haven't observed any inbound
        // header. available_credit() must therefore be zero,
        // not OUR_BUF_ALLOC (the wrong default would silently
        // grant unbounded send credit and reproduce the bug).
        let st = ProxyConnState::new(/*guest_dst*/ 12_345, /*host_src*/ 50_000);
        assert_eq!(st.peer_buf_alloc, 0);
        assert_eq!(st.peer_fwd_cnt, 0);
        assert_eq!(st.tx_cnt, 0);
        assert_eq!(st.available_credit(), 0);
        assert!(!st.paused_readable);
        Ok(())
    }

    #[test]
    fn credit_update_handler_updates_state() -> Result<(), Box<dyn std::error::Error>> {
        // T2: dispatch a CREDIT_UPDATE packet and verify the
        // tracked state reflects the buf_alloc + fwd_cnt the
        // header carried. Pre-fix, handle_credit was a no-op
        // and this test would observe peer_buf_alloc still 0.
        let muxer = VsockMuxer::new(3)?;
        let state = inject_state(&muxer, 3, 50_000, 12_345);
        let mut hdr = stream_header(3, 12_345, 50_000, VSOCK_OP_CREDIT_UPDATE);
        hdr.buf_alloc = 262_144;
        hdr.fwd_cnt = 1_000;
        let _ = muxer.handle_credit(&hdr);
        let st = state.lock().unwrap();
        assert_eq!(st.peer_buf_alloc, 262_144);
        assert_eq!(st.peer_fwd_cnt, 1_000);
        Ok(())
    }

    #[test]
    fn rw_packet_updates_credit() -> Result<(), Box<dyn std::error::Error>> {
        // T3: an RW packet ALSO carries credit info in its
        // header; the muxer must observe it on every RW, not
        // just CREDIT_UPDATE. Without this, the guest never
        // gets credit-acked for what it pushed; with the inverse
        // miss (host-side), peer_fwd_cnt would never advance.
        let muxer = VsockMuxer::new(3)?;
        let state = inject_state(&muxer, 3, 50_000, 12_345);
        let mut hdr = stream_header(3, 12_345, 50_000, VSOCK_OP_RW);
        hdr.buf_alloc = 131_072;
        hdr.fwd_cnt = 4_096;
        let _ = muxer.handle_rw(&hdr, b"hello");
        let st = state.lock().unwrap();
        assert_eq!(st.peer_buf_alloc, 131_072);
        assert_eq!(st.peer_fwd_cnt, 4_096);
        // our_fwd_cnt must NOT advance at receipt — credit for
        // guest→host bytes is granted on DELIVERY to the host socket
        // (the io thread's on_drained callback). Granting at receipt
        // let a guest pump an entire multi-GiB stream into host RAM
        // while the socket lagged (builder #6).
        assert_eq!(st.our_fwd_cnt, 0, "credit must be delivery-coupled");
        Ok(())
    }

    #[test]
    fn on_drained_grants_credit_and_emits_update_at_threshold(
    ) -> Result<(), Box<dyn std::error::Error>> {
        // The delivery-coupled half of the #6 fix: the io thread reports
        // written byte counts via on_drained; our_fwd_cnt advances there,
        // and a CREDIT_UPDATE lands in the rxq once a threshold's worth
        // of bytes has been delivered since the last update.
        let muxer = VsockMuxer::new(3)?;
        let state = inject_state(&muxer, 3, 50_000, 12_345);
        let on_drained = muxer.make_on_drained(3, 50_000, state.clone());

        // Below the threshold: credit advances, no update packet yet.
        on_drained(1_000);
        assert_eq!(state.lock().unwrap().our_fwd_cnt, 1_000);
        assert!(muxer.rxq.lock().unwrap().pop().is_none(), "no update yet");

        // Cross the threshold: CREDIT_UPDATE emitted with the running fwd_cnt.
        on_drained(CREDIT_UPDATE_THRESHOLD as usize);
        let expect_fwd = 1_000 + CREDIT_UPDATE_THRESHOLD;
        assert_eq!(state.lock().unwrap().our_fwd_cnt, expect_fwd);
        let pkt = muxer
            .rxq
            .lock()
            .unwrap()
            .pop()
            .expect("CREDIT_UPDATE after threshold crossed");
        assert_eq!(pkt.hdr.op, VSOCK_OP_CREDIT_UPDATE);
        assert_eq!(pkt.hdr.fwd_cnt, expect_fwd);
        assert_eq!(
            pkt.hdr.src_port, 50_000,
            "credit src = conn's expected peer"
        );
        assert_eq!(pkt.hdr.dst_port, 12_345);
        assert_eq!(pkt.hdr.buf_alloc, OUR_BUF_ALLOC);

        // The threshold is RELATIVE to the last update: small follow-up
        // drains don't spam updates.
        on_drained(10);
        assert!(
            muxer.rxq.lock().unwrap().pop().is_none(),
            "no update until another threshold's worth drains"
        );
        Ok(())
    }

    #[test]
    fn tx_cnt_advances_on_outbound() -> Result<(), Box<dyn std::error::Error>> {
        // T4: when on_data pushes N bytes toward the guest,
        // tx_cnt must advance by exactly N. We don't have a
        // real on_data closure here, but ProxyConnState exposes
        // the same accounting the closure performs.
        let st = Arc::new(Mutex::new(ProxyConnState::new(12_345, 50_000)));
        // Seed peer credit so available is positive.
        st.lock().unwrap().peer_buf_alloc = 65_536;
        // Simulate three on_data invocations.
        for n in [1024u32, 2048u32, 512u32] {
            let mut s = st.lock().unwrap();
            s.tx_cnt = s.tx_cnt.wrapping_add(n);
        }
        let s = st.lock().unwrap();
        assert_eq!(s.tx_cnt, 1024 + 2048 + 512);
        // Available credit dropped by exactly the total pushed.
        assert_eq!(s.available_credit(), 65_536 - (1024 + 2048 + 512));
        Ok(())
    }

    #[test]
    fn pause_when_credit_exhausted() -> Result<(), Box<dyn std::error::Error>> {
        // T5: peer advertises 4096 bytes. After pushing 4096
        // bytes, available_credit drops to 0 ≤ low_watermark,
        // so the on_data path must trip the pause flag. The
        // muxer-io thread checks this after each on_data and
        // drops READABLE interest — exactly the backpressure
        // the original bug was missing.
        let st = Arc::new(Mutex::new(ProxyConnState::new(12_345, 50_000)));
        {
            let mut s = st.lock().unwrap();
            s.peer_buf_alloc = 4096;
        }
        // Simulate the on_data body: decrement available, set
        // pause when at-or-below watermark.
        {
            let mut s = st.lock().unwrap();
            s.tx_cnt = s.tx_cnt.wrapping_add(4096);
            let avail = s.available_credit();
            if avail <= CREDIT_LOW_WATERMARK {
                s.paused_readable = true;
                s.pause_flag.store(true, Ordering::Release);
            }
        }
        let s = st.lock().unwrap();
        assert_eq!(s.available_credit(), 0);
        assert!(
            s.paused_readable,
            "pause flag must be set when credit exhausted"
        );
        assert!(s.pause_flag.load(Ordering::Acquire));
        Ok(())
    }

    #[test]
    fn resume_after_credit_update() -> Result<(), Box<dyn std::error::Error>> {
        // T6: a paused conn receives a CREDIT_UPDATE that
        // advances peer_fwd_cnt past the prior in-flight bytes.
        // observe_peer must clear pause and return true so the
        // caller (handle_credit) sends ResumeReadable.
        let muxer = VsockMuxer::new(3)?;
        let state = inject_state(&muxer, 3, 50_000, 12_345);
        // Set up: peer advertised 65_536, we sent 65_536, paused.
        // Picking a larger window than CREDIT_LOW_WATERMARK so
        // an fwd_cnt advance creates strictly-positive headroom
        // above the watermark.
        {
            let mut s = state.lock().unwrap();
            s.peer_buf_alloc = 65_536;
            s.tx_cnt = 65_536;
            s.paused_readable = true;
            s.pause_flag.store(true, Ordering::Release);
            assert_eq!(s.available_credit(), 0);
        }
        // Now CREDIT_UPDATE with fwd_cnt=65_536 (guest consumed
        // everything). buf_alloc unchanged.
        let mut hdr = stream_header(3, 12_345, 50_000, VSOCK_OP_CREDIT_UPDATE);
        hdr.buf_alloc = 65_536;
        hdr.fwd_cnt = 65_536;
        let _ = muxer.handle_credit(&hdr);
        let s = state.lock().unwrap();
        assert_eq!(s.peer_fwd_cnt, 65_536);
        assert_eq!(s.available_credit(), 65_536);
        assert!(
            !s.paused_readable,
            "pause must clear once credit > watermark"
        );
        assert!(!s.pause_flag.load(Ordering::Acquire));
        Ok(())
    }

    #[test]
    fn handle_credit_flushes_rxq_stash_in_order_then_resumes(
    ) -> Result<(), Box<dyn std::error::Error>> {
        // The zero-loss back-pressure cycle end to end: a conn that
        // overflowed the shared rxq has stashed packets in pending_rx and
        // is paused (this is what the on_data closure does on a full rxq).
        // When the guest drains and sends a CREDIT_UPDATE, handle_credit
        // must re-push the stash IN ORDER into the rxq, and only THEN
        // resume the conn — never reading ahead of the stashed bytes.
        let muxer = VsockMuxer::new(3)?;
        let state = inject_state(&muxer, 3, 50_000, 12_345);
        {
            let mut s = state.lock().unwrap();
            s.peer_buf_alloc = 65_536; // ample credit — the rxq, not credit, was the blocker
            s.paused_readable = true;
            s.pause_flag.store(true, Ordering::Release);
            // Two RX packets stashed by an earlier rxq overflow, tagged so
            // we can assert FIFO order survives the flush.
            s.pending_rx.push_back(RxPacket {
                hdr: stream_header(VSOCK_HOST_CID, 50_000, 12_345, VSOCK_OP_RW),
                data: vec![1, 2, 3],
            });
            s.pending_rx.push_back(RxPacket {
                hdr: stream_header(VSOCK_HOST_CID, 50_000, 12_345, VSOCK_OP_RW),
                data: vec![4, 5, 6, 7],
            });
        }
        // Guest CREDIT_UPDATE (it drained its RX queue); the shared rxq
        // has room, so the flush succeeds and the conn resumes.
        let mut hdr = stream_header(3, 12_345, 50_000, VSOCK_OP_CREDIT_UPDATE);
        hdr.buf_alloc = 65_536;
        hdr.fwd_cnt = 0;
        let _ = muxer.handle_credit(&hdr);
        {
            let s = state.lock().unwrap();
            assert!(s.pending_rx.is_empty(), "stash must be fully flushed");
            assert!(!s.paused_readable, "conn must resume once the stash drains");
            assert!(!s.pause_flag.load(Ordering::Acquire));
        }
        // The stashed packets landed in the shared rxq, in FIFO order,
        // byte-intact — zero loss, correct ordering.
        let mut q = muxer.rxq.lock().unwrap();
        assert_eq!(q.pop().expect("first stashed packet").data, vec![1, 2, 3]);
        assert_eq!(
            q.pop().expect("second stashed packet").data,
            vec![4, 5, 6, 7]
        );
        assert!(
            q.pop().is_none(),
            "exactly the two stashed packets, nothing more"
        );
        Ok(())
    }

    #[test]
    fn handle_rw_grants_no_credit_until_delivery() -> Result<(), Box<dyn std::error::Error>> {
        // The receipt half of the #6 fix: a threshold-sized RW through
        // handle_rw must produce NO CREDIT_UPDATE and leave our_fwd_cnt
        // untouched — the guest's window only refills as the io thread
        // actually writes the bytes to the host socket (on_drained,
        // covered by `on_drained_grants_credit_and_emits_update_at_
        // threshold`). The pre-fix behaviour (credit at receipt) is what
        // let a ~3 GiB exec stream balloon the worker and wedge the bake.
        let muxer = VsockMuxer::new(3)?;
        let state = inject_state(&muxer, 3, 50_000, 12_345);
        let mut hdr = stream_header(3, 12_345, 50_000, VSOCK_OP_RW);
        hdr.buf_alloc = 262_144;
        hdr.fwd_cnt = 0;
        let big = vec![0u8; (CREDIT_UPDATE_THRESHOLD + 1) as usize];
        let out = muxer.handle_rw(&hdr, &big);
        assert!(
            !out.iter().any(|p| p.hdr.op == VSOCK_OP_CREDIT_UPDATE),
            "no CREDIT_UPDATE at receipt — credit is delivery-coupled"
        );
        assert_eq!(
            state.lock().unwrap().our_fwd_cnt,
            0,
            "our_fwd_cnt must not advance before the bytes hit the host socket"
        );
        Ok(())
    }

    #[test]
    fn io_thread_drains_large_backlog_in_order_with_delivery_credit(
    ) -> Result<(), Box<dyn std::error::Error>> {
        // End-to-end through the REAL io thread: build a multi-megabyte
        // guest→host backlog against a socket nobody reads (so it queues
        // in `pending_out`), then drain it. Asserts the #6 fix's two
        // properties at once:
        //   1. Chunked O(1) drain delivers every byte IN ORDER (the old
        //      whole-buffer `Vec::drain(..n)` was quadratic — a multi-GiB
        //      backlog effectively froze; 8 MiB through a ~64 KiB-buffered
        //      socketpair exercises hundreds of partial writes).
        //   2. on_drained reports exactly the delivered byte count
        //      (credit = delivery, never more than what actually hit the
        //      socket).
        use std::io::Read as _;
        use std::os::unix::net::UnixStream;
        use std::sync::atomic::AtomicUsize;
        use std::time::Duration;

        let muxer = VsockMuxer::new(3)?;
        let (host_side, mut peer) = UnixStream::pair()?;
        let drained = Arc::new(AtomicUsize::new(0));
        let drained_c = drained.clone();
        let on_drained: Arc<dyn Fn(usize) + Send + Sync> = Arc::new(move |n| {
            drained_c.fetch_add(n, Ordering::AcqRel);
        });
        let on_data: Arc<dyn Fn(Vec<u8>) + Send + Sync> = Arc::new(|_| {});
        let _ = muxer.io_tx.send(MuxerCmd::Register {
            host_src_port: 7777,
            stream: MuxerStream::Unix(host_side),
            on_data,
            pause_flag: Arc::new(AtomicBool::new(false)),
            on_drained,
        });
        let _ = muxer.io_waker.wake();

        // 8 MiB in 64 KiB chunks, each filled with its sequence number, so
        // any reordering / loss / duplication is detectable byte-by-byte.
        const CHUNK: usize = 64 * 1024;
        const CHUNKS: usize = 128;
        for i in 0..CHUNKS {
            let _ = muxer.io_tx.send(MuxerCmd::Write {
                host_src_port: 7777,
                bytes: vec![i as u8; CHUNK],
            });
            let _ = muxer.io_waker.wake();
        }

        // Drain the peer fully on this thread (the io thread flushes
        // pending_out as we make room). read_exact loops internally.
        let mut got = vec![0u8; CHUNK * CHUNKS];
        peer.read_exact(&mut got)?;
        for (i, chunk) in got.chunks(CHUNK).enumerate() {
            assert!(
                chunk.iter().all(|&b| b == i as u8),
                "chunk {i} corrupted/reordered"
            );
        }
        // All delivered bytes credited (poll briefly: the last on_drained
        // fires on the io thread after our final read makes room).
        let deadline = Instant::now() + Duration::from_secs(5);
        while drained.load(Ordering::Acquire) < CHUNK * CHUNKS {
            assert!(
                Instant::now() < deadline,
                "on_drained must account every delivered byte (got {} of {})",
                drained.load(Ordering::Acquire),
                CHUNK * CHUNKS
            );
            std::thread::sleep(Duration::from_millis(10));
        }
        assert_eq!(drained.load(Ordering::Acquire), CHUNK * CHUNKS);
        Ok(())
    }

    #[test]
    fn push_rxq_and_kick_hands_back_on_full_never_drops() {
        let rxq = Arc::new(Mutex::new(MuxerRxQ::with_cap(2)));
        let kick: Option<Arc<dyn Fn() + Send + Sync>> = None;
        // Fill to the cap.
        assert!(push_rxq_and_kick(
            &rxq,
            &kick,
            RxPacket {
                hdr: stream_header(2, 1000, 2000, VSOCK_OP_RW),
                data: vec![1]
            },
        )
        .is_none());
        assert!(push_rxq_and_kick(
            &rxq,
            &kick,
            RxPacket {
                hdr: stream_header(2, 1000, 2000, VSOCK_OP_RW),
                data: vec![2]
            },
        )
        .is_none());
        // The overflowing push hands the packet BACK intact — it must
        // never drop it — so the caller can stash it (zero byte loss).
        let handed_back = push_rxq_and_kick(
            &rxq,
            &kick,
            RxPacket {
                hdr: stream_header(2, 1000, 2000, VSOCK_OP_RW),
                data: vec![3, 4, 5],
            },
        );
        let pkt = handed_back.expect("a full rxq must hand the packet back, not drop it");
        assert_eq!(pkt.data, vec![3, 4, 5], "handed-back packet must be intact");
        assert_eq!(rxq.lock().unwrap().len(), 2, "queue stays at its cap");
    }

    #[test]
    fn push_control_and_kick_uses_reserve_when_data_full() {
        let rxq = Arc::new(Mutex::new(MuxerRxQ::with_cap(2)));
        let kick: Option<Arc<dyn Fn() + Send + Sync>> = None;
        // Saturate the data path at the cap.
        for i in 0..2u8 {
            assert!(push_rxq_and_kick(
                &rxq,
                &kick,
                RxPacket {
                    hdr: stream_header(2, 1000, 2000, VSOCK_OP_RW),
                    data: vec![i]
                },
            )
            .is_none());
        }
        assert_eq!(rxq.lock().unwrap().len(), 2);
        // A connection-setup REQUEST still lands — drawn from the control
        // reserve ABOVE the data cap, so it is NOT dropped under load (the
        // data path back-pressures, but control has no stash to fall into).
        push_control_and_kick(
            &rxq,
            &kick,
            RxPacket {
                hdr: stream_header(2, 1000, 2000, VSOCK_OP_REQUEST),
                data: vec![],
            },
        );
        assert_eq!(
            rxq.lock().unwrap().len(),
            3,
            "control REQUEST must land in the reserve even when the data queue is full"
        );
    }

    #[test]
    fn push_control_and_kick_releases_rxq_lock_before_kicking() {
        // Regression: the real kick closure re-enters the rxq lock
        // (device.kick → try_drain_rx → muxer.rxq.lock). If
        // push_control_and_kick held the lock across the kick, that
        // re-lock would self-deadlock (std Mutex is non-reentrant),
        // wedging every host→guest connection setup — the bug that broke
        // exec/relay/ipv6 before this guard. The kick here `try_lock`s the
        // rxq and records success: it can only succeed if the lock was
        // released first. `try_lock` (not `lock`) so a regression FAILS
        // the assert instead of hanging the test runner.
        use std::sync::atomic::{AtomicBool, Ordering};
        let rxq = Arc::new(Mutex::new(MuxerRxQ::with_cap(4)));
        let rxq_for_kick = Arc::clone(&rxq);
        let kicked_with_lock_free = Arc::new(AtomicBool::new(false));
        let flag = Arc::clone(&kicked_with_lock_free);
        let kick: Option<Arc<dyn Fn() + Send + Sync>> = Some(Arc::new(move || {
            if rxq_for_kick.try_lock().is_ok() {
                flag.store(true, Ordering::SeqCst);
            }
        }));
        // Empty queue → push_control returns was_empty=true → kick fires.
        push_control_and_kick(
            &rxq,
            &kick,
            RxPacket {
                hdr: stream_header(2, 1000, 2000, VSOCK_OP_REQUEST),
                data: vec![],
            },
        );
        assert!(
            kicked_with_lock_free.load(Ordering::SeqCst),
            "kick must run with the rxq lock RELEASED (else it self-deadlocks in production)"
        );
    }

    #[test]
    fn observe_peer_will_not_resume_while_rxq_stash_pending() {
        let mut st = ProxyConnState::new(2000, 1000);
        st.paused_readable = true;
        st.pause_flag.store(true, Ordering::Release);
        // A back-pressure stash is outstanding. Even with ample credit,
        // resuming now would let the io thread read+push ahead of the
        // stashed (older) bytes — out-of-order corruption. Stay paused
        // until handle_rw / handle_credit drain the stash.
        st.pending_rx.push_back(RxPacket {
            hdr: stream_header(2, 1000, 2000, VSOCK_OP_RW),
            data: vec![0u8; 8],
        });
        assert!(
            !st.observe_peer(1 << 20, 0),
            "must not resume while the rxq stash is non-empty"
        );
        assert!(st.paused_readable);
        assert!(st.pause_flag.load(Ordering::Acquire));
        // Once the stash has been flushed, resume fires (credit is ample).
        st.pending_rx.clear();
        assert!(
            st.observe_peer(1 << 20, 0),
            "must resume once the stash is flushed and credit is ample"
        );
        assert!(!st.paused_readable);
        assert!(!st.pause_flag.load(Ordering::Acquire));
    }

    // ── ProxyConnState credit accounting (u32 wrapping arithmetic) ─────

    #[test]
    fn available_credit_basic_subtraction() {
        let mut st = ProxyConnState::new(12_345, 50_000);
        st.peer_buf_alloc = 1000;
        st.tx_cnt = 300;
        st.peer_fwd_cnt = 100;
        // in_flight = 300 - 100 = 200; credit = 1000 - 200 = 800.
        assert_eq!(st.available_credit(), 800);
    }

    #[test]
    fn available_credit_saturates_when_overcommitted() {
        // We believe more is in flight than the peer's buffer holds (e.g.
        // a CREDIT_UPDATE not yet observed) → must saturate to 0, never
        // underflow-panic.
        let mut st = ProxyConnState::new(12_345, 50_000);
        st.peer_buf_alloc = 100;
        st.tx_cnt = 500;
        st.peer_fwd_cnt = 0;
        assert_eq!(st.available_credit(), 0);
    }

    #[test]
    fn available_credit_wraps_across_u32_boundary() {
        // The whole reason vsock counters are u32 *wrapping*: our tx_cnt
        // has wrapped past u32::MAX while the peer's fwd_cnt has not yet.
        // The in-flight delta must still resolve to the true byte count.
        let mut st = ProxyConnState::new(12_345, 50_000);
        st.peer_buf_alloc = 10_000;
        st.tx_cnt = 10; // wrapped around to 10
        st.peer_fwd_cnt = u32::MAX - 5; // 16 bytes behind, pre-wrap
        assert_eq!(st.tx_cnt.wrapping_sub(st.peer_fwd_cnt), 16);
        assert_eq!(st.available_credit(), 10_000 - 16);
    }

    #[test]
    fn observe_peer_resumes_strictly_above_watermark() {
        // available_credit() == peer_buf_alloc here (tx_cnt == peer_fwd_cnt).
        let resume_at = |credit: u32| -> bool {
            let mut st = ProxyConnState::new(12_345, 50_000);
            st.paused_readable = true;
            st.pause_flag.store(true, Ordering::Release);
            st.tx_cnt = 0;
            st.observe_peer(credit, 0)
        };
        assert!(
            !resume_at(CREDIT_LOW_WATERMARK),
            "exactly at the watermark must stay paused (strict >)"
        );
        assert!(
            resume_at(CREDIT_LOW_WATERMARK + 1),
            "one byte above the watermark resumes"
        );
    }

    #[test]
    fn observe_peer_no_resume_when_not_paused() {
        // Not paused → never emits a spurious resume even with full credit.
        let mut st = ProxyConnState::new(12_345, 50_000);
        st.tx_cnt = 0;
        assert!(!st.observe_peer(u32::MAX / 2, 0));
        assert!(!st.paused_readable);
        assert!(!st.pause_flag.load(Ordering::Acquire));
    }

    proptest! {
        #![proptest_config(ProptestConfig::with_cases(1024))]

        /// available_credit() never panics and always equals the documented
        /// wrapping formula for any counter values (incl. across wrap).
        #[test]
        fn available_credit_matches_formula_for_all_counters(
            buf_alloc in any::<u32>(),
            tx in any::<u32>(),
            fwd in any::<u32>(),
        ) {
            let mut st = ProxyConnState::new(1, 2);
            st.peer_buf_alloc = buf_alloc;
            st.tx_cnt = tx;
            st.peer_fwd_cnt = fwd;
            let expected = buf_alloc.saturating_sub(tx.wrapping_sub(fwd));
            prop_assert_eq!(st.available_credit(), expected);
        }
    }

    // === Tier-3 fuzz: hostile TSI control wire input ===================
    //
    // `handle_tsi_control` decodes guest-controlled DGRAM bodies: a
    // u32-at-offset here, a sockaddr there, a NUL-terminated host there.
    // A malicious guest sends arbitrary dst_port (the opcode) with
    // arbitrary body bytes — short, oversized, garbage. The host must
    // never panic or read out of bounds, and every reply RxPacket it
    // emits must be self-consistent (declared header `len` == actual
    // payload length) so the rx path never over- or under-reads it.
    use proptest::prelude::*;

    fn assert_rx_consistent(resp: &[RxPacket]) {
        for rp in resp {
            assert_eq!(
                rp.hdr.len as usize,
                rp.data.len(),
                "RxPacket hdr.len ({}) != data ({} bytes) — reply length confusion",
                rp.hdr.len,
                rp.data.len(),
            );
        }
    }

    proptest! {
        #![proptest_config(ProptestConfig::with_cases(1024))]

        /// No-token muxer: arbitrary opcode + body never panics and emits
        /// only self-consistent replies.
        #[test]
        fn tsi_control_open_is_panic_free(
            // Bias dst_port toward the real opcode window 1024..=1031,
            // with occasional wild values to hit the default arm.
            dst_port in prop_oneof![1024u32..=1031, any::<u32>()],
            body in proptest::collection::vec(any::<u8>(), 0..512),
        ) {
            let muxer = VsockMuxer::new(3).expect("muxer");
            let hdr = control_header(dst_port);
            let resp = muxer.handle_tsi_control(&hdr, &body);
            assert_rx_consistent(&resp);
        }

        /// Token-gated muxer: a guest that doesn't know the token must be
        /// silently dropped (no reply) on every control op except the
        /// SENDTO_DATA data-path exemption — and never panics. A body that
        /// happens to carry the correct token prefix is exercised too.
        #[test]
        fn tsi_control_token_gate_is_panic_free(
            dst_port in prop_oneof![1024u32..=1031, any::<u32>()],
            prefix_correct in any::<bool>(),
            tail in proptest::collection::vec(any::<u8>(), 0..256),
        ) {
            let token = [0xABu8; TSI_TOKEN_LEN];
            let muxer = VsockMuxer::with_tsi_token(3, Some(token)).expect("muxer");
            let hdr = control_header(dst_port);

            let mut body = Vec::new();
            if prefix_correct {
                body.extend_from_slice(&token);
            } else {
                body.extend_from_slice(&[0x00u8; TSI_TOKEN_LEN]);
            }
            body.extend_from_slice(&tail);

            let resp = muxer.handle_tsi_control(&hdr, &body);
            assert_rx_consistent(&resp);

            // A wrong-token, non-data-path op must produce NO reply
            // (silent drop — no oracle for a probing attacker).
            if !prefix_correct && dst_port != TSI_SENDTO_DATA {
                prop_assert!(
                    resp.is_empty(),
                    "wrong-token control op on port {} leaked a reply",
                    dst_port,
                );
            }
        }
    }
}
