rns_net/interface/

auto.rs

//! AutoInterface: Zero-configuration LAN auto-discovery via IPv6 multicast.
//!
//! Matches Python `AutoInterface` from `RNS/Interfaces/AutoInterface.py`.
//!
//! Thread model (per adopted network interface):
//!   - Discovery sender: periodically sends discovery token via multicast
//!   - Discovery receiver (multicast): validates tokens, adds peers
//!   - Discovery receiver (unicast): validates reverse-peering tokens
//!   - Data receiver: UDP server receiving unicast data from peers
//!
//! Additionally one shared thread:
//!   - Peer jobs: periodically culls timed-out peers

use std::collections::{HashMap, HashSet, VecDeque};
use std::io;
use std::net::{Ipv6Addr, SocketAddrV6, UdpSocket};
use std::os::unix::io::AsRawFd;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use rns_core::transport::types::InterfaceId;

use crate::event::{Event, EventSender};
use crate::interface::{lock_or_recover, Writer};

// ── Constants (matching Python AutoInterface) ──────────────────────────────

/// Default UDP port for multicast discovery.
pub const DEFAULT_DISCOVERY_PORT: u16 = 29716;

/// Default UDP port for unicast data exchange.
pub const DEFAULT_DATA_PORT: u16 = 42671;

/// Default group identifier.
pub const DEFAULT_GROUP_ID: &[u8] = b"reticulum";

/// Default IFAC size for AutoInterface (bytes).
pub const DEFAULT_IFAC_SIZE: usize = 16;

/// Hardware MTU for AutoInterface packets.
pub const HW_MTU: usize = 1196;

/// Multicast scope: link-local.
pub const SCOPE_LINK: &str = "2";
/// Multicast scope: admin-local.
pub const SCOPE_ADMIN: &str = "4";
/// Multicast scope: site-local.
pub const SCOPE_SITE: &str = "5";
/// Multicast scope: organization-local.
pub const SCOPE_ORGANISATION: &str = "8";
/// Multicast scope: global.
pub const SCOPE_GLOBAL: &str = "e";

/// Permanent multicast address type.
pub const MULTICAST_PERMANENT_ADDRESS_TYPE: &str = "0";
/// Temporary multicast address type.
pub const MULTICAST_TEMPORARY_ADDRESS_TYPE: &str = "1";

/// How long before a peer is considered timed out (seconds).
pub const PEERING_TIMEOUT: f64 = 22.0;

/// How often to send multicast discovery announcements (seconds).
pub const ANNOUNCE_INTERVAL: f64 = 1.6;

/// How often to run peer maintenance jobs (seconds).
pub const PEER_JOB_INTERVAL: f64 = 4.0;

/// Multicast echo timeout (seconds). Used for carrier detection.
pub const MCAST_ECHO_TIMEOUT: f64 = 6.5;

/// Default bitrate guess for AutoInterface (10 Mbps).
pub const BITRATE_GUESS: u64 = 10_000_000;

/// Deduplication deque size.
pub const MULTI_IF_DEQUE_LEN: usize = 48;

/// Deduplication deque entry TTL (seconds).
pub const MULTI_IF_DEQUE_TTL: f64 = 0.75;

/// Reverse peering interval multiplier (announce_interval * 3.25).
pub const REVERSE_PEERING_MULTIPLIER: f64 = 3.25;

/// How often AutoInterface rescans local link-local interfaces.
pub const AUTO_RESCAN_INTERVAL: f64 = 1.25;

/// Default cap on active peers per discovery worker (one adopted network
/// interface). Discovery tokens from further sources are ignored until an
/// existing peer times out. Configurable with `max_peers`.
pub const DEFAULT_MAX_PEERS: usize = 128;

/// New peers each discovery worker admits within one
/// [`PEER_ADMISSION_WINDOW`].
pub const MAX_NEW_PEERS_PER_WINDOW: usize = 32;

/// Length of the window over which [`MAX_NEW_PEERS_PER_WINDOW`] applies.
pub const PEER_ADMISSION_WINDOW: Duration = Duration::from_secs(10);

/// Minimum spacing between warnings about ignored discovery sources, per worker.
const PEER_REFUSAL_LOG_INTERVAL: Duration = Duration::from_secs(10);

/// Interfaces always ignored.
pub const ALL_IGNORE_IFS: &[&str] = &["lo0"];

#[cfg_attr(not(any(target_os = "android", test)), allow(dead_code))]
const ANDROID_IGNORE_IFS: &[&str] = &[
    "dummy0", "lo", "tun0", "rmnet0", "rmnet1", "rmnet2", "rmnet3", "rmnet4", "rmnet5", "rmnet6",
    "rmnet7",
];

// ── Configuration ──────────────────────────────────────────────────────────

/// Configuration for an AutoInterface.
#[derive(Debug, Clone)]
pub struct AutoConfig {
    pub name: String,
    pub group_id: Vec<u8>,
    pub discovery_scope: String,
    pub discovery_port: u16,
    pub data_port: u16,
    pub multicast_address_type: String,
    pub allowed_interfaces: Vec<String>,
    pub ignored_interfaces: Vec<String>,
    pub configured_bitrate: u64,
    /// Maximum number of active peers per adopted network interface
    /// (`max_peers`, default [`DEFAULT_MAX_PEERS`]). Discovery tokens from
    /// further sources are ignored until an existing peer times out.
    pub max_peers: usize,
    /// Base interface ID. Per-peer IDs will be assigned dynamically.
    pub interface_id: InterfaceId,
    pub ingress_control: rns_core::transport::types::IngressControlConfig,
    pub runtime: Arc<Mutex<AutoRuntime>>,
}

#[derive(Debug, Clone)]
pub struct AutoRuntime {
    pub announce_interval_secs: f64,
    pub peer_timeout_secs: f64,
    pub peer_job_interval_secs: f64,
}

impl AutoRuntime {
    pub fn from_config(_config: &AutoConfig) -> Self {
        Self {
            announce_interval_secs: ANNOUNCE_INTERVAL,
            peer_timeout_secs: PEERING_TIMEOUT,
            peer_job_interval_secs: PEER_JOB_INTERVAL,
        }
    }
}

#[derive(Debug, Clone)]
pub struct AutoRuntimeConfigHandle {
    pub interface_name: String,
    pub runtime: Arc<Mutex<AutoRuntime>>,
    pub startup: AutoRuntime,
}

impl Default for AutoConfig {
    fn default() -> Self {
        let mut config = AutoConfig {
            name: String::new(),
            group_id: DEFAULT_GROUP_ID.to_vec(),
            discovery_scope: SCOPE_LINK.to_string(),
            discovery_port: DEFAULT_DISCOVERY_PORT,
            data_port: DEFAULT_DATA_PORT,
            multicast_address_type: MULTICAST_TEMPORARY_ADDRESS_TYPE.to_string(),
            allowed_interfaces: Vec::new(),
            ignored_interfaces: Vec::new(),
            configured_bitrate: BITRATE_GUESS,
            max_peers: DEFAULT_MAX_PEERS,
            interface_id: InterfaceId(0),
            ingress_control: rns_core::transport::types::IngressControlConfig::enabled(),
            runtime: Arc::new(Mutex::new(AutoRuntime {
                announce_interval_secs: ANNOUNCE_INTERVAL,
                peer_timeout_secs: PEERING_TIMEOUT,
                peer_job_interval_secs: PEER_JOB_INTERVAL,
            })),
        };
        let startup = AutoRuntime::from_config(&config);
        config.runtime = Arc::new(Mutex::new(startup));
        config
    }
}

// ── Multicast address derivation ───────────────────────────────────────────

/// Derive the IPv6 multicast discovery address from group_id, scope, and address type.
///
/// Algorithm (matching Python):
///   1. group_hash = SHA-256(group_id)
///   2. Build suffix from hash bytes 2..14 as 6 little-endian 16-bit words
///   3. First word is hardcoded "0"
///   4. Prefix = "ff" + address_type + scope
pub fn derive_multicast_address(group_id: &[u8], address_type: &str, scope: &str) -> String {
    let group_hash = rns_crypto::sha256::sha256(group_id);
    let g = &group_hash;

    // Build 6 LE 16-bit words from bytes 2..14
    let w1 = (g[2] as u16) << 8 | g[3] as u16;
    let w2 = (g[4] as u16) << 8 | g[5] as u16;
    let w3 = (g[6] as u16) << 8 | g[7] as u16;
    let w4 = (g[8] as u16) << 8 | g[9] as u16;
    let w5 = (g[10] as u16) << 8 | g[11] as u16;
    let w6 = (g[12] as u16) << 8 | g[13] as u16;

    format!(
        "ff{}{}:0:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}",
        address_type, scope, w1, w2, w3, w4, w5, w6
    )
}

/// Parse a multicast address string into an Ipv6Addr.
pub fn parse_multicast_addr(addr: &str) -> Option<Ipv6Addr> {
    addr.parse::<Ipv6Addr>().ok()
}

// ── Discovery token ────────────────────────────────────────────────────────

/// Compute the discovery token: SHA-256(group_id + link_local_address_string).
pub fn compute_discovery_token(group_id: &[u8], link_local_addr: &str) -> [u8; 32] {
    let mut input = group_id.to_vec();
    input.extend_from_slice(link_local_addr.as_bytes());
    rns_crypto::sha256::sha256(&input)
}

// ── Network interface enumeration ──────────────────────────────────────────

/// Information about a local network interface with an IPv6 link-local address.
#[derive(Debug, Clone)]
pub struct LocalInterface {
    pub name: String,
    pub link_local_addr: String,
    pub index: u32,
}

#[cfg(target_os = "android")]
fn platform_ignored_interfaces() -> &'static [&'static str] {
    ANDROID_IGNORE_IFS
}

#[cfg(not(target_os = "android"))]
fn platform_ignored_interfaces() -> &'static [&'static str] {
    &[]
}

fn should_adopt_interface_name(
    name: &str,
    allowed: &[String],
    ignored: &[String],
    platform_ignored: &[&str],
) -> bool {
    let is_allowed = allowed.iter().any(|a| a == name);
    let is_system_ignored = ALL_IGNORE_IFS.contains(&name) || platform_ignored.contains(&name);

    if is_system_ignored && !is_allowed {
        return false;
    }

    if ignored.iter().any(|ig| ig == name) {
        return false;
    }

    if !allowed.is_empty() && !is_allowed {
        return false;
    }

    true
}

