ados-protocol 0.106.2

//! Unix-socket IPC transport for the MAVLink and state sockets.
//!
//! Mirrors the semantics of `ADOSDroneAgent/src/ados/core/ipc.py`:
//!
//! - The owning service binds a Unix socket (perms `0o660`, group
//!   `ados-operator`, stale socket removed first) and broadcasts byte buffers to
//!   every connected client. The trusted local plane is root plus that group and
//!   nothing wider: two of these sockets forward inbound bytes straight to the
//!   flight controller. Plugins run as the `ados` user, which is never in the
//!   operator group, and every accept re-checks the peer's kernel credentials
//!   (see [`OperatorListener`]).
//! - Each client has a bounded outbound queue. A client whose queue fills is
//!   dropped rather than allowed to grow unbounded (slow-client policy).
//! - The state socket additionally replays the last buffer to a client the
//!   moment it connects, so a late subscriber does not wait for the next
//!   publish.
//! - The MAVLink socket is bidirectional: a frame written by a client is a
//!   command toward the flight controller, surfaced here on an inbound channel.
//!
//! Framing is the caller's concern. [`broadcast`](IpcBroadcast::broadcast)
//! sends the exact bytes given (the caller frames via [`crate::frame`] or
//! [`crate::state`]). The inbound reader, used only by the MAVLink socket,
//! decodes 4-byte big-endian length-prefixed frames.
//!
//! ## Who wrote this
//!
//! An inbound payload arrives as an [`InboundCommand`], which carries the
//! writing connection's [`IpcPeer`] alongside the bytes. The owning service
//! used to receive a bare buffer and had no way to tell one writer from
//! another, so it recorded a single hardcoded provenance for every one of
//! them. That reading was wrong for the writer that matters most: a process
//! that accepts bytes from off the node and forwards them here is not the same
//! as a process that produced them itself, and a service that cannot see the
//! difference reports remote traffic as locally originated.

use std::future::Future;
use std::io;
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

use bytes::Bytes;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{UnixListener, UnixStream};
use tokio::sync::{mpsc, Mutex};
use tokio::task::JoinHandle;

use crate::frame;

/// The control payload a writer sends to declare that the bytes it is about to
/// forward did not originate on this node.
///
/// Sent as an ordinary length-prefixed frame, so no framing changes and no
/// existing writer is affected. It cannot collide with real traffic on either
/// socket that carries inbound bytes: a MAVLink frame starts `0xFD` (v2) or
/// `0xFE` (v1) and an MSP one starts `$`, while this starts with a NUL.
///
/// The declaration is one-way and downward. A writer can only ever give up the
/// benefit of the doubt with it, never claim one, so a writer that does not
/// send it is treated exactly as it was before and a writer that forges it only
/// declassifies itself. It is consumed by the reader and never forwarded on.
pub const IPC_DECLARE_OFF_BOX_SOURCE: &[u8] = b"\x00ados-ipc/source=off-box";

mod injector;
pub use injector::{
    encode_injector_declaration, parse_injector_declaration, InjectorClaim,
    IPC_DECLARE_INJECTOR_PREFIX,
};

/// Who wrote an inbound payload, as the accepting socket can tell.
///
/// The credentials come from the kernel (`SO_PEERCRED` and its equivalents),
/// so they cannot be forged by the writer. `off_box_source` and `injector` are
/// the writer's own declarations (see [`IPC_DECLARE_OFF_BOX_SOURCE`] /
/// [`IPC_DECLARE_INJECTOR_PREFIX`]) and are sticky for the life of the
/// connection, except that a plugin-plane writer ([`is_plugin_peer`]) starts
/// out as an injector whatever it declares. Not `Copy`: the injector claim
/// carries owned strings.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct IpcPeer {
    pub uid: Option<u32>,
    pub gid: Option<u32>,
    pub pid: Option<i32>,
    /// The writing process has declared that it is forwarding bytes that
    /// reached this node from somewhere else.
    pub off_box_source: bool,
    /// The writer is an autonomous injector, so its commands are subject to
    /// the PIC arbiter: it declared itself one, or its kernel credentials put
    /// it on the plugin plane. `None` = an operator/human writer, never gated.
    pub injector: Option<InjectorClaim>,
    /// This connection's serial, unique within the accepting process, so a
    /// consumer can keep per-connection state (a partial frame) apart.
    pub conn: u64,
}

/// One inbound payload plus the identity of the connection that wrote it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InboundCommand {
    pub payload: Vec<u8>,
    pub peer: IpcPeer,
}

/// The shared service user plugin processes ran as before each plugin got its
/// own user. Still treated as plugin plane wherever it exists.
pub const LEGACY_PLUGIN_USER: &str = "ados";

/// Whether a connecting peer is on the plugin plane: the shared plugin user,
/// or any process carrying [`PLUGIN_GROUP`] as its primary or a supplementary
/// group. Such a writer is autonomous by construction, so it is gated as an
/// injector whether or not it declares itself one; a self-declaration only
/// binds the cooperating ones.
///
/// Pure over the kernel-reported credentials so the decision is testable
/// without a second uid.
pub fn is_plugin_peer(
    peer_uid: u32,
    peer_gids: &[u32],
    plugin_user_uid: Option<u32>,
    plugin_gid: Option<u32>,
) -> bool {
    plugin_user_uid == Some(peer_uid) || plugin_gid.is_some_and(|gid| peer_gids.contains(&gid))
}

/// The injector claim a plugin-plane writer carries from the moment it
/// connects. It has no ticket, so on a paired node it can never verify, and
/// the PIC arbiter lets it through only on an affirmative "nobody holds
/// control" report.
fn plugin_plane_claim(uid: u32) -> InjectorClaim {
    InjectorClaim {
        client_id: format!("uid:{uid}"),
        ticket: None,
    }
}

/// Read the connecting process's credentials. A platform or kernel that will
/// not answer leaves the fields empty rather than inventing a value.
fn peer_identity(stream: &UnixStream) -> IpcPeer {
    let Ok(cred) = stream.peer_cred() else {
        return IpcPeer::default();
    };
    #[allow(unused_mut)]
    let mut gids = vec![cred.gid()];
    #[cfg(target_os = "linux")]
    gids.extend(peer_supplementary_groups(stream));
    let plugin = is_plugin_peer(
        cred.uid(),
        &gids,
        user_uid(LEGACY_PLUGIN_USER),
        group_gid(PLUGIN_GROUP),
    );
    IpcPeer {
        uid: Some(cred.uid()),
        gid: Some(cred.gid()),
        pid: cred.pid(),
        off_box_source: false,
        injector: plugin.then(|| plugin_plane_claim(cred.uid())),
        conn: 0,
    }
}

fn user_uid(name: &str) -> Option<u32> {
    match nix::unistd::User::from_name(name) {
        Ok(user) => user.map(|u| u.uid.as_raw()),
        Err(err) => {
            tracing::debug!(error = %err, user = name, "resolving plugin user failed");
            None
        }
    }
}

/// How long the accept loops back off after a failed `accept()` before
/// retrying. Long enough that a persistent error cannot hot-spin the reactor,
/// short enough that a transient one costs a client no perceptible delay.
/// Shared by [`IpcBroadcast`] and [`serve_rpc`] so the two accept loops cannot
/// drift into different recovery policies.
const ACCEPT_RETRY_BACKOFF: Duration = Duration::from_millis(200);

/// The next accepted connection's [`IpcPeer::conn`].
static NEXT_CONN: AtomicU64 = AtomicU64::new(1);

/// Kernel send buffer requested on every accepted broadcast client.
///
/// A stalled consumer is evicted once its kernel send buffer AND its
/// outbound queue are full. Unpinned, the kernel buffer is
/// `net.core.wmem_default`, which the installer raises to 4 MiB for video,
/// so a stalled MAVLink, MSP or state reader buffered megabytes of stale
/// frames before eviction. Linux doubles the request and caps it at
/// `wmem_max`, so the effective buffer is at most 512 KiB.
const IPC_CLIENT_SEND_BUFFER_BYTES: usize = 256 * 1024;