/// Enumerate network interfaces that have IPv6 link-local addresses (fe80::/10).
///
/// Uses `libc::getifaddrs()`. Filters by allowed/ignored interface lists.
pub fn enumerate_interfaces(allowed: &[String], ignored: &[String]) -> Vec<LocalInterface> {
    let mut result = Vec::new();
    let platform_ignored = platform_ignored_interfaces();

    unsafe {
        let mut ifaddrs: *mut libc::ifaddrs = std::ptr::null_mut();
        if libc::getifaddrs(&mut ifaddrs) != 0 {
            return result;
        }

        let mut current = ifaddrs;
        while !current.is_null() {
            let ifa = &*current;
            current = ifa.ifa_next;

            // Must have an address
            if ifa.ifa_addr.is_null() {
                continue;
            }

            // Must be AF_INET6
            if (*ifa.ifa_addr).sa_family as i32 != libc::AF_INET6 {
                continue;
            }

            // Get interface name
            let name = match std::ffi::CStr::from_ptr(ifa.ifa_name).to_str() {
                Ok(s) => s.to_string(),
                Err(_) => continue,
            };

            if !should_adopt_interface_name(&name, allowed, ignored, platform_ignored) {
                continue;
            }

            // Extract IPv6 address
            let sa6 = ifa.ifa_addr as *const libc::sockaddr_in6;
            let addr_bytes = (*sa6).sin6_addr.s6_addr;
            let ipv6 = Ipv6Addr::from(addr_bytes);

            // Must be link-local (fe80::/10)
            let octets = ipv6.octets();
            if octets[0] != 0xfe || (octets[1] & 0xc0) != 0x80 {
                continue;
            }

            // Format the address (drop scope ID, matching Python's descope_linklocal)
            let addr_str = format!("{}", ipv6);

            // Get interface index
            let index = libc::if_nametoindex(ifa.ifa_name);
            if index == 0 {
                continue;
            }

            // Avoid duplicates (same interface may appear multiple times)
            if result.iter().any(|li: &LocalInterface| li.name == name) {
                continue;
            }

            result.push(LocalInterface {
                name,
                link_local_addr: addr_str,
                index,
            });
        }

        libc::freeifaddrs(ifaddrs);
    }

    result
}

mod folded;

// ── Peer and worker tracking ───────────────────────────────────────────────

#[cfg_attr(not(test), allow(dead_code))]
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct AutoWorkerKey {
    ifname: String,
    if_index: u32,
    link_local_addr: String,
}

#[allow(dead_code)]
impl AutoWorkerKey {
    fn from_local_interface(local: &LocalInterface) -> Self {
        Self {
            ifname: local.name.clone(),
            if_index: local.index,
            link_local_addr: local.link_local_addr.clone(),
        }
    }
}

#[cfg_attr(not(test), allow(dead_code))]
#[derive(Debug, Clone, PartialEq, Eq)]
enum WorkerReconcileAction {
    Add(AutoWorkerKey),
    Remove(AutoWorkerKey),
}

#[cfg_attr(not(test), allow(dead_code))]
fn reconcile_worker_keys(
    active: impl IntoIterator<Item = AutoWorkerKey>,
    desired: impl IntoIterator<Item = AutoWorkerKey>,
) -> Vec<WorkerReconcileAction> {
    let active = active.into_iter().collect::<HashSet<_>>();
    let desired = desired.into_iter().collect::<HashSet<_>>();
    let mut actions = Vec::new();

    for key in active.difference(&desired) {
        actions.push(WorkerReconcileAction::Remove(key.clone()));
    }
    for key in desired.difference(&active) {
        actions.push(WorkerReconcileAction::Add(key.clone()));
    }

    actions.sort_by(|a, b| format!("{a:?}").cmp(&format!("{b:?}")));
    actions
}

struct RunningAutoWorker {
    running: Arc<AtomicBool>,
}

impl RunningAutoWorker {
    fn stop(&self) {
        self.running.store(false, Ordering::Relaxed);
    }
}

enum AutoWorkerEntry {
    Running(RunningAutoWorker),
    Pending,
}

impl AutoWorkerEntry {
    fn stop(&self) {
        if let AutoWorkerEntry::Running(worker) = self {
            worker.stop();
        }
    }
}

fn reconcile_auto_workers<F>(
    workers: &mut HashMap<AutoWorkerKey, AutoWorkerEntry>,
    desired: Vec<AutoWorkerKey>,
    mut start_worker: F,
) where
    F: FnMut(&AutoWorkerKey) -> io::Result<RunningAutoWorker>,
{
    let mut desired_keys = desired.into_iter().collect::<HashSet<_>>();
    let active_keys = workers.keys().cloned().collect::<Vec<_>>();

    for key in active_keys {
        if !desired_keys.contains(&key) {
            if let Some(worker) = workers.remove(&key) {
                worker.stop();
            }
        }
    }

    let mut desired = desired_keys.drain().collect::<Vec<_>>();
    desired.sort_by(|a, b| a.ifname.cmp(&b.ifname).then(a.if_index.cmp(&b.if_index)));

    for key in desired {
        if matches!(workers.get(&key), Some(AutoWorkerEntry::Running(_))) {
            continue;
        }

        let entry = match start_worker(&key) {
            Ok(worker) => AutoWorkerEntry::Running(worker),
            Err(e) => {
                log::warn!(
                    "AutoInterface worker {} failed to start on {}%{}: {}",
                    key.ifname,
                    key.link_local_addr,
                    key.if_index,
                    e
                );
                AutoWorkerEntry::Pending
            }
        };
        workers.insert(key, entry);
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct PeerKey {
    link_local_addr: String,
    if_index: u32,
}

/// A discovered peer.
struct AutoPeer {
    interface_id: InterfaceId,
    #[allow(dead_code)]
    link_local_addr: String,
    #[allow(dead_code)]
    ifname: String,
    #[allow(dead_code)]
    if_index: u32,
    last_heard: f64,
}

/// Outcome of asking a discovery worker to admit a new peer.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PeerAdmissionDecision {
    Admit,
    /// The worker already has `max_peers` active peers.
    AtCapacity,
    /// The worker admitted [`MAX_NEW_PEERS_PER_WINDOW`] peers in the current window.
    RateLimited,
}

/// Per-worker bookkeeping for admitting new peers.
#[derive(Debug, Default)]
struct PeerAdmission {
    window_start: Option<Instant>,
    admitted_in_window: usize,
    last_refusal_log: Option<Instant>,
    refusals_since_log: usize,
}

impl PeerAdmission {
    /// Decide whether a new peer may be created, given the worker's current
    /// number of active peers. An admitted peer counts against the window.
    fn decide(
        &mut self,
        active_peers: usize,
        max_peers: usize,
        now: Instant,
    ) -> PeerAdmissionDecision {
        if active_peers >= max_peers {
            return PeerAdmissionDecision::AtCapacity;
        }
        let window_elapsed = self
            .window_start
            .is_none_or(|start| now.saturating_duration_since(start) >= PEER_ADMISSION_WINDOW);
        if window_elapsed {
            self.window_start = Some(now);
            self.admitted_in_window = 0;
        }
        if self.admitted_in_window >= MAX_NEW_PEERS_PER_WINDOW {
            return PeerAdmissionDecision::RateLimited;
        }
        self.admitted_in_window += 1;
        PeerAdmissionDecision::Admit
    }

    /// Record a refusal. Returns the number of refusals to report when a
    /// warning is due, or `None` while warnings are being spaced out.
    fn note_refusal(&mut self, now: Instant) -> Option<usize> {
        self.refusals_since_log = self.refusals_since_log.saturating_add(1);
        let due = self
            .last_refusal_log
            .is_none_or(|last| now.saturating_duration_since(last) >= PEER_REFUSAL_LOG_INTERVAL);
        if !due {
            return None;
        }
        self.last_refusal_log = Some(now);
        Some(std::mem::take(&mut self.refusals_since_log))
    }
}

/// Writer that sends UDP unicast data to a peer.
struct UdpWriter {
    socket: UdpSocket,
    target: SocketAddrV6,
    /// Called by the driver directly on a non-blocking socket (see
    /// [`Writer::sends_inline`]), with interface I/O folded.
    inline: bool,
}

impl Writer for UdpWriter {
    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
        match self.socket.send_to(data, self.target) {
            // A datagram the socket buffer cannot take is dropped, as the
            // network may drop it.
            Err(e) if self.inline && e.kind() == io::ErrorKind::WouldBlock => Ok(()),
            result => result.map(|_| ()),
        }
    }

    fn send_frame_confirmed(&mut self, data: &[u8]) -> io::Result<()> {
        self.socket.send_to(data, self.target).map(|_| ())
    }

    fn enqueue_confirmed(
        &mut self,
        data: &[u8],
        completion: &mut Option<crate::link_send::Completion>,
    ) -> io::Result<()> {
        // A full buffer returns WouldBlock with the completion untouched,
        // and the driver retries the frame.
        self.socket.send_to(data, self.target)?;
        if let Some(completion) = completion.take() {
            completion.finish(Ok(()));
        }
        Ok(())
    }

    fn sends_inline(&self) -> bool {
        self.inline
    }
}

/// Shared state for the AutoInterface across all threads.
struct SharedState {
    /// Known peers, scoped by link-local address and interface index.
    peers: HashMap<PeerKey, AutoPeer>,
    /// Our own scoped link-local addresses (for echo detection).
    link_local_addresses: HashSet<PeerKey>,
    /// Deduplication deque: (hash, timestamp)
    dedup_deque: VecDeque<([u8; 32], f64)>,
    /// Flag set when final_init is done
    online: bool,
    /// Next dynamic interface ID
    next_id: Arc<AtomicU64>,
    /// Maximum active peers per adopted network interface.
    max_peers: usize,
    /// New-peer admission state, keyed by adopted network interface name.
    peer_admission: HashMap<String, PeerAdmission>,
}

impl SharedState {
    fn new(next_id: Arc<AtomicU64>) -> Self {
        SharedState {
            peers: HashMap::new(),
            link_local_addresses: HashSet::new(),
            dedup_deque: VecDeque::new(),
            online: false,
            next_id,
            max_peers: DEFAULT_MAX_PEERS,
            peer_admission: HashMap::new(),
        }
    }

    /// Number of active peers discovered on the adopted interface `ifname`.
    fn active_peers_on(&self, ifname: &str) -> usize {
        self.peers
            .values()
            .filter(|peer| peer.ifname == ifname)
            .count()
    }

    /// Decide whether the worker on `ifname` may create one more peer.
    fn admit_new_peer(&mut self, ifname: &str, now: Instant) -> PeerAdmissionDecision {
        let active_peers = self.active_peers_on(ifname);
        let max_peers = self.max_peers;
        self.peer_admission
            .entry(ifname.to_string())
            .or_default()
            .decide(active_peers, max_peers, now)
    }

    /// Record an ignored discovery source for `ifname`; see [`PeerAdmission::note_refusal`].
    fn note_peer_refusal(&mut self, ifname: &str, now: Instant) -> Option<usize> {
        self.peer_admission
            .entry(ifname.to_string())
            .or_default()
            .note_refusal(now)
    }

    /// Remove peers not heard from within `peer_timeout_secs`, returning them.
    fn remove_timed_out_peers(
        &mut self,
        now: f64,
        peer_timeout_secs: f64,
    ) -> Vec<(PeerKey, InterfaceId)> {
        let timed_out = self
            .peers
            .iter()
            .filter(|(_, peer)| now > peer.last_heard + peer_timeout_secs)
            .map(|(peer_key, peer)| (peer_key.clone(), peer.interface_id))
            .collect::<Vec<_>>();
        for (peer_key, _) in &timed_out {
            self.peers.remove(peer_key);
        }
        timed_out
    }

    /// Check dedup deque for a data hash.
    fn is_duplicate(&self, hash: &[u8; 32], now: f64) -> bool {
        for (h, ts) in &self.dedup_deque {
            if h == hash && now < *ts + MULTI_IF_DEQUE_TTL {
                return true;
            }
        }
        false
    }

    /// Add to dedup deque, trimming to max length.
    fn add_dedup(&mut self, hash: [u8; 32], now: f64) {
        self.dedup_deque.push_back((hash, now));
        while self.dedup_deque.len() > MULTI_IF_DEQUE_LEN {
            self.dedup_deque.pop_front();
        }
    }

    /// Refresh a peer's last_heard timestamp.
    fn refresh_peer(&mut self, key: &PeerKey, now: f64) {
        if let Some(peer) = self.peers.get_mut(key) {
            peer.last_heard = now;
        }
    }
}

fn peer_key_from_src(src_addr: &SocketAddrV6, fallback_if_index: u32) -> PeerKey {
    let if_index = match src_addr.scope_id() {
        0 => fallback_if_index,
        scope_id => scope_id,
    };

    PeerKey {
        link_local_addr: src_addr.ip().to_string(),
        if_index,
    }
}

fn peer_target_addr(peer_key: &PeerKey, data_port: u16) -> io::Result<SocketAddrV6> {
    let peer_ip = peer_key.link_local_addr.parse::<Ipv6Addr>().map_err(|e| {
        io::Error::new(io::ErrorKind::InvalidInput, format!("bad peer IPv6: {}", e))
    })?;

    Ok(SocketAddrV6::new(peer_ip, data_port, 0, peer_key.if_index))
}

fn remove_peers_for_worker(
    shared: &Arc<Mutex<SharedState>>,
    tx: &EventSender,
    key: &AutoWorkerKey,
) {
    for iface_id in take_peers_for_worker(shared, key) {
        let _ = tx.send(Event::InterfaceDown(iface_id));
    }
}

/// Forget the peers found through the worker for `key`, returning their
/// interfaces, which the caller reports down.
fn take_peers_for_worker(
    shared: &Arc<Mutex<SharedState>>,
    key: &AutoWorkerKey,
) -> Vec<InterfaceId> {
    let removed = {
        let mut state = lock_or_recover(shared, "auto shared state");
        let removed = state
            .peers
            .iter()
            .filter_map(|(peer_key, peer)| {
                if peer.if_index == key.if_index && peer.ifname == key.ifname {
                    Some((peer_key.clone(), peer.interface_id))
                } else {
                    None
                }
            })
            .collect::<Vec<_>>();

        for (peer_key, _) in &removed {
            state.peers.remove(peer_key);
        }
        state.peer_admission.remove(&key.ifname);

        removed
    };

    removed.into_iter().map(|(_, iface_id)| iface_id).collect()
}

// ── Start function ─────────────────────────────────────────────────────────

impl AutoWorkerContext {
    fn discovery_handler(&self, key: &AutoWorkerKey) -> DiscoveryHandler {
        DiscoveryHandler {
            group_id: self.group_id.clone(),
            shared: self.shared.clone(),
            name: self.name.clone(),
            ifname: key.ifname.clone(),
            if_index: key.if_index,
            data_port: self.data_port,
            configured_bitrate: self.configured_bitrate,
            ingress_control: self.ingress_control,
            runtime: self.runtime.clone(),
            dynamic_template: self.dynamic_template.clone(),
            underlay_mark: self.underlay_mark,
            inline: crate::engine::FOLD_IO,
        }
    }
}

#[derive(Clone)]
struct AutoWorkerContext {
    group_id: Vec<u8>,
    mcast_ip: Ipv6Addr,
    discovery_port: u16,
    unicast_discovery_port: u16,
    data_port: u16,
    name: String,
    configured_bitrate: u64,
    ingress_control: rns_core::transport::types::IngressControlConfig,
    runtime: Arc<Mutex<AutoRuntime>>,
    shared: Arc<Mutex<SharedState>>,
    tx: EventSender,
    dynamic_template: Option<super::DynamicInterfaceTemplate>,
    underlay_mark: Option<u32>,
}

/// Start an AutoInterface. Discovers local IPv6 link-local interfaces,
/// sets up multicast discovery, and creates UDP data servers.
///
/// Returns a vec of (InterfaceId, Writer) for each initial peer (typically empty
/// since peers are discovered dynamically via InterfaceUp events).
pub fn start(
    config: AutoConfig,
    tx: EventSender,
    next_dynamic_id: Arc<AtomicU64>,
) -> io::Result<()> {
    start_with_template(config, tx, next_dynamic_id, None, None).map(|_| ())
}

fn start_with_template(
    config: AutoConfig,
    tx: EventSender,
    next_dynamic_id: Arc<AtomicU64>,
    dynamic_template: Option<super::DynamicInterfaceTemplate>,
    underlay_mark: Option<u32>,
) -> io::Result<super::ListenerControl> {
    let group_id = config.group_id.clone();
    let mcast_addr_str = derive_multicast_address(
        &group_id,
        &config.multicast_address_type,
        &config.discovery_scope,
    );

    let mcast_ip = match parse_multicast_addr(&mcast_addr_str) {
        Some(ip) => ip,
        None => {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                format!("Invalid multicast address: {}", mcast_addr_str),
            ));
        }
    };

    {
        let startup = AutoRuntime::from_config(&config);
        *lock_or_recover(&config.runtime, "auto runtime") = startup;
    }
    let runtime = Arc::clone(&config.runtime);
    let mut shared_state = SharedState::new(next_dynamic_id);
    shared_state.max_peers = config.max_peers;
    let shared = Arc::new(Mutex::new(shared_state));
    let running = Arc::new(AtomicBool::new(true));

    let worker_context = AutoWorkerContext {
        group_id,
        mcast_ip,
        discovery_port: config.discovery_port,
        unicast_discovery_port: config.discovery_port + 1,
        data_port: config.data_port,
        name: config.name.clone(),
        configured_bitrate: config.configured_bitrate,
        ingress_control: config.ingress_control,
        runtime: runtime.clone(),
        shared: shared.clone(),
        tx: tx.clone(),
        dynamic_template,
        underlay_mark,
    };

    if crate::engine::FOLD_IO {
        folded::start(
            config.interface_id,
            worker_context,
            config.allowed_interfaces.clone(),
            config.ignored_interfaces.clone(),
            running.clone(),
        )?;
    } else {
        start_threads(&config, worker_context, &shared, &runtime, &running)?;
    }

    let announce_interval = lock_or_recover(&runtime, "auto runtime").announce_interval_secs;
    let peering_wait = Duration::from_secs_f64(announce_interval * 1.2);
    thread::sleep(peering_wait);

    {
        let mut state = lock_or_recover(&shared, "auto shared state");
        state.online = true;
    }

    log::info!("[{}] AutoInterface online", config.name);

    Ok(super::ListenerControl::with_running_flag(running))
}

/// Run an AutoInterface on threads of its own: a supervisor that adopts
/// network interfaces (each served by its own threads) and a peer-jobs
/// thread that culls silent peers.
fn start_threads(
    config: &AutoConfig,
    worker_context: AutoWorkerContext,
    shared: &Arc<Mutex<SharedState>>,
    runtime: &Arc<Mutex<AutoRuntime>>,
    running: &Arc<AtomicBool>,
) -> io::Result<()> {
    {
        let shared = shared.clone();
        let tx = worker_context.tx.clone();
        let running = running.clone();
        let name = config.name.clone();
        let runtime = runtime.clone();

        thread::Builder::new()
            .name(format!("auto-peer-jobs-{}", name))
            .spawn(move || {
                peer_jobs_loop(shared, tx, runtime, &running, &name);
            })?;
    }

    {
        let allowed = config.allowed_interfaces.clone();
        let ignored = config.ignored_interfaces.clone();
        let running = running.clone();
        let name = config.name.clone();
        thread::Builder::new()
            .name(format!("auto-supervisor-{}", name))
            .spawn(move || {
                auto_supervisor_loop(allowed, ignored, worker_context, running, &name);
            })?;
    }

    Ok(())
}

fn auto_supervisor_loop(
    allowed_interfaces: Vec<String>,
    ignored_interfaces: Vec<String>,
    context: AutoWorkerContext,
    running: Arc<AtomicBool>,
    name: &str,
) {
    let mut workers: HashMap<AutoWorkerKey, AutoWorkerEntry> = HashMap::new();

    while running.load(Ordering::Relaxed) {
        let interfaces = enumerate_interfaces(&allowed_interfaces, &ignored_interfaces);
        let desired = interfaces
            .iter()
            .map(AutoWorkerKey::from_local_interface)
            .collect::<Vec<_>>();

        {
            let mut state = lock_or_recover(&context.shared, "auto shared state");
            state.link_local_addresses = desired
                .iter()
                .map(|key| PeerKey {
                    link_local_addr: key.link_local_addr.clone(),
                    if_index: key.if_index,
                })
                .collect();
        }

        let removed_workers = reconcile_worker_keys(workers.keys().cloned(), desired.clone())
            .into_iter()
            .filter_map(|action| match action {
                WorkerReconcileAction::Remove(key) => Some(key),
                WorkerReconcileAction::Add(_) => None,
            })
            .collect::<Vec<_>>();
        for key in &removed_workers {
            remove_peers_for_worker(&context.shared, &context.tx, key);
        }

        if desired.is_empty() && workers.is_empty() {
            log::warn!("[{}] No suitable IPv6 link-local interfaces found", name);
        }

        reconcile_auto_workers(&mut workers, desired, |key| {
            start_auto_worker(key, &context)
        });

        thread::sleep(Duration::from_secs_f64(AUTO_RESCAN_INTERVAL));
    }

    for (key, worker) in workers {
        remove_peers_for_worker(&context.shared, &context.tx, &key);
        worker.stop();
    }
}