/// Where an accept loop gets its client connections.
///
/// The real implementation is the bound [`UnixListener`]. It is a trait purely
/// so the accept-ERROR branch is reachable from a test: a live `accept()` only
/// fails under process-wide fd exhaustion or a kernel resource shortage, which
/// cannot be induced in-process without wrecking every other test sharing the
/// runtime. That is exactly why the branch which decides whether these sockets
/// survive an accept error carried no coverage, and why it was a `break`.
///
/// Written as `-> impl Future<..> + Send` rather than `async fn` on purpose, and
/// `clippy::manual_async_fn` is allowed for exactly that reason: the accept loop
/// runs under `tokio::spawn`, so the returned future MUST be `Send`, and an
/// `async fn` in a trait gives no way to say so. Simplifying to `async fn` makes
/// the spawn fail to compile.
trait ClientSource: Send + 'static {
    #[allow(clippy::manual_async_fn)]
    fn next_client(&mut self) -> impl Future<Output = io::Result<UnixStream>> + Send;
}

impl ClientSource for OperatorListener {
    #[allow(clippy::manual_async_fn)]
    fn next_client(&mut self) -> impl Future<Output = io::Result<UnixStream>> + Send {
        async move {
            OperatorListener::accept(self)
                .await
                .map(|(stream, _addr)| stream)
        }
    }
}

/// One connected client: its outbound queue plus the writer and reader tasks,
/// so both can be aborted when the client is pruned or the server is dropped.
///
/// The queue carries [`Bytes`], not `Vec<u8>`: one broadcast buffer is handed to
/// every attached client, so a `Vec` made that a full copy per client per frame.
struct ClientHandle {
    tx: mpsc::Sender<Bytes>,
    writer: JoinHandle<()>,
    reader: JoinHandle<()>,
}

/// A Unix-socket broadcast server. Drops slow clients; optionally replays the
/// last buffer on connect; optionally decodes inbound length-prefixed frames.
pub struct IpcBroadcast {
    path: PathBuf,
    queue_depth: usize,
    clients: Arc<Mutex<Vec<ClientHandle>>>,
    last: Arc<Mutex<Option<Bytes>>>,
    keep_last: bool,
    accept_task: JoinHandle<()>,
    /// Monotonic count of clients evicted for falling behind — a full outbound
    /// queue, and only that. A consumer that simply disconnected is pruned
    /// without advancing this counter, so the number means "a consumer could
    /// not keep up" rather than "a client went away". A slow consumer is
    /// otherwise pruned silently; surfacing this counter lets the owning
    /// service report the eviction the same way the radio TX-liveness watchdog
    /// surfaces a stalled link — a silent drop is itself the failure mode to
    /// detect, and a counter that also ticks on normal disconnects would drown
    /// that signal.
    dropped_clients: Arc<AtomicU64>,
}

impl IpcBroadcast {
    /// Bind the socket and start accepting clients.
    ///
    /// - `queue_depth`: per-client outbound queue size (256 for MAVLink, 32 for
    ///   state, per the contracts).
    /// - `keep_last`: replay the last broadcast buffer to a newly connected
    ///   client (state socket behaviour).
    /// - `inbound`: if `Some`, each client connection is also read as a stream
    ///   of length-prefixed frames whose payloads are forwarded on the channel
    ///   as [`InboundCommand`]s carrying the writing connection's identity
    ///   (MAVLink command path). The capacity bounds the inbound backlog.
    pub async fn bind(
        path: impl AsRef<Path>,
        queue_depth: usize,
        keep_last: bool,
        inbound: Option<usize>,
    ) -> io::Result<(Self, Option<mpsc::Receiver<InboundCommand>>)> {
        let path = path.as_ref().to_path_buf();
        // Owner+group rw (0o660, group `ados-operator`), matching every other
        // command socket in the tree; the shared helper owns the create-dir /
        // remove-stale / bind / chmod / chgrp hygiene and the per-accept peer
        // check.
        //
        // Two of these sockets — the MAVLink and MSP lanes — accept inbound
        // bytes that are written straight through to the flight controller, so
        // a peer that can open one can arm an aircraft. The trusted-local plane
        // is root plus the operator group; plugins run as `ados`, outside it,
        // and reach the flight controller through the plugin host, which gates
        // every write on the plugin's granted capabilities.
        let listener = bind_command_socket(&path, 0o660)?;

        Ok(Self::from_source(
            path,
            listener,
            queue_depth,
            keep_last,
            inbound,
        ))
    }

    /// Start the server over an arbitrary [`ClientSource`].
    ///
    /// Split out of [`Self::bind`] so the accept loop's error-recovery branch is
    /// exercisable: the source is the only part of the loop a test cannot drive
    /// through a real socket.
    fn from_source<S: ClientSource>(
        path: PathBuf,
        mut source: S,
        queue_depth: usize,
        keep_last: bool,
        inbound: Option<usize>,
    ) -> (Self, Option<mpsc::Receiver<InboundCommand>>) {
        let clients: Arc<Mutex<Vec<ClientHandle>>> = Arc::new(Mutex::new(Vec::new()));
        let last: Arc<Mutex<Option<Bytes>>> = Arc::new(Mutex::new(None));
        let dropped_clients = Arc::new(AtomicU64::new(0));

        let (inbound_tx, inbound_rx) = match inbound {
            Some(cap) => {
                let (tx, rx) = mpsc::channel::<InboundCommand>(cap);
                (Some(tx), Some(rx))
            }
            None => (None, None),
        };

        let accept_clients = clients.clone();
        let accept_last = last.clone();
        let accept_task = tokio::spawn(async move {
            loop {
                let stream = match source.next_client().await {
                    Ok(s) => s,
                    Err(e) => {
                        // Never exit. This was `break`, which ended the accept
                        // task for the life of the process on ONE transient
                        // error — EMFILE/ENFILE under fd pressure, ENOBUFS,
                        // ECONNABORTED — while leaving the struct looking
                        // healthy: `client_count()` still reported the
                        // already-connected clients and `dropped_clients()`
                        // never ticked, so nothing anywhere could tell the
                        // socket had stopped accepting. Both the MAVLink and the
                        // state socket are bound through this one function, so a
                        // single hiccup silently closed the FC command surface
                        // and the telemetry snapshot to every future client
                        // until a manual restart.
                        //
                        // Back off so a persistent error cannot hot-spin, then
                        // retry forever — the same policy `serve_rpc` below
                        // already applies to the command sockets.
                        tracing::warn!(error = %e, "ipc broadcast socket accept failed");
                        tokio::time::sleep(ACCEPT_RETRY_BACKOFF).await;
                        continue;
                    }
                };
                Self::on_client(
                    stream,
                    queue_depth,
                    keep_last,
                    accept_clients.clone(),
                    accept_last.clone(),
                    inbound_tx.clone(),
                )
                .await;
            }
        });

        (
            Self {
                path,
                queue_depth,
                clients,
                last,
                keep_last,
                accept_task,
                dropped_clients,
            },
            inbound_rx,
        )
    }