fn start_auto_worker(
    key: &AutoWorkerKey,
    context: &AutoWorkerContext,
) -> io::Result<RunningAutoWorker> {
    let underlay_mark = context.underlay_mark;
    let mcast_socket = create_multicast_recv_socket(
        &context.mcast_ip,
        context.discovery_port,
        key.if_index,
        context.underlay_mark,
    )?;
    let unicast_socket = create_unicast_recv_socket(
        &key.link_local_addr,
        context.unicast_discovery_port,
        key.if_index,
        context.underlay_mark,
    )?;
    let data_socket = create_data_recv_socket(
        &key.link_local_addr,
        context.data_port,
        key.if_index,
        context.underlay_mark,
    )?;

    let worker_running = Arc::new(AtomicBool::new(true));

    {
        let group_id = context.group_id.clone();
        let link_local = key.link_local_addr.clone();
        let running = worker_running.clone();
        let name = context.name.clone();
        let runtime = context.runtime.clone();
        let mcast_ip = context.mcast_ip;
        let discovery_port = context.discovery_port;
        let if_index = key.if_index;
        thread::Builder::new()
            .name(format!("auto-disc-tx-{}", key.ifname))
            .spawn(move || {
                discovery_sender_loop(DiscoverySenderContext {
                    group_id,
                    link_local_addr: link_local,
                    mcast_ip,
                    discovery_port,
                    if_index,
                    runtime,
                    running,
                    name,
                    underlay_mark,
                });
            })?;
    }

    let handler = context.discovery_handler(key);
    for (socket, thread_name) in [
        (mcast_socket, format!("auto-disc-rx-{}", key.ifname)),
        (unicast_socket, format!("auto-udisc-rx-{}", key.ifname)),
    ] {
        let handler = handler.clone();
        let tx = context.tx.clone();
        let running = worker_running.clone();
        thread::Builder::new()
            .name(thread_name)
            .spawn(move || discovery_receiver_loop(socket, handler, tx, running))?;
    }

    {
        let shared = context.shared.clone();
        let tx = context.tx.clone();
        let running = worker_running.clone();
        let name = context.name.clone();
        let if_index = key.if_index;
        thread::Builder::new()
            .name(format!("auto-data-rx-{}", key.ifname))
            .spawn(move || {
                data_receiver_loop(data_socket, shared, tx, &running, &name, if_index);
            })?;
    }

    log::info!(
        "[{}] AutoInterface worker online on {} {}%{}",
        context.name,
        key.ifname,
        key.link_local_addr,
        key.if_index
    );

    Ok(RunningAutoWorker {
        running: worker_running,
    })
}

// ── Socket creation helpers ────────────────────────────────────────────────

fn create_multicast_recv_socket(
    mcast_ip: &Ipv6Addr,
    port: u16,
    if_index: u32,
    underlay_mark: Option<u32>,
) -> io::Result<UdpSocket> {
    let socket = socket2::Socket::new(
        socket2::Domain::IPV6,
        socket2::Type::DGRAM,
        Some(socket2::Protocol::UDP),
    )?;
    super::apply_underlay_mark(socket.as_raw_fd(), underlay_mark)?;

    socket.set_reuse_address(true)?;
    #[cfg(not(target_os = "windows"))]
    socket.set_reuse_port(true)?;

    // Bind to [::]:port on the specific interface
    let bind_addr = SocketAddrV6::new(Ipv6Addr::UNSPECIFIED, port, 0, if_index);
    socket.bind(&bind_addr.into())?;

    // Join multicast group on the specific interface
    socket.join_multicast_v6(mcast_ip, if_index)?;

    socket.set_nonblocking(false)?;
    let std_socket: UdpSocket = socket.into();
    std_socket.set_read_timeout(Some(Duration::from_secs(2)))?;
    Ok(std_socket)
}

fn create_unicast_recv_socket(
    link_local: &str,
    port: u16,
    if_index: u32,
    underlay_mark: Option<u32>,
) -> io::Result<UdpSocket> {
    let ip: Ipv6Addr = link_local
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad IPv6: {}", e)))?;

    let socket = socket2::Socket::new(
        socket2::Domain::IPV6,
        socket2::Type::DGRAM,
        Some(socket2::Protocol::UDP),
    )?;
    super::apply_underlay_mark(socket.as_raw_fd(), underlay_mark)?;

    socket.set_reuse_address(true)?;
    #[cfg(not(target_os = "windows"))]
    socket.set_reuse_port(true)?;

    let bind_addr = SocketAddrV6::new(ip, port, 0, if_index);
    socket.bind(&bind_addr.into())?;

    socket.set_nonblocking(false)?;
    let std_socket: UdpSocket = socket.into();
    std_socket.set_read_timeout(Some(Duration::from_secs(2)))?;
    Ok(std_socket)
}

fn create_data_recv_socket(
    link_local: &str,
    port: u16,
    if_index: u32,
    underlay_mark: Option<u32>,
) -> io::Result<UdpSocket> {
    let ip: Ipv6Addr = link_local
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad IPv6: {}", e)))?;

    let socket = socket2::Socket::new(
        socket2::Domain::IPV6,
        socket2::Type::DGRAM,
        Some(socket2::Protocol::UDP),
    )?;
    super::apply_underlay_mark(socket.as_raw_fd(), underlay_mark)?;

    socket.set_reuse_address(true)?;
    #[cfg(not(target_os = "windows"))]
    socket.set_reuse_port(true)?;

    let bind_addr = SocketAddrV6::new(ip, port, 0, if_index);
    socket.bind(&bind_addr.into())?;

    socket.set_nonblocking(false)?;
    let std_socket: UdpSocket = socket.into();
    std_socket.set_read_timeout(Some(Duration::from_secs(2)))?;
    Ok(std_socket)
}

// ── Thread loops ───────────────────────────────────────────────────────────

/// Discovery sender: periodically sends discovery token via multicast.
struct DiscoverySenderContext {
    group_id: Vec<u8>,
    link_local_addr: String,
    mcast_ip: Ipv6Addr,
    discovery_port: u16,
    if_index: u32,
    runtime: Arc<Mutex<AutoRuntime>>,
    running: Arc<AtomicBool>,
    name: String,
    underlay_mark: Option<u32>,
}

fn discovery_sender_loop(context: DiscoverySenderContext) {
    let DiscoverySenderContext {
        group_id,
        link_local_addr,
        mcast_ip,
        discovery_port,
        if_index,
        runtime,
        running,
        name,
        underlay_mark,
    } = context;
    let token = compute_discovery_token(&group_id, &link_local_addr);

    while running.load(Ordering::Relaxed) {
        // Create a fresh socket for each send (matches Python)
        if let Ok(socket) = UdpSocket::bind("[::]:0") {
            if let Err(error) = super::apply_underlay_mark(socket.as_raw_fd(), underlay_mark) {
                log::error!("[{}] failed to mark discovery socket: {}", name, error);
                return;
            }
            // Set multicast interface
            let if_bytes = if_index.to_ne_bytes();
            unsafe {
                libc::setsockopt(
                    socket_fd(&socket),
                    libc::IPPROTO_IPV6,
                    libc::IPV6_MULTICAST_IF,
                    if_bytes.as_ptr() as *const libc::c_void,
                    4,
                );
            }

            let target = SocketAddrV6::new(mcast_ip, discovery_port, 0, 0);
            if let Err(e) = socket.send_to(&token, target) {
                log::debug!("[{}] multicast send error: {}", name, e);
            }
        }

        let sleep_dur = Duration::from_secs_f64(
            lock_or_recover(&runtime, "auto runtime")
                .announce_interval_secs
                .max(0.1),
        );
        thread::sleep(sleep_dur);
    }
}

/// What handling a discovery token needs: validating it and admitting the
/// peer that sent it.
#[derive(Clone)]
struct DiscoveryHandler {
    group_id: Vec<u8>,
    shared: Arc<Mutex<SharedState>>,
    name: String,
    ifname: String,
    if_index: u32,
    data_port: u16,
    configured_bitrate: u64,
    ingress_control: rns_core::transport::types::IngressControlConfig,
    runtime: Arc<Mutex<AutoRuntime>>,
    dynamic_template: Option<super::DynamicInterfaceTemplate>,
    underlay_mark: Option<u32>,
    /// Whether new peers' writers are called by the driver directly.
    inline: bool,
}

impl DiscoveryHandler {
    /// Handle one discovery datagram, returning the event that brings up
    /// a newly admitted peer.
    fn handle(&self, datagram: &[u8], src: std::net::SocketAddr) -> Option<Event> {
        if datagram.len() < 32 {
            return None;
        }
        // Extract source IPv6 address
        let std::net::SocketAddr::V6(src_addr) = src else {
            return None;
        };
        let peer_key = peer_key_from_src(&src_addr, self.if_index);
        let peering_hash = &datagram[..32];
        let expected = compute_discovery_token(&self.group_id, &peer_key.link_local_addr);
        if peering_hash != expected {
            log::debug!(
                "[{}] invalid peering hash from {}",
                self.name,
                peer_key.link_local_addr
            );
            return None;
        }

        {
            // Peers are accepted before the interface is fully online, for
            // initial peering (Python processes after final_init_done).
            let mut state = lock_or_recover(&self.shared, "auto shared state");
            // Multicast echo from ourselves.
            if state.link_local_addresses.contains(&peer_key) {
                return None;
            }
            // Already known.
            if state.peers.contains_key(&peer_key) {
                state.refresh_peer(&peer_key, crate::time::now());
                return None;
            }
        }

        // New peer! Create a data writer to send to them.
        add_peer(AddPeerContext {
            shared: &self.shared,
            peer_key: &peer_key,
            data_port: self.data_port,
            name: &self.name,
            ifname: &self.ifname,
            configured_bitrate: self.configured_bitrate,
            ingress_control: self.ingress_control,
            runtime: &self.runtime,
            dynamic_template: self.dynamic_template.as_ref(),
            underlay_mark: self.underlay_mark,
            inline: self.inline,
        })
    }
}