    async fn on_client(
        stream: UnixStream,
        queue_depth: usize,
        keep_last: bool,
        clients: Arc<Mutex<Vec<ClientHandle>>>,
        last: Arc<Mutex<Option<Bytes>>>,
        inbound_tx: Option<mpsc::Sender<InboundCommand>>,
    ) {
        pin_send_buffer(&stream);
        let mut peer = peer_identity(&stream);
        peer.conn = NEXT_CONN.fetch_add(1, Ordering::Relaxed);
        let (mut read_half, mut write_half) = stream.into_split();
        let (tx, mut rx) = mpsc::channel::<Bytes>(queue_depth);

        // Replay the last buffer immediately for the state socket.
        if keep_last {
            if let Some(buf) = last.lock().await.clone() {
                let _ = tx.try_send(buf);
            }
        }

        // Writer task: drain the queue to the socket, await flush so kernel
        // backpressure stays inside the task and never blocks the producer.
        let writer = tokio::spawn(async move {
            while let Some(buf) = rx.recv().await {
                if write_half.write_all(&buf).await.is_err() {
                    break;
                }
                if write_half.flush().await.is_err() {
                    break;
                }
            }
        });

        // Reader task. The MAVLink socket (inbound channel present) decodes
        // length-prefixed frames and forwards the payloads as commands toward
        // the flight controller. The write-only state socket has no inbound
        // protocol, so it just drains raw bytes to detect EOF, matching the
        // Python state server which only waits for the client to disconnect.
        let reader = tokio::spawn(async move {
            match inbound_tx {
                Some(tx) => {
                    let mut peer = peer;
                    loop {
                        let mut header = [0u8; frame::HEADER_SIZE];
                        if read_half.read_exact(&mut header).await.is_err() {
                            break;
                        }
                        let len = match frame::decode_len(header, frame::MAVLINK_MAX_FRAME, false) {
                            Ok(n) => n,
                            Err(_) => break,
                        };
                        let mut payload = vec![0u8; len];
                        if len > 0 && read_half.read_exact(&mut payload).await.is_err() {
                            break;
                        }
                        // The origin declaration is a control frame, not
                        // traffic: it latches for the rest of the connection
                        // and is not passed on to the flight controller.
                        if payload == IPC_DECLARE_OFF_BOX_SOURCE {
                            peer.off_box_source = true;
                            continue;
                        }
                        if let Some(claim) = parse_injector_declaration(&payload) {
                            // Sticky, one-way: a writer subjects itself to the PIC
                            // arbiter. Never forwarded to the FC.
                            peer.injector = Some(claim);
                            continue;
                        }
                        if tx
                            .send(InboundCommand {
                                payload,
                                peer: peer.clone(),
                            })
                            .await
                            .is_err()
                        {
                            break;
                        }
                    }
                }
                None => {
                    let mut scratch = [0u8; 256];
                    while read_half.read(&mut scratch).await.unwrap_or(0) > 0 {}
                }
            }
        });

        clients
            .lock()
            .await
            .push(ClientHandle { tx, writer, reader });
    }

    /// Broadcast a byte buffer to all connected clients. Clients whose queue is
    /// full are dropped. If `keep_last` is set, the buffer is also stored as the
    /// last-state replayed to future clients.
    ///
    /// Takes [`Bytes`] rather than `Vec<u8>` because every clone below — one per
    /// attached client, plus one for the retained last-state — used to be a full
    /// copy of the buffer. On the FC frame lane that is ~66 copies/s per
    /// connected consumer, growing with the number of consumers; as `Bytes` each
    /// is a refcount bump and the cost stops scaling with the client count. A
    /// producer holding a freshly built `Vec<u8>` converts with `.into()`, which
    /// is a move, not a copy.
    pub async fn broadcast(&self, buf: Bytes) {
        if self.keep_last {
            *self.last.lock().await = Some(buf.clone());
        }
        let mut clients = self.clients.lock().await;
        let mut keep = Vec::with_capacity(clients.len());
        let mut dropped = 0u64;
        for client in clients.drain(..) {
            match client.tx.try_send(buf.clone()) {
                Ok(()) => keep.push(client),
                // Queue full: the consumer is genuinely not draining. Evict it
                // and count it — a slow consumer is a real fault, and this
                // counter is the only producer-visible signal that it happened.
                Err(mpsc::error::TrySendError::Full(_)) => {
                    client.reader.abort();
                    client.writer.abort();
                    dropped += 1;
                }
                // Channel closed: the consumer already went away. That is a
                // normal disconnect, NOT a slow client, so it is pruned without
                // touching the eviction counter. Counting it reported every
                // short-lived reader as a fault: `ados-cloud` opens a fresh
                // connection per enrichment tick (~1 Hz), reads the replayed
                // last-state and closes, which alone logged ~86k spurious
                // `ipc_slow_client_evicted` warnings a day and buried the real
                // evictions in the noise.
                Err(mpsc::error::TrySendError::Closed(_)) => {
                    client.reader.abort();
                    client.writer.abort();
                }
            }
        }
        *clients = keep;
        // One relaxed add for the whole broadcast rather than per pruned
        // client; the owning service polls this and reports it.
        if dropped > 0 {
            self.dropped_clients.fetch_add(dropped, Ordering::Relaxed);
        }
    }

    /// Number of currently connected clients.
    pub async fn client_count(&self) -> usize {
        self.clients.lock().await.len()
    }

    /// The per-client queue depth this server was bound with.
    pub fn queue_depth(&self) -> usize {
        self.queue_depth
    }

    /// Monotonic count of clients evicted for falling behind — a full outbound
    /// queue — since this server was bound. A client that disconnected cleanly
    /// is not counted here.
    ///
    /// A slow consumer is pruned without an error reaching it, so this counter
    /// is the producer-visible signal that an eviction happened. The owning
    /// service folds it into the state snapshot it already publishes so the
    /// eviction is observable remotely rather than silent. Lock-free.
    pub fn dropped_clients(&self) -> u64 {
        self.dropped_clients.load(Ordering::Relaxed)
    }
}

impl Drop for IpcBroadcast {
    fn drop(&mut self) {
        self.accept_task.abort();
        // Abort the per-client reader/writer tasks so none survive the server.
        if let Ok(clients) = self.clients.try_lock() {
            for client in clients.iter() {
                client.reader.abort();
                client.writer.abort();
            }
        }
        let _ = std::fs::remove_file(&self.path);
    }
}

/// The group that owns every agent command-plane socket: the control socket,
/// the `*-cmd.sock` command sockets, the MAVLink/MSP/state lanes, and the rest
/// of the sockets bound through [`bind_command_socket`]. The installer creates
/// it and adds the human operator; the plugin user is never a member, whatever
/// capabilities it holds (a sensor-bus reader joins [`SENSORS_GROUP`]).
pub const OPERATOR_GROUP: &str = "ados-operator";

/// The group that may write `/run/ados/capture` (the capture-active marker
/// directory, `2775 root:ados-capture`) and the capture root, which the
/// supervisor's janitor and the installer keep at `3775 root:ados-capture`.
/// A capture plugin joins it only while it holds `capture.storage.write`; it
/// guards nothing else.
pub const CAPTURE_GROUP: &str = "ados-capture";

/// The reader group of the sensor bus (`/run/ados/sensing/sensors.sock`), bound with
/// [`bind_reader_socket`]. A member may connect and subscribe; what it
/// publishes is dropped. A plugin joins it only while it holds
/// `sensor.stream.read`, which is how a sandboxed plugin reads the bus
/// without joining [`OPERATOR_GROUP`]. The installer also adds the human
/// operator, whose operator membership still decides what it may publish.
pub const SENSORS_GROUP: &str = "ados-sensors";

/// The group every plugin unit runs in as its primary group. Each plugin has
/// its own system user; this shared group owns only the per-plugin sockets
/// ([`bind_plugin_socket`]), which gate every request on the plugin's own
/// capability token, and it is what the plugin loopback guard and the router
/// match a plugin process by.
pub const PLUGIN_GROUP: &str = "ados-plugins";

/// Whether a connecting peer may use a command-plane socket.
///
/// Root always may. So may a peer running as the same uid as the serving
/// process: every agent service runs as root, so on a node this adds nothing,
/// and on a dev host it lets the owning user's own tools in. Anyone else must
/// carry the operator group, primary or supplementary. With the group absent
/// from the host (`operator_gid` is `None`) nobody else is admitted.
///
/// Pure over the kernel-reported credentials so the decision is testable
/// without a second uid.
pub fn operator_peer_allowed(
    peer_uid: u32,
    peer_gids: &[u32],
    server_uid: u32,
    operator_gid: Option<u32>,
) -> bool {
    peer_uid == 0
        || peer_uid == server_uid
        || operator_gid.is_some_and(|gid| peer_gids.contains(&gid))
}