/// Discovery receiver: listens for discovery tokens and adds peers.
fn discovery_receiver_loop(
    socket: UdpSocket,
    handler: DiscoveryHandler,
    tx: EventSender,
    running: Arc<AtomicBool>,
) {
    let mut buf = [0u8; 1024];

    while running.load(Ordering::Relaxed) {
        match socket.recv_from(&mut buf) {
            Ok((n, src)) => {
                if let Some(event) = handler.handle(&buf[..n], src) {
                    let _ = tx.send(event);
                }
            }
            Err(ref e)
                if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut =>
            {
                // Timeout, loop again
                continue;
            }
            Err(e) => {
                log::warn!("[{}] discovery recv error: {}", handler.name, e);
                if !running.load(Ordering::Relaxed) {
                    return;
                }
                thread::sleep(Duration::from_millis(100));
            }
        }
    }
}

/// Add a new peer, creating a writer and emitting InterfaceUp.
struct AddPeerContext<'a> {
    shared: &'a Arc<Mutex<SharedState>>,
    peer_key: &'a PeerKey,
    data_port: u16,
    name: &'a str,
    ifname: &'a str,
    configured_bitrate: u64,
    ingress_control: rns_core::transport::types::IngressControlConfig,
    runtime: &'a Arc<Mutex<AutoRuntime>>,
    dynamic_template: Option<&'a super::DynamicInterfaceTemplate>,
    underlay_mark: Option<u32>,
    inline: bool,
}

/// Admit a new peer, returning the event that brings up its interface.
fn add_peer(context: AddPeerContext<'_>) -> Option<Event> {
    let AddPeerContext {
        shared,
        peer_key,
        data_port,
        name,
        ifname,
        configured_bitrate,
        ingress_control,
        runtime: _runtime,
        dynamic_template,
        underlay_mark,
        inline,
    } = context;

    {
        let mut state = lock_or_recover(shared, "auto shared state");
        if state.peers.contains_key(peer_key) {
            state.refresh_peer(peer_key, crate::time::now());
            return None;
        }
        let now = Instant::now();
        let decision = state.admit_new_peer(ifname, now);
        if decision != PeerAdmissionDecision::Admit {
            let report = state.note_peer_refusal(ifname, now);
            let max_peers = state.max_peers;
            drop(state);
            log_peer_refusal(name, ifname, peer_key, decision, max_peers, report);
            return None;
        }
    }

    // Create UDP writer to send data to this peer
    let send_socket = match UdpSocket::bind("[::]:0") {
        Ok(s) => s,
        Err(e) => {
            log::warn!(
                "[{}] failed to create writer for peer {}: {}",
                name,
                peer_key.link_local_addr,
                e
            );
            return None;
        }
    };
    if let Err(error) = super::apply_underlay_mark(send_socket.as_raw_fd(), underlay_mark) {
        log::error!("[{}] failed to mark peer socket: {}", name, error);
        return None;
    }
    if inline {
        if let Err(error) = send_socket.set_nonblocking(true) {
            log::warn!("[{}] failed to set up peer socket: {}", name, error);
            return None;
        }
    }

    let target = match peer_target_addr(peer_key, data_port) {
        Ok(target) => target,
        Err(e) => {
            log::warn!(
                "[{}] failed to create target for peer {}%{}: {}",
                name,
                peer_key.link_local_addr,
                peer_key.if_index,
                e
            );
            return None;
        }
    };

    let mut state = lock_or_recover(shared, "auto shared state");

    // Double-check not already added (race)
    if state.peers.contains_key(peer_key) {
        state.refresh_peer(peer_key, crate::time::now());
        return None;
    }
    // Both discovery receivers of a worker may admit concurrently; recheck the cap.
    if state.active_peers_on(ifname) >= state.max_peers {
        let report = state.note_peer_refusal(ifname, Instant::now());
        let max_peers = state.max_peers;
        drop(state);
        log_peer_refusal(
            name,
            ifname,
            peer_key,
            PeerAdmissionDecision::AtCapacity,
            max_peers,
            report,
        );
        return None;
    }

    let peer_id = InterfaceId(state.next_id.fetch_add(1, Ordering::Relaxed));

    // Create a boxed writer for the driver
    let driver_writer: Box<dyn Writer> = Box::new(UdpWriter {
        socket: send_socket,
        target,
        inline,
    });

    let peer_info = rns_core::transport::types::InterfaceInfo {
        stamp_costs: Default::default(),
        id: peer_id,
        name: format!(
            "{}:{}%{}",
            name, peer_key.link_local_addr, peer_key.if_index
        ),
        mode: rns_core::constants::MODE_FULL,
        gravity: 0,
        recursive_prs: false,
        announces_from_internal: true,
        announces_to_internal: None,
        out_capable: true,
        in_capable: true,
        bitrate: Some(configured_bitrate),
        airtime_profile: None,
        announce_rate_target: None,
        announce_rate_grace: 0,
        announce_rate_penalty: 0.0,
        announce_cap: rns_core::constants::ANNOUNCE_CAP,
        is_local_client: false,
        wants_tunnel: false,
        tunnel_id: None,
        mtu: HW_MTU as u32,
        ia_freq: 0.0,
        ip_freq: 0.0,
        op_freq: 0.0,
        op_samples: 0,
        started: 0.0,
        ingress_control,
    };

    let now = crate::time::now();
    state.peers.insert(
        peer_key.clone(),
        AutoPeer {
            interface_id: peer_id,
            link_local_addr: peer_key.link_local_addr.clone(),
            ifname: ifname.to_string(),
            if_index: peer_key.if_index,
            last_heard: now,
        },
    );

    log::info!(
        "[{}] Peer discovered: {} (id={})",
        name,
        peer_key.link_local_addr,
        peer_id.0
    );

    // Notify driver of new dynamic interface
    let event = if let Some(template) = dynamic_template {
        Event::DynamicInterfaceUp {
            id: peer_id,
            writer: driver_writer,
            registration: Box::new(template.registration(peer_info)),
        }
    } else {
        Event::InterfaceUp(peer_id, Some(driver_writer), Some(Box::new(peer_info)))
    };
    Some(event)
}

/// Warn about an ignored discovery source when `report` says a warning is due.
fn log_peer_refusal(
    name: &str,
    ifname: &str,
    peer_key: &PeerKey,
    decision: PeerAdmissionDecision,
    max_peers: usize,
    report: Option<usize>,
) {
    let Some(ignored) = report else {
        return;
    };
    let reason = match decision {
        PeerAdmissionDecision::AtCapacity => format!("{} active peers is the limit", max_peers),
        PeerAdmissionDecision::RateLimited => format!(
            "at most {} new peers per {}s",
            MAX_NEW_PEERS_PER_WINDOW,
            PEER_ADMISSION_WINDOW.as_secs()
        ),
        PeerAdmissionDecision::Admit => return,
    };
    log::warn!(
        "[{}] ignoring new peer {}%{} on {} ({}); {} new peer(s) ignored since the last warning",
        name,
        peer_key.link_local_addr,
        peer_key.if_index,
        ifname,
        reason,
        ignored
    );
}

/// Handle one datagram on a data socket, returning the frame it carries
/// for its peer's interface, or `None` if it is dropped.
fn handle_data(
    shared: &Mutex<SharedState>,
    if_index: u32,
    data: &[u8],
    src: std::net::SocketAddr,
) -> Option<Event> {
    if data.is_empty() {
        return None;
    }
    let std::net::SocketAddr::V6(src_addr) = src else {
        return None;
    };
    let peer_key = peer_key_from_src(&src_addr, if_index);

    // Only data from known peers is considered (as in Python, which looks
    // up the spawned peer before deduplicating).
    let mut state = lock_or_recover(shared, "auto shared state");
    if !state.online || !state.peers.contains_key(&peer_key) {
        return None;
    }
    let now = crate::time::now();
    let data_hash = rns_crypto::sha256::sha256(data);
    // Deduplication
    if state.is_duplicate(&data_hash, now) {
        return None;
    }
    state.add_dedup(data_hash, now);
    state.refresh_peer(&peer_key, now);
    let iface_id = state.peers.get(&peer_key)?.interface_id;
    Some(Event::Frame {
        interface_id: iface_id,
        data: data.into(),
        rssi: None,
        snr: None,
    })
}

/// Data receiver: receives unicast UDP data from peers and dispatches as frames.
fn data_receiver_loop(
    socket: UdpSocket,
    shared: Arc<Mutex<SharedState>>,
    tx: EventSender,
    running: &AtomicBool,
    name: &str,
    if_index: u32,
) {
    let mut buf = [0u8; HW_MTU + 64]; // a bit extra

    while running.load(Ordering::Relaxed) {
        match socket.recv_from(&mut buf) {
            Ok((n, src)) => {
                if let Some(frame) = handle_data(&shared, if_index, &buf[..n], src) {
                    if tx.send(frame).is_err() {
                        return;
                    }
                }
            }
            Err(ref e)
                if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut =>
            {
                continue;
            }
            Err(e) => {
                log::warn!("[{}] data recv error: {}", name, e);
                if !running.load(Ordering::Relaxed) {
                    return;
                }
                thread::sleep(Duration::from_millis(100));
            }
        }
    }
}

/// Peer jobs: periodically cull timed-out peers.
fn peer_jobs_loop(
    shared: Arc<Mutex<SharedState>>,
    tx: EventSender,
    runtime: Arc<Mutex<AutoRuntime>>,
    running: &AtomicBool,
    name: &str,
) {
    while running.load(Ordering::Relaxed) {
        let interval = Duration::from_secs_f64(
            lock_or_recover(&runtime, "auto runtime")
                .peer_job_interval_secs
                .max(0.1),
        );
        thread::sleep(interval);

        let now = crate::time::now();
        let peer_timeout_secs = lock_or_recover(&runtime, "auto runtime").peer_timeout_secs;

        let timed_out = lock_or_recover(&shared, "auto shared state")
            .remove_timed_out_peers(now, peer_timeout_secs);

        for (peer_key, iface_id) in &timed_out {
            log::info!(
                "[{}] Peer timed out: {}%{}",
                name,
                peer_key.link_local_addr,
                peer_key.if_index
            );
            let _ = tx.send(Event::InterfaceDown(*iface_id));
        }
    }
}

// ── Helper ─────────────────────────────────────────────────────────────────

/// Get the raw file descriptor from a UdpSocket (for setsockopt).
#[cfg(unix)]
fn socket_fd(socket: &UdpSocket) -> i32 {
    use std::os::unix::io::AsRawFd;
    socket.as_raw_fd()
}

#[cfg(not(unix))]
fn socket_fd(_socket: &UdpSocket) -> i32 {
    0
}

// ── Factory implementation ─────────────────────────────────────────────────

use super::{InterfaceConfigData, InterfaceFactory, StartContext, StartResult};