/// A bound command-plane socket. [`accept`](Self::accept) yields only peers
/// [`operator_peer_allowed`] admits (plus, on a socket bound with
/// [`bind_reader_socket`], members of its reader group); any other peer is
/// closed before a byte of it is read.
///
/// The socket file is already `0o660 root:ados-operator` (or `root:<reader
/// group>`), so the kernel refuses `connect()` to anyone outside the group. The
/// per-accept check is the second line: it holds when the chgrp could not be
/// applied, when the mode is widened by hand, or when a descriptor to the
/// socket is passed to a process that could not have opened it.
pub struct OperatorListener {
    inner: UnixListener,
    path: PathBuf,
    server_uid: u32,
    operator_gid: Option<u32>,
    /// The reader group's gid on a [`bind_reader_socket`] socket.
    reader_gid: Option<u32>,
}

impl OperatorListener {
    /// Accept the next admitted peer. Refused peers are logged and dropped; the
    /// call keeps waiting for one that is admitted. An `accept()` error is
    /// returned to the caller's own retry policy unchanged.
    pub async fn accept(&self) -> io::Result<(UnixStream, tokio::net::unix::SocketAddr)> {
        loop {
            let (stream, addr) = self.inner.accept().await?;
            if self.classify(&stream).is_some() {
                return Ok((stream, addr));
            }
        }
    }

    /// [`accept`](Self::accept), also returning what the admitted peer is
    /// ([`classify_peer`]), so a server can narrow what each class may do.
    pub async fn accept_classified(&self) -> io::Result<(UnixStream, PeerClass)> {
        loop {
            let (stream, _) = self.inner.accept().await?;
            if let Some(class) = self.classify(&stream) {
                return Ok((stream, class));
            }
        }
    }

    fn classify(&self, stream: &UnixStream) -> Option<PeerClass> {
        match peer_credentials(stream) {
            Ok((uid, gids)) => {
                let class = classify_peer(
                    uid,
                    &gids,
                    self.server_uid,
                    self.operator_gid,
                    self.reader_gid,
                );
                if class.is_none() {
                    tracing::warn!(
                        path = %self.path.display(),
                        peer_uid = uid,
                        "command socket refused a peer outside the operator group"
                    );
                }
                class
            }
            // The narrowest class the socket admits.
            Err(err) if peer_hung_up_before_accept(&err) => Some(if self.reader_gid.is_some() {
                PeerClass::Reader
            } else {
                PeerClass::Operator
            }),
            Err(err) => {
                tracing::warn!(
                    path = %self.path.display(),
                    error = %err,
                    "command socket refused a peer with unreadable credentials"
                );
                None
            }
        }
    }
}

/// The connecting peer's uid and group set, as the kernel reports them. An
/// error means the platform would not answer, which the caller treats as a
/// refusal (see `peer_hung_up_before_accept` for the one exception).
fn peer_credentials(stream: &UnixStream) -> io::Result<(u32, Vec<u32>)> {
    let cred = stream.peer_cred()?;
    #[allow(unused_mut)]
    let mut gids = vec![cred.gid()];
    #[cfg(target_os = "linux")]
    gids.extend(peer_supplementary_groups(stream));
    Ok((cred.uid(), gids))
}

/// Whether the process on the other end of `stream` is on the operator plane:
/// root, this process's own uid, or a member of [`OPERATOR_GROUP`] (the same
/// policy [`OperatorListener`] enforces). For a socket that must also accept
/// plugin-plane peers (the log ingest sink), this tells the two apart so what a
/// plugin writes is recorded as the plugin's. Unreadable credentials read as
/// not-operator, which can only narrow trust, except for the development-host
/// case `peer_hung_up_before_accept` documents.
pub fn is_operator_peer(stream: &UnixStream) -> bool {
    match peer_credentials(stream) {
        Ok((uid, gids)) => operator_peer_allowed(
            uid,
            &gids,
            nix::unistd::geteuid().as_raw(),
            group_gid(OPERATOR_GROUP),
        ),
        Err(err) => peer_hung_up_before_accept(&err),
    }
}

/// Linux records `SO_PEERCRED` at `connect()` and answers it even after the
/// peer has closed. macOS answers `LOCAL_PEERCRED` with `ENOTCONN` once the peer
/// has hung up, even while its bytes are still queued, so a client that sends
/// and closes before the accept would lose everything it sent. macOS is a
/// development host only, and the socket file mode still gates `connect()`
/// there, so a peer that already hung up is admitted on non-Linux hosts.
#[cfg(not(target_os = "linux"))]
fn peer_hung_up_before_accept(err: &io::Error) -> bool {
    err.kind() == io::ErrorKind::NotConnected
}

#[cfg(target_os = "linux")]
fn peer_hung_up_before_accept(_err: &io::Error) -> bool {
    false
}

/// The peer's supplementary groups at `connect()` time (`SO_PEERGROUPS`). An
/// error reads as no supplementary groups, which can only narrow admission.
#[cfg(target_os = "linux")]
fn peer_supplementary_groups(stream: &UnixStream) -> Vec<u32> {
    use nix::libc;
    use std::os::fd::AsRawFd;

    const GID_SIZE: usize = std::mem::size_of::<libc::gid_t>();
    let fd = stream.as_raw_fd();
    let mut groups: Vec<libc::gid_t> = vec![0; 32];
    // One retry: on ERANGE the kernel writes the size it needs into `len`.
    for _ in 0..2 {
        let mut len = (groups.len() * GID_SIZE) as libc::socklen_t;
        // SAFETY: `groups` is a live, writable buffer of exactly `len` bytes,
        // and the kernel writes at most `len` bytes into it and updates `len`.
        let rc = unsafe {
            libc::getsockopt(
                fd,
                libc::SOL_SOCKET,
                libc::SO_PEERGROUPS,
                groups.as_mut_ptr().cast(),
                &mut len,
            )
        };
        if rc == 0 {
            groups.truncate(len as usize / GID_SIZE);
            return groups;
        }
        let needed = len as usize / GID_SIZE;
        if io::Error::last_os_error().raw_os_error() != Some(libc::ERANGE) || needed <= groups.len()
        {
            break;
        }
        groups.resize(needed, 0);
    }
    Vec::new()
}

/// Pin the kernel send buffer of an accepted broadcast client to
/// [`IPC_CLIENT_SEND_BUFFER_BYTES`], so how much a stalled reader can hold
/// before eviction does not follow the system-wide `wmem_default`. A failure
/// leaves the system default in place; the client is still served.
fn pin_send_buffer(stream: &UnixStream) {
    use nix::libc;
    use std::os::fd::AsRawFd;

    let size = IPC_CLIENT_SEND_BUFFER_BYTES as libc::c_int;
    // SAFETY: the pointer refers to a live `c_int` and the length passed is
    // exactly its size; the kernel only reads from it.
    let rc = unsafe {
        libc::setsockopt(
            stream.as_raw_fd(),
            libc::SOL_SOCKET,
            libc::SO_SNDBUF,
            std::ptr::from_ref(&size).cast(),
            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
        )
    };
    if rc != 0 {
        tracing::warn!(
            error = %io::Error::last_os_error(),
            requested = IPC_CLIENT_SEND_BUFFER_BYTES,
            "ipc_client_sndbuf_set_failed"
        );
    }
}