/// Factory for `AutoInterface`.
pub struct AutoFactory;

impl InterfaceFactory for AutoFactory {
    fn type_name(&self) -> &str {
        "AutoInterface"
    }

    fn parse_config(
        &self,
        name: &str,
        id: InterfaceId,
        params: &HashMap<String, String>,
    ) -> Result<Box<dyn InterfaceConfigData>, String> {
        let group_id = params
            .get("group_id")
            .map(|v| v.as_bytes().to_vec())
            .unwrap_or_else(|| DEFAULT_GROUP_ID.to_vec());

        let discovery_scope = params
            .get("discovery_scope")
            .map(|v| match v.to_lowercase().as_str() {
                "link" => SCOPE_LINK.to_string(),
                "admin" => SCOPE_ADMIN.to_string(),
                "site" => SCOPE_SITE.to_string(),
                "organisation" | "organization" => SCOPE_ORGANISATION.to_string(),
                "global" => SCOPE_GLOBAL.to_string(),
                _ => v.clone(),
            })
            .unwrap_or_else(|| SCOPE_LINK.to_string());

        let discovery_port = params
            .get("discovery_port")
            .and_then(|v| v.parse().ok())
            .unwrap_or(DEFAULT_DISCOVERY_PORT);

        let data_port = params
            .get("data_port")
            .and_then(|v| v.parse().ok())
            .unwrap_or(DEFAULT_DATA_PORT);

        let multicast_address_type = params
            .get("multicast_address_type")
            .map(|v| match v.to_lowercase().as_str() {
                "permanent" => MULTICAST_PERMANENT_ADDRESS_TYPE.to_string(),
                "temporary" => MULTICAST_TEMPORARY_ADDRESS_TYPE.to_string(),
                _ => v.clone(),
            })
            .unwrap_or_else(|| MULTICAST_TEMPORARY_ADDRESS_TYPE.to_string());

        let configured_bitrate = params
            .get("configured_bitrate")
            .or_else(|| params.get("bitrate"))
            .and_then(|v| v.parse().ok())
            .unwrap_or(BITRATE_GUESS);

        let allowed_interfaces = params
            .get("devices")
            .or_else(|| params.get("allowed_interfaces"))
            .map(|v| {
                v.split(',')
                    .map(|s| s.trim().to_string())
                    .filter(|s| !s.is_empty())
                    .collect()
            })
            .unwrap_or_default();

        let ignored_interfaces = params
            .get("ignored_devices")
            .or_else(|| params.get("ignored_interfaces"))
            .map(|v| {
                v.split(',')
                    .map(|s| s.trim().to_string())
                    .filter(|s| !s.is_empty())
                    .collect()
            })
            .unwrap_or_default();

        let max_peers = params
            .get("max_peers")
            .and_then(|v| v.parse::<usize>().ok())
            .filter(|max_peers| *max_peers > 0)
            .unwrap_or(DEFAULT_MAX_PEERS);

        Ok(Box::new(AutoConfig {
            name: name.to_string(),
            group_id,
            discovery_scope,
            discovery_port,
            data_port,
            multicast_address_type,
            allowed_interfaces,
            ignored_interfaces,
            configured_bitrate,
            max_peers,
            interface_id: id,
            ingress_control: rns_core::transport::types::IngressControlConfig::enabled(),
            runtime: Arc::new(Mutex::new(AutoRuntime {
                announce_interval_secs: ANNOUNCE_INTERVAL,
                peer_timeout_secs: PEERING_TIMEOUT,
                peer_job_interval_secs: PEER_JOB_INTERVAL,
            })),
        }))
    }

    fn start(
        &self,
        config: Box<dyn InterfaceConfigData>,
        ctx: StartContext,
    ) -> std::io::Result<StartResult> {
        let mut auto_config = *config.into_any().downcast::<AutoConfig>().map_err(|_| {
            std::io::Error::new(std::io::ErrorKind::InvalidData, "wrong config type")
        })?;

        auto_config.ingress_control = ctx.ingress_control;
        let parent_id = auto_config.interface_id;
        let control = start_with_template(
            auto_config,
            ctx.tx,
            ctx.next_dynamic_id,
            Some(super::DynamicInterfaceTemplate {
                parent_id,
                interface_type: "AutoInterfacePeer".into(),
                ifac: ctx.ifac,
                mode: ctx.mode,
                gravity: ctx.gravity,
                recursive_prs: ctx.recursive_prs,
                announces_from_internal: ctx.announces_from_internal,
                announces_to_internal: ctx.announces_to_internal,
                stamp_costs: ctx.stamp_costs,
            }),
            ctx.underlay_mark,
        )?;
        Ok(StartResult::Listener {
            control: Some(control),
        })
    }
}

pub(crate) fn auto_runtime_handle_from_config(config: &AutoConfig) -> AutoRuntimeConfigHandle {
    AutoRuntimeConfigHandle {
        interface_name: config.name.clone(),
        runtime: Arc::clone(&config.runtime),
        startup: AutoRuntime::from_config(config),
    }
}