/// Bind a command-plane socket: create the parent directory, remove a stale
/// socket left by a prior run (so `bind` cannot fail with `EADDRINUSE`), bind,
/// set the file mode, and hand the file to [`OPERATOR_GROUP`]. Every accept on
/// the returned listener re-checks the peer (see [`OperatorListener`]).
///
/// [`IpcBroadcast::bind`] and the one-shot command sockets share it so the
/// hygiene and the peer policy live in one place.
///
/// Like [`UnixListener::bind`], this must be called from within a Tokio runtime
/// context.
pub fn bind_command_socket(path: impl AsRef<Path>, mode: u32) -> io::Result<OperatorListener> {
    let path = path.as_ref();
    let inner = bind_socket_file(path, mode)?;
    let operator_gid = group_gid(OPERATOR_GROUP);
    set_socket_group(path, OPERATOR_GROUP, operator_gid);
    Ok(OperatorListener {
        inner,
        path: path.to_path_buf(),
        server_uid: nix::unistd::geteuid().as_raw(),
        operator_gid,
        reader_gid: None,
    })
}

mod peer_class;
pub use peer_class::{bind_reader_socket, classify_peer, PeerClass};

/// Bind a socket on the plugin-reachable plane: the same hygiene as
/// [`bind_command_socket`], owned by [`PLUGIN_GROUP`] so a plugin's unit can
/// connect. There is no peer check here, so it is only for surfaces a plugin
/// may legitimately reach: the per-plugin sockets, where the server gates every
/// request on the plugin's own capability token, and the log ingest sink, which
/// only accepts log frames. Never use it for an agent command socket.
pub fn bind_plugin_socket(path: impl AsRef<Path>, mode: u32) -> io::Result<UnixListener> {
    let path = path.as_ref();
    let listener = bind_socket_file(path, mode)?;
    set_socket_group(path, PLUGIN_GROUP, group_gid(PLUGIN_GROUP));
    Ok(listener)
}

fn bind_socket_file(path: &Path, mode: u32) -> io::Result<UnixListener> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).ok();
    }
    // Remove a stale socket so bind does not fail with EADDRINUSE.
    let _ = std::fs::remove_file(path);
    let listener = UnixListener::bind(path)?;
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode))?;
    Ok(listener)
}

fn group_gid(name: &str) -> Option<u32> {
    match nix::unistd::Group::from_name(name) {
        Ok(group) => group.map(|g| g.gid.as_raw()),
        Err(err) => {
            tracing::debug!(error = %err, group = name, "resolving socket group failed");
            None
        }
    }
}

/// Give a freshly-bound socket to `group`.
///
/// This belongs next to the mode, not in each caller, because the two are one
/// decision: every service here runs as root, so a `0o660` socket grants
/// *nobody but root* until the group owns the file.
///
/// Best-effort by design: the installer creates the groups, and on a dev host
/// where one is absent this is a quiet no-op, which leaves the socket reachable
/// by its owner only. Linux-only; a stub elsewhere.
#[cfg(target_os = "linux")]
fn set_socket_group(path: &Path, group: &str, gid: Option<u32>) {
    let Some(gid) = gid else {
        tracing::debug!(
            group,
            "socket group not present; leaving socket group as-is"
        );
        return;
    };
    if let Err(err) = nix::unistd::chown(path, None, Some(nix::unistd::Gid::from_raw(gid))) {
        tracing::debug!(error = %err, path = %path.display(), group, "chgrp socket failed");
    }
}

#[cfg(not(target_os = "linux"))]
fn set_socket_group(_path: &Path, _group: &str, _gid: Option<u32>) {}

/// Serve a one-shot request/response command socket.
///
/// The accept loop never exits on a transient `accept()` error: it logs, backs
/// off briefly, and retries, so a momentary fd-pressure hiccup cannot silently
/// kill the command surface for the process lifetime (a command socket that
/// dies while the service stays up would need a manual restart). Each
/// connection is handled on its own task: read exactly one newline-terminated
/// request (capped at
/// `max_request` bytes so a peer that never sends a newline cannot grow the
/// buffer without bound; an over-cap request closes the connection with no
/// response), call `handler` with the request bytes (trailing newline
/// stripped), write the returned bytes followed by a single trailing newline,
/// flush, and close.
///
/// `handler` carries the pure parse + dispatch: it takes the raw request line
/// and returns the raw response line. Framing (the trailing newline) and the
/// socket lifecycle are owned here, so a service only supplies its dispatch.
///
/// Use this only for sockets that are unambiguously one-shot newline
/// request/response. A length-prefixed, streaming, stateful (handshake), or
/// multi-request socket keeps its own accept loop.
pub async fn serve_rpc<H, Fut>(listener: OperatorListener, max_request: usize, handler: H)
where
    H: Fn(Vec<u8>) -> Fut + Clone + Send + Sync + 'static,
    Fut: Future<Output = Vec<u8>> + Send + 'static,
{
    loop {
        let mut stream = match listener.accept().await {
            Ok((s, _addr)) => s,
            Err(e) => {
                // Backoff so a persistent accept error cannot hot-spin; never
                // exit, so the command surface survives transient fd pressure.
                tracing::warn!(error = %e, "command socket accept failed");
                tokio::time::sleep(ACCEPT_RETRY_BACKOFF).await;
                continue;
            }
        };
        let handler = handler.clone();
        tokio::spawn(async move {
            // One newline-terminated request per connection. A clean EOF before
            // any request, or a line that exceeds the cap, just closes.
            let request = match read_newline_line(&mut stream, max_request).await {
                Ok(Some(req)) => req,
                Ok(None) | Err(_) => return,
            };
            let mut response = handler(request).await;
            response.push(b'\n');
            if stream.write_all(&response).await.is_err() {
                return;
            }
            let _ = stream.flush().await;
            // `stream` drops here, closing the connection.
        });
    }
}

/// Connect to a Unix socket with bounded retry (mirrors the Python client's
/// connect-with-retry). Returns the stream once connected.
pub async fn connect_with_retry(
    path: impl AsRef<Path>,
    retries: u32,
    delay: Duration,
) -> io::Result<UnixStream> {
    let path = path.as_ref();
    let mut last_err: Option<io::Error> = None;
    for attempt in 0..retries.max(1) {
        match UnixStream::connect(path).await {
            Ok(s) => return Ok(s),
            Err(e) => {
                last_err = Some(e);
                if attempt + 1 < retries {
                    tokio::time::sleep(delay).await;
                }
            }
        }
    }
    Err(last_err.unwrap_or_else(|| io::Error::new(io::ErrorKind::NotFound, "connect failed")))
}

/// Read one length-prefixed frame from a stream. Returns `Ok(None)` on clean
/// EOF. Used to consume the MAVLink broadcast on the client side.
pub async fn read_length_prefixed<R: AsyncReadExt + Unpin>(
    reader: &mut R,
    max: usize,
    reject_zero: bool,
) -> io::Result<Option<Vec<u8>>> {
    let mut header = [0u8; frame::HEADER_SIZE];
    match reader.read_exact(&mut header).await {
        Ok(_) => {}
        Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),
        Err(e) => return Err(e),
    }
    let len = frame::decode_len(header, max, reject_zero)
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?;
    let mut payload = vec![0u8; len];
    if len > 0 {
        reader.read_exact(&mut payload).await?;
    }
    Ok(Some(payload))
}