// ── Tests ──────────────────────────────────────────────────────────────────

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

    #[test]
    fn listener_control_stops_auto_supervisor() {
        let running = Arc::new(AtomicBool::new(true));
        let control = super::super::ListenerControl::with_running_flag(Arc::clone(&running));

        control.request_stop();

        assert!(!running.load(Ordering::Relaxed));
        assert!(control.should_stop());
    }

    // ── Multicast address derivation ──────────────────────────────────

    #[test]
    fn multicast_address_default_group() {
        // Python vector: ff12:0:d70b:fb1c:16e4:5e39:485e:31e1
        let addr = derive_multicast_address(
            DEFAULT_GROUP_ID,
            MULTICAST_TEMPORARY_ADDRESS_TYPE,
            SCOPE_LINK,
        );
        assert_eq!(addr, "ff12:0:d70b:fb1c:16e4:5e39:485e:31e1");
    }

    #[test]
    fn multicast_address_custom_group() {
        let addr =
            derive_multicast_address(b"testgroup", MULTICAST_TEMPORARY_ADDRESS_TYPE, SCOPE_LINK);
        // Just verify format
        assert!(addr.starts_with("ff12:0:"));
        // Must be different from default
        assert_ne!(addr, "ff12:0:d70b:fb1c:16e4:5e39:485e:31e1");
    }

    #[test]
    fn multicast_address_scope_admin() {
        let addr = derive_multicast_address(
            DEFAULT_GROUP_ID,
            MULTICAST_TEMPORARY_ADDRESS_TYPE,
            SCOPE_ADMIN,
        );
        assert!(addr.starts_with("ff14:0:"));
    }

    #[test]
    fn multicast_address_permanent_type() {
        let addr = derive_multicast_address(
            DEFAULT_GROUP_ID,
            MULTICAST_PERMANENT_ADDRESS_TYPE,
            SCOPE_LINK,
        );
        assert!(addr.starts_with("ff02:0:"));
    }

    #[test]
    fn multicast_address_parseable() {
        let addr = derive_multicast_address(
            DEFAULT_GROUP_ID,
            MULTICAST_TEMPORARY_ADDRESS_TYPE,
            SCOPE_LINK,
        );
        let ip = parse_multicast_addr(&addr);
        assert!(ip.is_some());
        assert!(ip.unwrap().is_multicast());
    }

    // ── Discovery token ──────────────────────────────────────────────

    #[test]
    fn discovery_token_interop() {
        // Python vector: fe80::1 → 97b25576749ea936b0d8a8536ffaf442d157cf47d460dcf13c48b7bd18b6c163
        let token = compute_discovery_token(DEFAULT_GROUP_ID, "fe80::1");
        let expected = "97b25576749ea936b0d8a8536ffaf442d157cf47d460dcf13c48b7bd18b6c163";
        let got = token
            .iter()
            .map(|b| format!("{:02x}", b))
            .collect::<String>();
        assert_eq!(got, expected);
    }

    #[test]
    fn discovery_token_interop_2() {
        // Python vector: fe80::dead:beef:1234:5678
        let token = compute_discovery_token(DEFAULT_GROUP_ID, "fe80::dead:beef:1234:5678");
        let expected = "46b6ec7595504b6a35f06bd4bfff71567fb82fcf2706cd361bab20409c42d072";
        let got = token
            .iter()
            .map(|b| format!("{:02x}", b))
            .collect::<String>();
        assert_eq!(got, expected);
    }

    #[test]
    fn discovery_token_different_groups() {
        let t1 = compute_discovery_token(b"reticulum", "fe80::1");
        let t2 = compute_discovery_token(b"othergroup", "fe80::1");
        assert_ne!(t1, t2);
    }

    #[test]
    fn discovery_token_different_addrs() {
        let t1 = compute_discovery_token(DEFAULT_GROUP_ID, "fe80::1");
        let t2 = compute_discovery_token(DEFAULT_GROUP_ID, "fe80::2");
        assert_ne!(t1, t2);
    }

    // ── Deduplication ────────────────────────────────────────────────

    #[test]
    fn dedup_basic() {
        let next_id = Arc::new(AtomicU64::new(1));
        let mut state = SharedState::new(next_id);

        let hash = [0xAA; 32];
        let now = 1000.0;

        assert!(!state.is_duplicate(&hash, now));
        state.add_dedup(hash, now);
        assert!(state.is_duplicate(&hash, now));
    }

    #[test]
    fn dedup_expired() {
        let next_id = Arc::new(AtomicU64::new(1));
        let mut state = SharedState::new(next_id);

        let hash = [0xBB; 32];
        state.add_dedup(hash, 1000.0);

        // Within TTL
        assert!(state.is_duplicate(&hash, 1000.5));
        // Expired
        assert!(!state.is_duplicate(&hash, 1001.0));
    }

    #[test]
    fn dedup_max_length() {
        let next_id = Arc::new(AtomicU64::new(1));
        let mut state = SharedState::new(next_id);

        // Fill beyond max
        for i in 0..MULTI_IF_DEQUE_LEN + 10 {
            let mut hash = [0u8; 32];
            hash[0] = (i & 0xFF) as u8;
            hash[1] = ((i >> 8) & 0xFF) as u8;
            state.add_dedup(hash, 1000.0);
        }

        assert_eq!(state.dedup_deque.len(), MULTI_IF_DEQUE_LEN);
    }

    // ── Peer tracking ────────────────────────────────────────────────

    #[test]
    fn peer_refresh() {
        let next_id = Arc::new(AtomicU64::new(100));
        let mut state = SharedState::new(next_id);
        let key = peer_key("fe80::1", 2);

        state.peers.insert(
            key.clone(),
            AutoPeer {
                interface_id: InterfaceId(100),
                link_local_addr: key.link_local_addr.clone(),
                ifname: "eth0".to_string(),
                if_index: key.if_index,
                last_heard: 1000.0,
            },
        );

        state.refresh_peer(&key, 2000.0);
        assert_eq!(state.peers[&key].last_heard, 2000.0);
    }

    #[test]
    fn peer_not_found_refresh() {
        let next_id = Arc::new(AtomicU64::new(100));
        let mut state = SharedState::new(next_id);
        // Should not panic
        state.refresh_peer(&peer_key("fe80::999", 4), 1000.0);
    }

    fn peer_key(addr: &str, if_index: u32) -> PeerKey {
        PeerKey {
            link_local_addr: addr.to_string(),
            if_index,
        }
    }

    #[test]
    fn peer_key_distinguishes_same_link_local_on_different_interfaces() {
        let next_id = Arc::new(AtomicU64::new(100));
        let mut state = SharedState::new(next_id);
        let eth = peer_key("fe80::1", 2);
        let wlan = peer_key("fe80::1", 4);

        state.peers.insert(
            eth.clone(),
            AutoPeer {
                interface_id: InterfaceId(100),
                link_local_addr: eth.link_local_addr.clone(),
                ifname: "eth0".to_string(),
                if_index: eth.if_index,
                last_heard: 1000.0,
            },
        );
        state.peers.insert(
            wlan.clone(),
            AutoPeer {
                interface_id: InterfaceId(101),
                link_local_addr: wlan.link_local_addr.clone(),
                ifname: "wlan0".to_string(),
                if_index: wlan.if_index,
                last_heard: 1000.0,
            },
        );

        assert_eq!(state.peers.len(), 2);
        assert_eq!(state.peers[&eth].interface_id, InterfaceId(100));
        assert_eq!(state.peers[&wlan].interface_id, InterfaceId(101));
    }

    #[test]
    fn peer_target_addr_uses_scope_id() {
        let target = peer_target_addr(&peer_key("fe80::1", 7), 42671).unwrap();

        assert_eq!(target.port(), 42671);
        assert_eq!(target.scope_id(), 7);
    }

    #[test]
    fn remove_peers_for_worker_emits_interface_down() {
        let next_id = Arc::new(AtomicU64::new(100));
        let shared = Arc::new(Mutex::new(SharedState::new(next_id)));
        let worker = worker_key("wlan0", 4, "fe80::abcd");
        let removed = peer_key("fe80::1", 4);
        let kept = peer_key("fe80::1", 5);
        let (tx, rx) = crate::event::channel_with_capacity(4);

        {
            let mut state = lock_or_recover(&shared, "test auto shared state");
            state.peers.insert(
                removed.clone(),
                AutoPeer {
                    interface_id: InterfaceId(100),
                    link_local_addr: removed.link_local_addr.clone(),
                    ifname: "wlan0".to_string(),
                    if_index: removed.if_index,
                    last_heard: 1000.0,
                },
            );
            state.peers.insert(
                kept.clone(),
                AutoPeer {
                    interface_id: InterfaceId(101),
                    link_local_addr: kept.link_local_addr.clone(),
                    ifname: "wlan1".to_string(),
                    if_index: kept.if_index,
                    last_heard: 1000.0,
                },
            );
        }

        remove_peers_for_worker(&shared, &tx, &worker);

        let state = lock_or_recover(&shared, "test auto shared state");
        assert!(!state.peers.contains_key(&removed));
        assert!(state.peers.contains_key(&kept));
        drop(state);

        assert!(matches!(
            rx.recv_timeout(std::time::Duration::from_secs(1)).unwrap(),
            Event::InterfaceDown(InterfaceId(100))
        ));
        assert!(rx.try_recv().is_err());
    }

    // ── Network interface enumeration ────────────────────────────────

    #[test]
    fn enumerate_returns_vec() {
        // This test just verifies the function runs without crashing.
        // Results depend on the system's network configuration.
        let interfaces = enumerate_interfaces(&[], &[]);
        // On CI/test machines, we may or may not have IPv6 link-local
        for iface in &interfaces {
            assert!(!iface.name.is_empty());
            assert!(iface.link_local_addr.starts_with("fe80"));
            assert!(iface.index > 0);
        }
    }

    #[test]
    fn enumerate_with_ignored() {
        // Ignore everything
        let interfaces = enumerate_interfaces(
            &[],
            &[
                "lo".to_string(),
                "eth0".to_string(),
                "wlan0".to_string(),
                "enp0s3".to_string(),
                "docker0".to_string(),
            ],
        );
        // May still have some interfaces, but known ones should be filtered
        for iface in &interfaces {
            assert_ne!(iface.name, "lo");
            assert_ne!(iface.name, "eth0");
            assert_ne!(iface.name, "wlan0");
        }
    }

    #[test]
    fn enumerate_with_allowed_nonexistent() {
        // Only allow an interface that doesn't exist
        let interfaces = enumerate_interfaces(&["nonexistent_if_12345".to_string()], &[]);
        assert!(interfaces.is_empty());
    }

    fn worker_key(name: &str, index: u32, addr: &str) -> AutoWorkerKey {
        AutoWorkerKey {
            ifname: name.to_string(),
            if_index: index,
            link_local_addr: addr.to_string(),
        }
    }

    #[test]
    fn reconcile_worker_keys_adds_new_interface() {
        let desired = vec![worker_key("wlan0", 4, "fe80::1")];

        assert_eq!(
            reconcile_worker_keys(Vec::new(), desired.clone()),
            vec![WorkerReconcileAction::Add(desired[0].clone())]
        );
    }

    #[test]
    fn reconcile_worker_keys_keeps_unchanged_interface() {
        let key = worker_key("wlan0", 4, "fe80::1");

        assert!(reconcile_worker_keys(vec![key.clone()], vec![key]).is_empty());
    }

    #[test]
    fn reconcile_worker_keys_removes_missing_interface() {
        let active = vec![worker_key("wlan0", 4, "fe80::1")];

        assert_eq!(
            reconcile_worker_keys(active.clone(), Vec::new()),
            vec![WorkerReconcileAction::Remove(active[0].clone())]
        );
    }

    #[test]
    fn reconcile_worker_keys_replaces_changed_interface_identity() {
        let old = worker_key("wlan0", 4, "fe80::1");
        let new = worker_key("wlan0", 9, "fe80::2");
        let actions = reconcile_worker_keys(vec![old.clone()], vec![new.clone()]);

        assert!(actions.contains(&WorkerReconcileAction::Remove(old)));
        assert!(actions.contains(&WorkerReconcileAction::Add(new)));
        assert_eq!(actions.len(), 2);
    }

    #[test]
    fn reconcile_auto_workers_starts_new_worker() {
        let key = worker_key("wlan0", 4, "fe80::1");
        let mut workers = HashMap::new();
        let mut starts = Vec::new();

        reconcile_auto_workers(&mut workers, vec![key.clone()], |key| {
            starts.push(key.clone());
            Ok(RunningAutoWorker {
                running: Arc::new(AtomicBool::new(true)),
            })
        });

        assert_eq!(starts, vec![key.clone()]);
        assert!(matches!(
            workers.get(&key),
            Some(AutoWorkerEntry::Running(_))
        ));
    }

    #[test]
    fn reconcile_auto_workers_stops_removed_worker() {
        let key = worker_key("wlan0", 4, "fe80::1");
        let running = Arc::new(AtomicBool::new(true));
        let mut workers = HashMap::from([(
            key.clone(),
            AutoWorkerEntry::Running(RunningAutoWorker {
                running: running.clone(),
            }),
        )]);

        reconcile_auto_workers(&mut workers, Vec::new(), |_| unreachable!());

        assert!(!running.load(Ordering::Relaxed));
        assert!(!workers.contains_key(&key));
    }

    #[test]
    fn reconcile_auto_workers_leaves_unchanged_worker_running() {
        let key = worker_key("wlan0", 4, "fe80::1");
        let running = Arc::new(AtomicBool::new(true));
        let mut workers = HashMap::from([(
            key.clone(),
            AutoWorkerEntry::Running(RunningAutoWorker {
                running: running.clone(),
            }),
        )]);
        let mut starts = 0;

        reconcile_auto_workers(&mut workers, vec![key.clone()], |_| {
            starts += 1;
            unreachable!()
        });

        assert_eq!(starts, 0);
        assert!(running.load(Ordering::Relaxed));
        assert!(workers.contains_key(&key));
    }

    #[test]
    fn reconcile_auto_workers_records_pending_start_failure() {
        let key = worker_key("wlan0", 4, "fe80::1");
        let mut workers = HashMap::new();

        reconcile_auto_workers(&mut workers, vec![key.clone()], |_| {
            Err(io::Error::new(io::ErrorKind::AddrInUse, "busy"))
        });

        assert!(matches!(workers.get(&key), Some(AutoWorkerEntry::Pending)));
    }

    #[test]
    fn reconcile_auto_workers_retries_pending_worker() {
        let key = worker_key("wlan0", 4, "fe80::1");
        let running = Arc::new(AtomicBool::new(true));
        let mut workers = HashMap::from([(key.clone(), AutoWorkerEntry::Pending)]);
        let mut starts = 0;

        reconcile_auto_workers(&mut workers, vec![key.clone()], |_| {
            starts += 1;
            Ok(RunningAutoWorker {
                running: running.clone(),
            })
        });

        assert_eq!(starts, 1);
        assert!(matches!(
            workers.get(&key),
            Some(AutoWorkerEntry::Running(_))
        ));
    }

    #[test]
    fn reconcile_auto_workers_keeps_pending_worker_on_retry_failure() {
        let key = worker_key("wlan0", 4, "fe80::1");
        let mut workers = HashMap::from([(key.clone(), AutoWorkerEntry::Pending)]);
        let mut starts = 0;

        reconcile_auto_workers(&mut workers, vec![key.clone()], |_| {
            starts += 1;
            Err(io::Error::new(io::ErrorKind::AddrInUse, "busy"))
        });

        assert_eq!(starts, 1);
        assert!(matches!(workers.get(&key), Some(AutoWorkerEntry::Pending)));
    }

    #[test]
    fn reconcile_auto_workers_removes_pending_worker() {
        let key = worker_key("wlan0", 4, "fe80::1");
        let mut workers = HashMap::from([(key.clone(), AutoWorkerEntry::Pending)]);

        reconcile_auto_workers(&mut workers, Vec::new(), |_| unreachable!());

        assert!(!workers.contains_key(&key));
    }

    #[test]
    fn filter_skips_android_system_interfaces() {
        let allowed = Vec::new();
        let ignored = Vec::new();

        for name in ["dummy0", "lo", "tun0", "rmnet0", "rmnet7"] {
            assert!(
                !should_adopt_interface_name(name, &allowed, &ignored, ANDROID_IGNORE_IFS),
                "{name} should be skipped by Android AutoInterface defaults"
            );
        }
    }

    #[test]
    fn filter_does_not_skip_rmnet8_by_android_defaults() {
        assert!(should_adopt_interface_name(
            "rmnet8",
            &[],
            &[],
            ANDROID_IGNORE_IFS
        ));
    }

    #[test]
    fn filter_allowed_overrides_system_ignored_interface() {
        assert!(should_adopt_interface_name(
            "rmnet0",
            &["rmnet0".to_string()],
            &[],
            ANDROID_IGNORE_IFS
        ));

        assert!(should_adopt_interface_name(
            "lo0",
            &["lo0".to_string()],
            &[],
            &[]
        ));
    }

    #[test]
    fn filter_ignored_wins_over_allowed_interface() {
        assert!(!should_adopt_interface_name(
            "rmnet0",
            &["rmnet0".to_string()],
            &["rmnet0".to_string()],
            ANDROID_IGNORE_IFS
        ));
    }

    #[test]
    fn filter_allowed_list_excludes_unlisted_interfaces() {
        assert!(!should_adopt_interface_name(
            "wlan0",
            &["eth0".to_string()],
            &[],
            ANDROID_IGNORE_IFS
        ));

        assert!(should_adopt_interface_name(
            "wlan0",
            &["wlan0".to_string()],
            &[],
            ANDROID_IGNORE_IFS
        ));
    }

    // ── Config defaults ──────────────────────────────────────────────

    #[test]
    fn config_defaults() {
        let config = AutoConfig::default();
        assert_eq!(config.group_id, DEFAULT_GROUP_ID);
        assert_eq!(config.discovery_scope, SCOPE_LINK);
        assert_eq!(config.discovery_port, DEFAULT_DISCOVERY_PORT);
        assert_eq!(config.data_port, DEFAULT_DATA_PORT);
        assert_eq!(
            config.multicast_address_type,
            MULTICAST_TEMPORARY_ADDRESS_TYPE
        );
        assert_eq!(config.configured_bitrate, BITRATE_GUESS);
        assert!(config.allowed_interfaces.is_empty());
        assert!(config.ignored_interfaces.is_empty());
        assert_eq!(config.max_peers, DEFAULT_MAX_PEERS);
    }

    fn parse_auto_config(params: &[(&str, &str)]) -> AutoConfig {
        let params = params
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect::<HashMap<_, _>>();
        let config = AutoFactory
            .parse_config("auto", InterfaceId(1), &params)
            .unwrap();
        *config.into_any().downcast::<AutoConfig>().unwrap()
    }

    #[test]
    fn parse_config_reads_max_peers() {
        assert_eq!(parse_auto_config(&[]).max_peers, DEFAULT_MAX_PEERS);
        assert_eq!(parse_auto_config(&[("max_peers", "16")]).max_peers, 16);
        assert_eq!(
            parse_auto_config(&[("max_peers", "0")]).max_peers,
            DEFAULT_MAX_PEERS
        );
        assert_eq!(
            parse_auto_config(&[("max_peers", "many")]).max_peers,
            DEFAULT_MAX_PEERS
        );
    }

    // ── Peer admission ───────────────────────────────────────────────

    fn insert_peer(state: &mut SharedState, addr: &str, ifname: &str, last_heard: f64) {
        let key = peer_key(addr, 2);
        let id = InterfaceId(state.next_id.fetch_add(1, Ordering::Relaxed));
        state.peers.insert(
            key.clone(),
            AutoPeer {
                interface_id: id,
                link_local_addr: key.link_local_addr.clone(),
                ifname: ifname.to_string(),
                if_index: key.if_index,
                last_heard,
            },
        );
    }

    #[test]
    fn peer_cap_refuses_peer_beyond_limit_and_admits_after_timeout() {
        let mut state = SharedState::new(Arc::new(AtomicU64::new(100)));
        state.max_peers = 3;
        let now = Instant::now();

        for (i, addr) in ["fe80::1", "fe80::2", "fe80::3"].iter().enumerate() {
            assert_eq!(
                state.admit_new_peer("eth0", now),
                PeerAdmissionDecision::Admit
            );
            insert_peer(&mut state, addr, "eth0", 1000.0 + i as f64);
        }
        assert_eq!(
            state.admit_new_peer("eth0", now),
            PeerAdmissionDecision::AtCapacity
        );
        // Another adopted interface has its own cap.
        assert_eq!(
            state.admit_new_peer("wlan0", now),
            PeerAdmissionDecision::Admit
        );

        // fe80::1 was last heard at 1000.0 and times out first.
        let removed = state.remove_timed_out_peers(1000.0 + PEERING_TIMEOUT + 0.5, PEERING_TIMEOUT);
        assert_eq!(removed.len(), 1);
        assert_eq!(removed[0].0, peer_key("fe80::1", 2));
        assert_eq!(state.active_peers_on("eth0"), 2);

        assert_eq!(
            state.admit_new_peer("eth0", now),
            PeerAdmissionDecision::Admit
        );
    }

    #[test]
    fn peer_admission_rate_limit_refuses_within_window_and_admits_after() {
        let mut admission = PeerAdmission::default();
        let start = Instant::now();

        for _ in 0..MAX_NEW_PEERS_PER_WINDOW {
            assert_eq!(
                admission.decide(0, usize::MAX, start),
                PeerAdmissionDecision::Admit
            );
        }
        assert_eq!(
            admission.decide(0, usize::MAX, start),
            PeerAdmissionDecision::RateLimited
        );
        let almost = start + PEER_ADMISSION_WINDOW - Duration::from_millis(1);
        assert_eq!(
            admission.decide(0, usize::MAX, almost),
            PeerAdmissionDecision::RateLimited
        );

        let next_window = start + PEER_ADMISSION_WINDOW;
        assert_eq!(
            admission.decide(0, usize::MAX, next_window),
            PeerAdmissionDecision::Admit
        );
    }

    #[test]
    fn peer_admission_cap_takes_precedence_and_does_not_use_window() {
        let mut admission = PeerAdmission::default();
        let now = Instant::now();
        for _ in 0..MAX_NEW_PEERS_PER_WINDOW * 2 {
            assert_eq!(
                admission.decide(4, 4, now),
                PeerAdmissionDecision::AtCapacity
            );
        }
        assert_eq!(admission.decide(3, 4, now), PeerAdmissionDecision::Admit);
    }

    #[test]
    fn peer_refusal_warnings_are_spaced_out() {
        let mut admission = PeerAdmission::default();
        let start = Instant::now();
        assert_eq!(admission.note_refusal(start), Some(1));
        assert_eq!(admission.note_refusal(start), None);
        assert_eq!(admission.note_refusal(start + Duration::from_secs(1)), None);
        assert_eq!(
            admission.note_refusal(start + PEER_REFUSAL_LOG_INTERVAL),
            Some(3)
        );
    }

    #[test]
    fn peers_advertise_no_more_than_the_receive_buffer_takes() {
        let shared = Arc::new(Mutex::new(SharedState::new(Arc::new(AtomicU64::new(900)))));
        let runtime = Arc::new(Mutex::new(AutoRuntime::from_config(&AutoConfig::default())));
        let event = add_peer(AddPeerContext {
            shared: &shared,
            peer_key: &PeerKey {
                link_local_addr: "fe80::1".into(),
                if_index: 1,
            },
            data_port: DEFAULT_DATA_PORT,
            name: "auto-test",
            ifname: "test0",
            configured_bitrate: BITRATE_GUESS,
            ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
            runtime: &runtime,
            dynamic_template: None,
            underlay_mark: None,
            inline: false,
        });
        let Some(Event::InterfaceUp(_, Some(_), Some(info))) = event else {
            panic!("expected the new peer's InterfaceUp");
        };
        // Packets up to the MTU must fit the data receivers' buffer, or
        // link traffic above it is truncated and lost.
        assert_eq!(info.mtu as usize, HW_MTU);
    }

    #[test]
    fn remove_peers_for_worker_resets_peer_admission() {
        let next_id = Arc::new(AtomicU64::new(100));
        let shared = Arc::new(Mutex::new(SharedState::new(next_id)));
        let now = Instant::now();
        {
            let mut state = lock_or_recover(&shared, "auto shared state");
            for _ in 0..MAX_NEW_PEERS_PER_WINDOW {
                state.admit_new_peer("eth0", now);
            }
            assert_eq!(
                state.admit_new_peer("eth0", now),
                PeerAdmissionDecision::RateLimited
            );
        }
        let (tx, _rx) = crate::event::channel();
        let key = AutoWorkerKey {
            ifname: "eth0".to_string(),
            if_index: 2,
            link_local_addr: "fe80::10".to_string(),
        };
        remove_peers_for_worker(&shared, &tx, &key);
        assert_eq!(
            lock_or_recover(&shared, "auto shared state").admit_new_peer("eth0", now),
            PeerAdmissionDecision::Admit
        );
    }

    // ── Constants ────────────────────────────────────────────────────

    #[test]
    fn constants_match_python() {
        assert_eq!(DEFAULT_DISCOVERY_PORT, 29716);
        assert_eq!(DEFAULT_DATA_PORT, 42671);
        assert_eq!(HW_MTU, 1196);
        assert_eq!(MULTI_IF_DEQUE_LEN, 48);
        assert!((MULTI_IF_DEQUE_TTL - 0.75).abs() < f64::EPSILON);
        assert!((PEERING_TIMEOUT - 22.0).abs() < f64::EPSILON);
        assert!((ANNOUNCE_INTERVAL - 1.6).abs() < f64::EPSILON);
        assert!((PEER_JOB_INTERVAL - 4.0).abs() < f64::EPSILON);
        assert!((MCAST_ECHO_TIMEOUT - 6.5).abs() < f64::EPSILON);
        assert_eq!(BITRATE_GUESS, 10_000_000);
    }

    #[test]
    fn unicast_discovery_port() {
        // Python: unicast_discovery_port = discovery_port + 1
        let unicast_port = DEFAULT_DISCOVERY_PORT + 1;
        assert_eq!(unicast_port, 29717);
    }

    #[test]
    fn reverse_peering_interval() {
        let interval = ANNOUNCE_INTERVAL * REVERSE_PEERING_MULTIPLIER;
        assert!((interval - 5.2).abs() < 0.01);
    }
}