/// Read one newline-terminated line (state socket v1) from a stream. Returns
/// `Ok(None)` on clean EOF. The returned buffer excludes the trailing newline.
///
/// `max` caps the line length so a peer that never sends a newline cannot grow
/// the buffer without bound. A line that reaches `max` without a newline
/// returns `InvalidData`.
pub async fn read_newline_line<R: AsyncReadExt + Unpin>(
    reader: &mut R,
    max: usize,
) -> io::Result<Option<Vec<u8>>> {
    let mut buf = Vec::new();
    let mut byte = [0u8; 1];
    loop {
        match reader.read_exact(&mut byte).await {
            Ok(_) => {
                if byte[0] == b'\n' {
                    return Ok(Some(buf));
                }
                if buf.len() >= max {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidData,
                        "state line exceeded maximum length without a newline",
                    ));
                }
                buf.push(byte[0]);
            }
            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
                return if buf.is_empty() {
                    Ok(None)
                } else {
                    Ok(Some(buf))
                };
            }
            Err(e) => return Err(e),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::frame::{encode_frame, MAVLINK_MAX_FRAME};

    fn temp_sock(name: &str) -> PathBuf {
        let mut p = std::env::temp_dir();
        let unique = format!("ados-ipc-test-{}-{}.sock", std::process::id(), name);
        p.push(unique);
        let _ = std::fs::remove_file(&p);
        p
    }

    /// A client source that fails a fixed number of times before delegating to
    /// a real listener.
    ///
    /// EMFILE is chosen because it is the error that actually happens: the agent
    /// runs a dozen services against one fd table, and a burst of plugin or
    /// proxy connections is the realistic way `accept()` fails on a flight
    /// daemon. It cannot be induced in-process without lowering the whole
    /// process's fd limit, which is why this seam exists.
    struct FlakyAcceptor {
        listener: OperatorListener,
        failures_left: usize,
    }

    impl ClientSource for FlakyAcceptor {
        // Same reason as the trait: the accept loop is spawned, so the future
        // must be `Send`, which `async fn` in a trait cannot state.
        #[allow(clippy::manual_async_fn)]
        fn next_client(&mut self) -> impl Future<Output = io::Result<UnixStream>> + Send {
            async move {
                if self.failures_left > 0 {
                    self.failures_left -= 1;
                    // 24 == EMFILE on both Linux and macOS.
                    return Err(io::Error::from_raw_os_error(24));
                }
                self.listener.accept().await.map(|(s, _addr)| s)
            }
        }
    }

    /// One transient `accept()` error must not close the socket for the life of
    /// the process.
    ///
    /// Why the `Err(_) => break` this replaces survived review: nothing observable
    /// changes when the accept task exits. The socket is still bound and the
    /// kernel still accepts into its backlog, so a client's `connect()` SUCCEEDS
    /// and the client then hangs forever with no frames. Meanwhile
    /// `client_count()` keeps reporting the clients that connected before the
    /// error and `dropped_clients()` never ticks, so no counter, log line or
    /// health probe anywhere can tell that the socket stopped accepting. Both
    /// `mavlink.sock` and `state.sock` are bound through this one function, so a
    /// single hiccup under fd pressure silently closed the FC command surface and
    /// the telemetry snapshot to every future client until a manual restart.
    ///
    /// The assertion is therefore on DELIVERY to a client accepted after the
    /// errors, not on the loop's internal state — which is the only thing that
    /// distinguishes the two behaviours from outside.
    #[tokio::test]
    async fn the_accept_loop_survives_an_accept_error_and_still_serves_the_next_client() {
        let path = temp_sock("acceptretry");
        let listener = bind_command_socket(&path, 0o660).unwrap();
        let (server, _inbound) = IpcBroadcast::from_source(
            path.clone(),
            FlakyAcceptor {
                listener,
                failures_left: 2,
            },
            256,
            false,
            None,
        );

        // A client that connects only AFTER the loop has already absorbed two
        // accept failures. Before the fix the loop had exited on the first one,
        // so this connect succeeded at the kernel (the socket is still bound and
        // its backlog still accepts) and then hung forever with no frames — the
        // failure mode that made this invisible.
        let mut client = connect_with_retry(&path, 200, Duration::from_millis(20))
            .await
            .unwrap();
        // Two backoffs plus registration.
        tokio::time::sleep(ACCEPT_RETRY_BACKOFF * 3).await;
        assert_eq!(server.client_count().await, 1);

        server
            .broadcast(
                encode_frame(b"post-emfile", MAVLINK_MAX_FRAME)
                    .unwrap()
                    .into(),
            )
            .await;
        let got = tokio::time::timeout(
            Duration::from_secs(2),
            read_length_prefixed(&mut client, MAVLINK_MAX_FRAME, false),
        )
        .await
        .expect("a frame must reach a client accepted after an accept error")
        .unwrap();
        assert_eq!(got.as_deref(), Some(&b"post-emfile"[..]));
    }

    #[tokio::test]
    async fn broadcasts_framed_payloads_to_all_clients() {
        let path = temp_sock("bcast");
        let (server, _inbound) = IpcBroadcast::bind(&path, 256, false, None).await.unwrap();

        let mut c1 = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        let mut c2 = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        // Let the accept loop register both clients.
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert_eq!(server.client_count().await, 2);

        server
            .broadcast(encode_frame(b"hello", MAVLINK_MAX_FRAME).unwrap().into())
            .await;

        let f1 = read_length_prefixed(&mut c1, MAVLINK_MAX_FRAME, false)
            .await
            .unwrap();
        let f2 = read_length_prefixed(&mut c2, MAVLINK_MAX_FRAME, false)
            .await
            .unwrap();
        assert_eq!(f1.as_deref(), Some(&b"hello"[..]));
        assert_eq!(f2.as_deref(), Some(&b"hello"[..]));
    }

    #[tokio::test]
    async fn replays_last_state_on_connect() {
        let path = temp_sock("laststate");
        let (server, _inbound) = IpcBroadcast::bind(&path, 32, true, None).await.unwrap();

        // Publish before any client connects.
        server
            .broadcast(Bytes::from_static(b"{\"armed\":false}\n"))
            .await;

        let mut client = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        let line = read_newline_line(&mut client, crate::state::STATE_V2_MAX_FRAME)
            .await
            .unwrap();
        assert_eq!(line.as_deref(), Some(&b"{\"armed\":false}"[..]));
    }

    #[tokio::test]
    async fn slow_client_is_dropped_fast_client_survives() {
        let path = temp_sock("slow");
        // Small queue so a non-draining client fills it once the kernel socket
        // buffer is also full.
        let (server, _inbound) = IpcBroadcast::bind(&path, 2, false, None).await.unwrap();

        // "slow" connects but never reads; "fast" connects and drains continuously.
        let _slow = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        let mut fast = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        let fast_reader = tokio::spawn(async move {
            while let Ok(Some(_)) = read_length_prefixed(&mut fast, MAVLINK_MAX_FRAME, false).await
            {
            }
        });
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert_eq!(server.client_count().await, 2);

        // Large payloads fill the slow client's kernel send buffer; its writer
        // then blocks, its queue fills to depth 2, and the next broadcast drops
        // it. The fast client keeps draining so it never fills.
        let big = vec![0xABu8; 60_000];
        for _ in 0..60u32 {
            server
                .broadcast(encode_frame(&big, MAVLINK_MAX_FRAME).unwrap().into())
                .await;
            tokio::time::sleep(Duration::from_millis(2)).await;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;

        // The slow client has been pruned; the fast client remains.
        assert_eq!(server.client_count().await, 1);
        // The eviction is observable: the drop counter advanced by exactly the
        // one stalled consumer, not the fast one that kept draining.
        assert_eq!(server.dropped_clients(), 1);
        fast_reader.abort();
    }

    #[tokio::test]
    async fn dropped_client_counter_advances_when_a_consumer_stalls() {
        let path = temp_sock("dropcount");
        // Depth-2 queue so a non-draining client is evicted quickly once its
        // kernel send buffer and then its queue fill.
        let (server, _inbound) = IpcBroadcast::bind(&path, 2, false, None).await.unwrap();

        // A counter starts at zero before any client exists.
        assert_eq!(server.dropped_clients(), 0);

        // One consumer connects and then never reads a single byte.
        let _stalled = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert_eq!(server.client_count().await, 1);

        // Push large payloads until the stalled consumer's kernel buffer and
        // queue fill and the next broadcast evicts it.
        let big = vec![0x5Au8; 60_000];
        for _ in 0..60u32 {
            server
                .broadcast(encode_frame(&big, MAVLINK_MAX_FRAME).unwrap().into())
                .await;
            tokio::time::sleep(Duration::from_millis(2)).await;
            if server.client_count().await == 0 {
                break;
            }
        }
        tokio::time::sleep(Duration::from_millis(50)).await;

        // The stalled consumer was pruned, and the eviction is no longer
        // silent: the producer-visible counter advanced.
        assert_eq!(server.client_count().await, 0);
        assert_eq!(server.dropped_clients(), 1);

        // The counter is monotonic: it does not reset across further broadcasts
        // once the client is gone (no remaining client to drop).
        server
            .broadcast(encode_frame(b"tail", MAVLINK_MAX_FRAME).unwrap().into())
            .await;
        assert_eq!(server.dropped_clients(), 1);
    }

    #[tokio::test]
    async fn a_clean_disconnect_is_not_counted_as_a_slow_client() {
        // The state-socket shape: replay the last buffer on connect, which is
        // what a short-lived reader comes for.
        let path = temp_sock("cleandisconnect");
        let (server, _inbound) = IpcBroadcast::bind(&path, 32, true, None).await.unwrap();
        server
            .broadcast(encode_frame(b"snapshot", MAVLINK_MAX_FRAME).unwrap().into())
            .await;

        // This is exactly the `ados-cloud` enrichment pattern: connect, take the
        // replayed snapshot, disconnect. Repeated at ~1 Hz it used to log an
        // `ipc_slow_client_evicted` every second on a perfectly healthy agent.
        for _ in 0..3u32 {
            let client = connect_with_retry(&path, 10, Duration::from_millis(20))
                .await
                .unwrap();
            tokio::time::sleep(Duration::from_millis(30)).await;
            drop(client);
            tokio::time::sleep(Duration::from_millis(30)).await;

            // Two broadcasts: the first may still find the queue writable
            // before the writer task notices the peer is gone, the second is
            // guaranteed to meet a closed channel.
            for _ in 0..2u32 {
                server
                    .broadcast(encode_frame(b"tick", MAVLINK_MAX_FRAME).unwrap().into())
                    .await;
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        }

        // The departed clients are pruned...
        assert_eq!(server.client_count().await, 0);
        // ...but none of them was slow, so the eviction counter must be
        // untouched. Before the Full/Closed split this read 3.
        assert_eq!(server.dropped_clients(), 0);
    }

    #[tokio::test]
    async fn inbound_frames_from_client_reach_the_channel() {
        let path = temp_sock("inbound");
        let (_server, inbound) = IpcBroadcast::bind(&path, 256, false, Some(16))
            .await
            .unwrap();
        let mut inbound = inbound.expect("inbound channel requested");

        let mut client = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        // Client writes a framed command toward the FC.
        let framed = encode_frame(b"command", MAVLINK_MAX_FRAME).unwrap();
        client.write_all(&framed).await.unwrap();
        client.flush().await.unwrap();

        let got = tokio::time::timeout(Duration::from_millis(500), inbound.recv())
            .await
            .unwrap()
            .expect("a command");
        assert_eq!(got.payload, b"command");
        // The kernel answers for a local peer, so the writing process is
        // identified rather than assumed.
        assert!(
            got.peer.uid.is_some(),
            "the accepting socket must record who wrote the payload"
        );
        assert!(!got.peer.off_box_source);
    }

    #[tokio::test]
    async fn a_declared_off_box_source_sticks_and_is_not_forwarded() {
        let path = temp_sock("offbox");
        let (_server, inbound) = IpcBroadcast::bind(&path, 256, false, Some(16))
            .await
            .unwrap();
        let mut inbound = inbound.expect("inbound channel requested");

        let mut client = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        for payload in [IPC_DECLARE_OFF_BOX_SOURCE, b"first", b"second"] {
            let framed = encode_frame(payload, MAVLINK_MAX_FRAME).unwrap();
            client.write_all(&framed).await.unwrap();
        }
        client.flush().await.unwrap();

        // The declaration itself never reaches the consumer, and both payloads
        // behind it are marked for the whole life of the connection.
        for expected in [&b"first"[..], &b"second"[..]] {
            let got = tokio::time::timeout(Duration::from_millis(500), inbound.recv())
                .await
                .unwrap()
                .expect("a command");
            assert_eq!(got.payload, expected);
            assert!(
                got.peer.off_box_source,
                "a declared forwarder must not read as a local producer"
            );
        }
    }

    #[test]
    fn encode_injector_declaration_round_trips_through_the_parser() {
        // The producer's encoder and the router's parser must agree byte-for-byte,
        // with and without a ticket — a drift here silently un-arms the gate.
        for (id, ticket) in [("ai-mission", Some("abc.def")), ("autopilot-1", None)] {
            let frame = encode_injector_declaration(id, ticket);
            assert!(frame.starts_with(IPC_DECLARE_INJECTOR_PREFIX));
            assert_eq!(
                parse_injector_declaration(&frame),
                Some(InjectorClaim {
                    client_id: id.into(),
                    ticket: ticket.map(str::to_string),
                })
            );
        }
    }

    #[test]
    fn parse_injector_declaration_reads_id_and_optional_ticket() {
        // id only
        let mut d = IPC_DECLARE_INJECTOR_PREFIX.to_vec();
        d.extend_from_slice(b"ai-mission");
        assert_eq!(
            parse_injector_declaration(&d),
            Some(InjectorClaim {
                client_id: "ai-mission".into(),
                ticket: None
            })
        );
        // id + ticket
        let mut d2 = IPC_DECLARE_INJECTOR_PREFIX.to_vec();
        d2.extend_from_slice(b"ai-mission;ticket=abc.def");
        assert_eq!(
            parse_injector_declaration(&d2),
            Some(InjectorClaim {
                client_id: "ai-mission".into(),
                ticket: Some("abc.def".into())
            })
        );
        // an empty client id is not a claim
        let mut empty = IPC_DECLARE_INJECTOR_PREFIX.to_vec();
        empty.extend_from_slice(b";ticket=abc");
        assert_eq!(parse_injector_declaration(&empty), None);
        // an unrelated payload is not an injector declaration
        assert_eq!(parse_injector_declaration(b"command"), None);
        assert_eq!(parse_injector_declaration(IPC_DECLARE_OFF_BOX_SOURCE), None);
    }

    #[tokio::test]
    async fn a_declared_injector_sticks_and_is_not_forwarded() {
        let path = temp_sock("injector");
        let (_server, inbound) = IpcBroadcast::bind(&path, 256, false, Some(16))
            .await
            .unwrap();
        let mut inbound = inbound.expect("inbound channel requested");

        let mut client = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        let mut decl = IPC_DECLARE_INJECTOR_PREFIX.to_vec();
        decl.extend_from_slice(b"ai-mission;ticket=tkt");
        for payload in [decl.as_slice(), b"first", b"second"] {
            let framed = encode_frame(payload, MAVLINK_MAX_FRAME).unwrap();
            client.write_all(&framed).await.unwrap();
        }
        client.flush().await.unwrap();

        // The declaration never reaches the FC, and both payloads behind it carry
        // the sticky injector claim.
        for expected in [&b"first"[..], &b"second"[..]] {
            let got = tokio::time::timeout(Duration::from_millis(500), inbound.recv())
                .await
                .unwrap()
                .expect("a command");
            assert_eq!(got.payload, expected);
            let claim = got.peer.injector.expect("a declared injector claim");
            assert_eq!(claim.client_id, "ai-mission");
            assert_eq!(claim.ticket.as_deref(), Some("tkt"));
        }
    }

    #[tokio::test]
    async fn one_connection_declaring_does_not_mark_another() {
        let path = temp_sock("offbox-scope");
        let (_server, inbound) = IpcBroadcast::bind(&path, 256, false, Some(16))
            .await
            .unwrap();
        let mut inbound = inbound.expect("inbound channel requested");

        let mut declaring = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        declaring
            .write_all(&encode_frame(IPC_DECLARE_OFF_BOX_SOURCE, MAVLINK_MAX_FRAME).unwrap())
            .await
            .unwrap();
        declaring.flush().await.unwrap();

        let mut plain = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        plain
            .write_all(&encode_frame(b"local", MAVLINK_MAX_FRAME).unwrap())
            .await
            .unwrap();
        plain.flush().await.unwrap();

        let got = tokio::time::timeout(Duration::from_millis(500), inbound.recv())
            .await
            .unwrap()
            .expect("a command");
        assert_eq!(got.payload, b"local");
        assert!(!got.peer.off_box_source);
    }

    #[tokio::test]
    async fn read_newline_line_caps_an_endless_line() {
        // A peer that never sends a newline must not grow the buffer past the
        // cap: the reader returns InvalidData instead of allocating unbounded.
        let (mut a, mut b) = UnixStream::pair().unwrap();
        let writer = tokio::spawn(async move {
            // Send more bytes than the cap, with no newline.
            let chunk = vec![b'x'; 64];
            for _ in 0..10 {
                if a.write_all(&chunk).await.is_err() {
                    break;
                }
                let _ = a.flush().await;
            }
            // Keep the stream open so the reader hits the cap, not EOF.
            tokio::time::sleep(Duration::from_secs(2)).await;
        });

        let err = read_newline_line(&mut b, 128).await.unwrap_err();
        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
        writer.abort();
    }

    #[tokio::test]
    async fn bind_command_socket_removes_stale_and_sets_mode() {
        let path = temp_sock("bindhelper");
        // Leave a stale regular file where the socket will go; bind must remove
        // it rather than fail with EADDRINUSE.
        std::fs::write(&path, b"stale").unwrap();
        assert!(path.exists());

        let listener = bind_command_socket(&path, 0o660).unwrap();

        // The requested permission bits are set on the bound socket.
        let mode = std::fs::metadata(&path).unwrap().permissions().mode();
        assert_eq!(mode & 0o777, 0o660);

        drop(listener);
        let _ = std::fs::remove_file(&path);
    }

    #[tokio::test]
    async fn a_broadcast_socket_is_never_world_accessible() {
        // Regression guard. These sockets were bound 0o666 to match an older
        // Python server, and two of them — the MAVLink and MSP lanes — forward
        // inbound bytes straight to the flight controller, so world-write meant
        // any local process could arm an aircraft. The trusted local plane is
        // root plus the operator group and nothing wider.
        let path = temp_sock("bcast-perm");
        let (server, _inbound) = IpcBroadcast::bind(&path, 8, false, None).await.unwrap();

        let mode = std::fs::metadata(&path).unwrap().permissions().mode();
        assert_eq!(
            mode & 0o007,
            0,
            "broadcast socket must carry no world bits, got {:o}",
            mode & 0o777
        );
        assert_eq!(mode & 0o777, 0o660, "expected owner+group rw only");

        drop(server);
        let _ = std::fs::remove_file(&path);
    }

    #[tokio::test]
    async fn serve_rpc_round_trips_a_request_and_closes_over_cap() {
        let path = temp_sock("rpc");
        let listener = bind_command_socket(&path, 0o660).unwrap();
        // Echo handler: prefix the request so the test sees the request bytes
        // (trailing newline stripped) reached the dispatch closure.
        let server = tokio::spawn(serve_rpc(listener, 16, |req: Vec<u8>| async move {
            let mut resp = b"echo:".to_vec();
            resp.extend_from_slice(&req);
            resp
        }));

        // A well-formed one-shot request gets one newline-terminated response.
        let mut client = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        client.write_all(b"ping\n").await.unwrap();
        client.flush().await.unwrap();
        let line = read_newline_line(&mut client, 256).await.unwrap();
        assert_eq!(line.as_deref(), Some(&b"echo:ping"[..]));
        drop(client);

        // A request longer than the cap (16) with no newline closes the
        // connection with no response. The client then reads no reply: on
        // some platforms that is a clean EOF (`Ok(None)`), on others the
        // server closing with unread data resets the connection (`Err`).
        // Both mean "closed without a reply".
        let mut over = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        let _ = over.write_all(&[b'x'; 64]).await;
        let _ = over.flush().await;
        let outcome = read_newline_line(&mut over, 256).await;
        assert!(
            matches!(outcome, Ok(None) | Err(_)),
            "over-cap request must get no reply, got {outcome:?}"
        );

        server.abort();
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn plugin_plane_peers_are_injectors_whatever_they_declare() {
        let plugin_gid = Some(980);
        let legacy_user = Some(995);
        // A per-plugin user: primary group is the plugin group.
        assert!(is_plugin_peer(1201, &[980], legacy_user, plugin_gid));
        // The plugin group as a supplementary group counts too.
        assert!(is_plugin_peer(1201, &[1201, 980], legacy_user, plugin_gid));
        // The legacy shared plugin user, whatever its groups.
        assert!(is_plugin_peer(995, &[995], legacy_user, plugin_gid));
        // Root and an operator are not.
        assert!(!is_plugin_peer(0, &[0], legacy_user, plugin_gid));
        assert!(!is_plugin_peer(1000, &[1000, 990], legacy_user, plugin_gid));
        // Neither the user nor the group exists on this host: nobody matches.
        assert!(!is_plugin_peer(1201, &[980], None, None));
        // The claim a plugin peer starts with can never verify on a paired node.
        assert_eq!(plugin_plane_claim(1201).ticket, None);
    }

    #[test]
    fn operator_peer_policy_admits_root_the_server_uid_and_operator_members_only() {
        let operator_gid = Some(990);
        // A plugin process: uid 1000, primary group 1000, no operator group.
        assert!(!operator_peer_allowed(1000, &[1000], 0, operator_gid));
        // The same uid with the operator group as a supplementary group.
        assert!(operator_peer_allowed(1000, &[1000, 990], 0, operator_gid));
        // The operator group as the primary group counts too.
        assert!(operator_peer_allowed(1000, &[990], 0, operator_gid));
        // Root and the serving process's own uid are always admitted.
        assert!(operator_peer_allowed(0, &[0], 0, operator_gid));
        assert!(operator_peer_allowed(501, &[20], 501, operator_gid));
        // With the operator group absent from the host nobody else gets in,
        // whatever groups it carries.
        assert!(!operator_peer_allowed(1000, &[1000, 990], 0, None));
    }

    #[tokio::test]
    async fn a_refused_peer_is_closed_before_its_request_is_read() {
        // Root is admitted by rule, so this can only be observed as non-root.
        if nix::unistd::geteuid().is_root() {
            return;
        }
        let path = temp_sock("refused");
        let mut listener = bind_command_socket(&path, 0o660).unwrap();
        // Serve as a different uid with no operator group on the host: the
        // test's own uid is then an outsider, exactly like a plugin process.
        listener.server_uid = u32::MAX - 1;
        listener.operator_gid = None;
        let reached = Arc::new(AtomicU64::new(0));
        let counter = reached.clone();
        let server = tokio::spawn(serve_rpc(listener, 64, move |_req: Vec<u8>| {
            let counter = counter.clone();
            async move {
                counter.fetch_add(1, Ordering::Relaxed);
                b"served".to_vec()
            }
        }));

        let mut client = connect_with_retry(&path, 10, Duration::from_millis(20))
            .await
            .unwrap();
        let _ = client.write_all(b"ping\n").await;
        let _ = client.flush().await;
        let outcome =
            tokio::time::timeout(Duration::from_secs(2), read_newline_line(&mut client, 256))
                .await
                .expect("a refused peer must be closed, not left hanging");
        assert!(
            matches!(outcome, Ok(None) | Err(_)),
            "a refused peer must get no reply, got {outcome:?}"
        );
        assert_eq!(
            reached.load(Ordering::Relaxed),
            0,
            "the handler must never run"
        );

        server.abort();
        let _ = std::fs::remove_file(&path);
    }
}