cellos_supervisor/

ebpf_flow.rs

//! E7-4 — eBPF host-side flow monitor (aya loader + ring-buffer drainer).
//!
//! This module ships two things:
//!
//! 1. **The original FC-38 Phase 2 scaffold** ([`FlowEvent`],
//!    [`FlowEventListener`], [`NoopFlowListener`], the env-flag reader, the
//!    reason-code constants, and the [`flow_event_to_network_flow_decision`]
//!    pure translator). These are preserved verbatim — downstream callers
//!    in `per_flow.rs`, `supervisor.rs`, and external integration tests
//!    consume them, and the scaffold's unit tests stay green.
//!
//! 2. **The new [`EbpfFlowMonitor`] type** — the host-side aya loader the
//!    supervisor activates when `CELLOS_PER_FLOW_REALTIME=1` AND
//!    `CELLOS_PER_FLOW_BACKEND=ebpf`. On Linux with the `ebpf-aya` Cargo
//!    feature enabled it `setns(2)`'s into the cell's network namespace,
//!    loads a tc-clsact egress program from an operator-supplied BPF
//!    object, attaches it to `tap_iface`, spawns a tokio task draining a
//!    ring buffer into an mpsc channel, and returns to the original
//!    netns. On macOS, on Linux without the feature, or when any of the
//!    above fails, it returns an [`EbpfMonitorError`] that the supervisor
//!    treats as "fall back to nflog" — never a fatal error.
//!
//! ## Why a separate type rather than another [`FlowEventListener`] impl
//!
//! The scaffold's [`FlowEventListener`] trait surfaces a single
//! `drain_events` hook that the supervisor calls AFTER `child.wait()`
//! returns, in the same `spawn_blocking` closure as the nft counter
//! scrape. Ring-buffer-based eBPF flow monitors need to drain
//! *continuously* (kernel ring buffers overflow if left unread) on a
//! dedicated tokio task. The [`EbpfFlowMonitor::drain`] semantics
//! (non-blocking pull from an mpsc channel) match that lifecycle far
//! better than the trait's single-shot post-run hook. Both surfaces can
//! coexist — the supervisor selects which one to use based on
//! [`crate::per_flow::PerFlowBackend`].
//!
//! ## Honest scope under macOS
//!
//! - `[EbpfFlowMonitor::start]` always returns
//!   [`EbpfMonitorError::Unsupported`] on `cfg(not(target_os = "linux"))`.
//! - On Linux without the `ebpf-aya` Cargo feature, `start` returns
//!   [`EbpfMonitorError::FeatureDisabled`].
//! - On Linux with the feature but no BPF object available
//!   (`CELLOS_EBPF_OBJECT_PATH` unset / file missing), `start` returns
//!   [`EbpfMonitorError::ObjectMissing`].
//! - Real kernel attachment + ring-buffer drain are only exercised when
//!   all three conditions are met.
//!
//! The supervisor's call site (`per_flow.rs`) catches any
//! [`EbpfMonitorError`], emits a single `tracing::warn`, and reverts to
//! the existing nflog path so per-flow telemetry is never silently lost.

#![allow(dead_code)] // scaffold + feature-gated implementation; consumers vary

use cellos_core::{NetworkFlowDecision, NetworkFlowDecisionOutcome, NetworkFlowDirection};

// ────────────────────────────────────────────────────────────────────────
// Original FC-38 Phase 2 scaffold (preserved for downstream callers)
// ────────────────────────────────────────────────────────────────────────

/// Operator-opt-in env flag for FC-38 Phase 2 eBPF/nflog real-time
/// per-flow events.
///
/// Set to `1` to enable. Default OFF: Phase 2 listener has kernel
/// requirements (CAP_BPF, kernel ≥ 5.8 for full BPF surface, or
/// CAP_NET_ADMIN for nflog) that Phase 1's post-run nft scrape does not.
/// Operators must explicitly opt in to acknowledge they have provisioned
/// the right capability surface.
pub const PER_FLOW_EBPF_ENV: &str = "CELLOS_FIRECRACKER_PER_FLOW_EBPF";

/// Phase 2 reason code: an eBPF-attached program (likely
/// `cgroup_skb/egress` or `tc clsact`) observed an XDP-style drop on the
/// cell's egress.
pub const REASON_EBPF_XDP_DROP: &str = "ebpf_xdp_drop";

/// Phase 2 reason code: an `NFLOG` netlink consumer observed a packet
/// that matched a logging-enabled nft rule.
pub const REASON_NFLOG_MATCH: &str = "nflog_match";

/// Returns `true` when the operator has explicitly opted into Phase 2
/// eBPF/nflog real-time per-flow events via the
/// [`PER_FLOW_EBPF_ENV`] env var.
pub fn is_per_flow_ebpf_enabled() -> bool {
    std::env::var(PER_FLOW_EBPF_ENV).as_deref() == Ok("1")
}

/// One real-time observation a Phase 2 listener surfaces.
///
/// Backend-neutral: an eBPF `cgroup_skb` program populates the same
/// fields a `nflog` netlink consumer does, AND the new ring-buffer-based
/// [`EbpfFlowMonitor`] populates the same fields.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FlowEvent {
    /// Outbound vs inbound. Phase 2 starts on egress (matching Phase 1)
    /// and may add ingress later.
    pub direction: NetworkFlowDirection,
    /// Allow vs deny.
    pub decision: NetworkFlowDecisionOutcome,
    /// Reason code distinguishing the data source.
    pub reason_code: &'static str,
    /// Optional destination IP.
    pub dst_addr: Option<String>,
    /// Optional destination port.
    pub dst_port: Option<u16>,
    /// Optional protocol (`udp` / `tcp` / `icmp`).
    pub protocol: Option<String>,
    /// Optional bytes observed for this event.
    pub byte_count: Option<u64>,
}

/// Trait the supervisor calls on its hot path to drain pending events
/// from a Phase 2 listener.
pub trait FlowEventListener: Send + Sync {
    /// Drain events accumulated since the last call.
    fn drain_events(&mut self) -> Vec<FlowEvent>;

    /// Identifier the supervisor stamps into the audit log.
    fn backend_name(&self) -> &'static str;
}

/// Default Phase 2 listener: yields zero events, identifies as `"noop"`.
#[derive(Debug, Default)]
pub struct NoopFlowListener;

impl FlowEventListener for NoopFlowListener {
    fn drain_events(&mut self) -> Vec<FlowEvent> {
        Vec::new()
    }

    fn backend_name(&self) -> &'static str {
        "noop"
    }
}

/// Build the default scaffold listener — always [`NoopFlowListener`] in
/// the scaffolding pass.
pub fn build_default_listener() -> Box<dyn FlowEventListener> {
    Box::new(NoopFlowListener)
}

/// Translate a [`FlowEvent`] into a [`NetworkFlowDecision`] payload
/// suitable for emitting on the existing
/// `dev.cellos.events.cell.observability.v1.network_flow_decision` channel.
#[allow(clippy::too_many_arguments)]
pub fn flow_event_to_network_flow_decision(
    event: &FlowEvent,
    cell_id: &str,
    run_id: &str,
    decision_id: &str,
    policy_digest: Option<&str>,
    keyset_id: Option<&str>,
    issuer_kid: Option<&str>,
    correlation_id: Option<&str>,
    observed_at: &str,
) -> NetworkFlowDecision {
    NetworkFlowDecision {
        schema_version: "1.0.0".into(),
        cell_id: cell_id.to_string(),
        run_id: run_id.to_string(),
        decision_id: decision_id.to_string(),
        direction: event.direction,
        decision: event.decision,
        reason_code: event.reason_code.to_string(),
        nft_rule_ref: None,
        dst_addr: event.dst_addr.clone(),
        dst_port: event.dst_port,
        protocol: event.protocol.clone(),
        packet_count: event.byte_count.map(|_| 1u64),
        byte_count: event.byte_count,
        policy_digest: policy_digest.map(|s| s.to_string()),
        keyset_id: keyset_id.map(|s| s.to_string()),
        issuer_kid: issuer_kid.map(|s| s.to_string()),
        correlation_id: correlation_id.map(|s| s.to_string()),
        observed_at: observed_at.to_string(),
    }
}

// ────────────────────────────────────────────────────────────────────────
// E7-4 — EbpfFlowMonitor (host-side aya loader)
// ────────────────────────────────────────────────────────────────────────

/// Errors returned by [`EbpfFlowMonitor::start`].
///
/// All variants are non-fatal at the supervisor level: the per-flow
/// activation path (`per_flow::activate_per_flow_listener`) catches them,
/// logs a `tracing::warn`, and falls back to the nflog listener so the
/// cell still emits `network_flow_decision` events on the existing path.
#[derive(Debug, thiserror::Error)]
pub enum EbpfMonitorError {
    /// Host platform is not Linux. eBPF requires Linux kernel support.
    #[error("eBPF flow monitor is Linux-only (host platform: {host})")]
    Unsupported { host: &'static str },

    /// Linux host but the crate was built without the `ebpf-aya` Cargo
    /// feature. The supervisor binary ships with the feature off by
    /// default so macOS / non-eBPF Linux builds don't pay for a kernel
    /// dependency they cannot use.
    #[error(
        "eBPF flow monitor requested but the `ebpf-aya` Cargo feature is \
         not enabled on this build"
    )]
    FeatureDisabled,

    /// The BPF object file referenced by `CELLOS_EBPF_OBJECT_PATH` (or
    /// the default path) was not present on disk. Real deployments
    /// pre-build `cellos-supervisor-ebpf` via `cargo bpf` and stage the
    /// resulting object alongside the supervisor binary; the env var
    /// lets operators override the location.
    #[error("eBPF object file missing at {path}")]
    ObjectMissing { path: String },

    /// `setns(2)` into the cell's network namespace failed. Usually
    /// `EPERM` (missing `CAP_SYS_ADMIN`) or `ENOENT` (cell already torn
    /// down).
    #[error("setns(CLONE_NEWNET) into cell netns failed: {source}")]
    Setns {
        #[source]
        source: std::io::Error,
    },

    /// Loading the BPF object via aya failed. Surfaces verifier
    /// rejections, license issues, missing program sections, etc.
    #[error("aya BPF object load failed: {message}")]
    LoadFailed { message: String },

    /// Attaching the loaded program to the tap interface via
    /// `tc clsact` failed. Usually `ENODEV` (tap iface missing) or
    /// `EEXIST` (clsact qdisc collision with an operator-provisioned
    /// qdisc on the same interface).
    #[error("tc clsact attach to {iface} failed: {message}")]
    AttachFailed { iface: String, message: String },

    /// Opening the ring-buffer map for draining failed.
    #[error("ring-buffer map open failed: {message}")]
    RingBufferOpenFailed { message: String },
}

/// Host-side eBPF flow monitor.
///
/// Owns the loaded aya BPF object (so it stays alive for the duration of
/// the cell run — dropping `aya::Ebpf` detaches all programs) plus the
/// receiver end of an mpsc channel into which the ring-buffer drainer
/// task pushes [`FlowEvent`]s.
///
/// Lifecycle:
///
/// 1. [`EbpfFlowMonitor::start`] enters the cell's netns, loads + attaches
///    the BPF program, spawns the drainer task, and returns to the original
///    netns.
/// 2. The supervisor periodically calls [`EbpfFlowMonitor::drain`] (or
///    consumes events directly via the channel) on its hot path.
/// 3. [`EbpfFlowMonitor::stop`] aborts the drainer task and drops the
///    aya object, which detaches the BPF program and releases the
///    ring-buffer map.
///
/// The struct is constructed only via `start`; `Default` is intentionally
/// not provided to prevent callers from skipping the netns + load
/// sequence.
pub struct EbpfFlowMonitor {
    /// Holds the loaded eBPF program + ring buffer FD on Linux with the
    /// aya feature enabled. `None` on all other configurations — the
    /// monitor in that case is a thin wrapper around an empty channel
    /// (the constructor returns `Err` so a `None` here at runtime is a
    /// programmer error, not a deployment scenario).
    #[cfg(all(target_os = "linux", feature = "ebpf-aya"))]
    _bpf: Option<aya::Ebpf>,

    /// Receiver end of the mpsc channel the drainer task pushes events
    /// onto. Construction always produces a paired sender (held by the
    /// drainer task) + receiver (held here); dropping `_bpf` cancels the
    /// drainer task which closes the sender side, signalling EOF.
    rx: tokio::sync::mpsc::UnboundedReceiver<FlowEvent>,

    /// Optional handle to the drainer task so [`Self::stop`] can abort
    /// it deterministically rather than waiting for the channel close.
    drainer: Option<tokio::task::JoinHandle<()>>,

    /// Cached interface name for tracing breadcrumbs.
    iface: String,
}

impl EbpfFlowMonitor {
    /// Load and attach the BPF program inside the cell's netns. Falls
    /// back to nflog (via the caller in `per_flow.rs`) if any step
    /// fails by returning an [`EbpfMonitorError`].
    ///
    /// `netns_fd` MUST be an open file descriptor for the cell's
    /// `/proc/<child_pid>/ns/net`. The caller retains ownership.
    /// `tap_iface` is the supervisor-provisioned tap interface name
    /// (e.g. `"vethC0"`).
    #[allow(unused_variables)] // platform-dependent path
    pub async fn start(
        netns_fd: std::os::unix::io::RawFd,
        tap_iface: &str,
    ) -> Result<Self, EbpfMonitorError> {
        #[cfg(all(target_os = "linux", feature = "ebpf-aya"))]
        {
            Self::start_linux_aya(netns_fd, tap_iface).await
        }
        #[cfg(all(target_os = "linux", not(feature = "ebpf-aya")))]
        {
            Err(EbpfMonitorError::FeatureDisabled)
        }
        #[cfg(not(target_os = "linux"))]
        {
            // tap_iface is intentionally unused on non-Linux — surface
            // it through the error message so the operator's log line
            // identifies which cell's monitor declined to start.
            tracing::debug!(
                target: "cellos.supervisor.ebpf_flow",
                iface = %tap_iface,
                "EbpfFlowMonitor::start on non-Linux host — falling back to nflog"
            );
            Err(EbpfMonitorError::Unsupported {
                host: std::env::consts::OS,
            })
        }
    }

    /// Drain pending flow events into a `Vec` without blocking.
    ///
    /// Used by callers that want a synchronous batch pull instead of
    /// `.recv().await` per event. Returns an empty `Vec` when no events
    /// are buffered.
    pub fn drain(&mut self) -> Vec<FlowEvent> {
        let mut out = Vec::new();
        while let Ok(ev) = self.rx.try_recv() {
            out.push(ev);
        }
        out
    }

    /// Async pull of one event. Returns `None` when the drainer task
    /// has exited (i.e. the channel sender side has been dropped).
    pub async fn recv(&mut self) -> Option<FlowEvent> {
        self.rx.recv().await
    }

    /// Stop the monitor and clean up. Aborts the drainer task; dropping
    /// the held [`aya::Ebpf`] object detaches the BPF program.
    pub async fn stop(mut self) {
        if let Some(drainer) = self.drainer.take() {
            drainer.abort();
            // Best-effort join — we don't care about the result, only
            // that the task has stopped scheduling.
            let _ = drainer.await;
        }
        // `_bpf` (when present) drops here, releasing kernel resources.
    }

    // ────────────────────────────────────────────────────────────────
    // Linux + aya implementation
    // ────────────────────────────────────────────────────────────────

    #[cfg(all(target_os = "linux", feature = "ebpf-aya"))]
    async fn start_linux_aya(
        netns_fd: std::os::unix::io::RawFd,
        tap_iface: &str,
    ) -> Result<Self, EbpfMonitorError> {
        // The BPF object path is operator-controllable so deployments
        // can stage the object wherever their packaging puts it. The
        // default path matches the cellos-supervisor-ebpf crate's
        // expected cargo-bpf output location.
        let object_path = std::env::var("CELLOS_EBPF_OBJECT_PATH")
            .ok()
            .unwrap_or_else(|| "/usr/local/lib/cellos/cellos-supervisor-ebpf.o".to_string());
        let object_bytes =
            std::fs::read(&object_path).map_err(|_| EbpfMonitorError::ObjectMissing {
                path: object_path.clone(),
            })?;

        // Save the supervisor's original netns so we can return to it
        // after attaching the program. `setns` is per-thread on Linux,
        // so we MUST restore before the tokio worker thread we're on
        // services any other task.
        let self_netns = std::fs::File::open("/proc/self/ns/net")
            .map_err(|source| EbpfMonitorError::Setns { source })?;

        // Enter the cell's netns.
        use std::os::unix::io::AsRawFd;
        let setns_rc = unsafe { libc::setns(netns_fd, libc::CLONE_NEWNET) };
        if setns_rc != 0 {
            return Err(EbpfMonitorError::Setns {
                source: std::io::Error::last_os_error(),
            });
        }

        // Load the BPF object inside the cell's netns. aya 0.13's
        // `Ebpf::load` returns a `Result<Ebpf, EbpfError>` — surface
        // its display string into our error type so the caller's
        // tracing line carries the verifier message verbatim.
        let load_result = aya::Ebpf::load(&object_bytes);

        // Restore the supervisor's netns BEFORE we propagate any
        // load error — leaving the worker thread in the cell's netns
        // would break unrelated tokio tasks scheduled later.
        let restore_rc = unsafe { libc::setns(self_netns.as_raw_fd(), libc::CLONE_NEWNET) };
        if restore_rc != 0 {
            // Restoration failure is a real problem (this thread is
            // now stuck in the cell's netns). Surface as Setns rather
            // than discarding — the supervisor will refuse to use
            // this monitor and the caller falls back to nflog.
            return Err(EbpfMonitorError::Setns {
                source: std::io::Error::last_os_error(),
            });
        }

        let mut bpf = load_result.map_err(|e| EbpfMonitorError::LoadFailed {
            message: format!("{e}"),
        })?;

        // tc clsact attach. The program name is a contract with the
        // ebpf crate's source; we don't hardcode it as a constant
        // here because the ebpf crate ships its own constants module
        // — when the ebpf crate lands the supervisor will import
        // `cellos_supervisor_ebpf::PROGRAM_NAME` instead. For now we
        // probe a small set of conventional names so a partial BPF
        // crate doesn't force a supervisor recompile.
        const CANDIDATE_PROGRAM_NAMES: &[&str] =
            &["cellos_flow_egress", "tc_egress", "flow_classifier"];
        let mut attached = false;
        let mut last_attach_err: Option<String> = None;
        for prog_name in CANDIDATE_PROGRAM_NAMES {
            // aya 0.13: `program_mut` returns Option; tc programs need
            // `SchedClassifier::try_from` + `.load()` + `.attach()`.
            // We avoid hard-coding the call shape against an unloaded
            // feature dep by recording the attempt; the actual probe
            // and attach happens via the helper below so this file
            // still compiles when aya's surface evolves.
            match attach_tc_egress(&mut bpf, prog_name, tap_iface) {
                Ok(()) => {
                    attached = true;
                    break;
                }
                Err(e) => {
                    last_attach_err = Some(e);
                }
            }
        }
        if !attached {
            return Err(EbpfMonitorError::AttachFailed {
                iface: tap_iface.to_string(),
                message: last_attach_err
                    .unwrap_or_else(|| "no candidate program found".to_string()),
            });
        }

        // Spawn the ring-buffer drainer. We pull the ring-buffer map
        // by name (`"flow_events"`); when the ebpf crate lands it will
        // export this constant.
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<FlowEvent>();
        let drainer = spawn_ring_buffer_drainer(&mut bpf, tx)
            .map_err(|e| EbpfMonitorError::RingBufferOpenFailed { message: e })?;

        Ok(Self {
            _bpf: Some(bpf),
            rx,
            drainer: Some(drainer),
            iface: tap_iface.to_string(),
        })
    }
}

// ────────────────────────────────────────────────────────────────────
// Linux+aya helpers — intentionally narrow so this file stays
// compilable on macOS regardless of aya's API drift.
// ────────────────────────────────────────────────────────────────

/// Attach the loaded eBPF `SchedClassifier` program to `iface` on the tc
/// clsact egress hook.
///
/// Steps (mirrors aya 0.13's documented attach flow):
///   1. Ensure a `clsact` qdisc exists on the interface — required before
///      a `SchedClassifier` can attach. `qdisc_add_clsact` is idempotent
///      from our point of view: if the qdisc already exists we treat
///      `EEXIST` as success because some operator setups pre-provision
///      `clsact` via `tc` and we should not collide.
///   2. Pull the program by name from the loaded Ebpf object and coerce
///      it into a `SchedClassifier` via `TryInto`.
///   3. `program.load()` — finalises verifier-side validation.
///   4. `program.attach(iface, TcAttachType::Egress)` — wires it to the
///      egress hook. The returned `LinkId` is dropped here; the link's
///      kernel-side lifetime is tied to the parent `aya::Ebpf` object,
///      which the caller (`EbpfFlowMonitor`) holds for the duration of
///      the cell run. Dropping `aya::Ebpf` detaches the link.
///
/// All error paths surface as `Err(String)` so the caller's candidate-name
/// loop can record the failure and try the next program name without
/// having to thread an `aya::ProgramError` through the signature.
#[cfg(all(target_os = "linux", feature = "ebpf-aya"))]
fn attach_tc_egress(bpf: &mut aya::Ebpf, program_name: &str, iface: &str) -> Result<(), String> {
    use aya::programs::{tc, SchedClassifier, TcAttachType};

    // Step 1 — clsact qdisc. EEXIST is benign (operator may have
    // pre-provisioned it, or a prior cell on the same shared iface left
    // it behind). Any other error we surface so the caller can fall back
    // to nflog.
    if let Err(e) = tc::qdisc_add_clsact(iface) {
        let msg = format!("{e}");
        // The error type wraps an io::Error; the only string-safe way to
        // detect "already exists" portably across aya versions is a
        // substring match on the rendered message. False negatives just
        // mean we surface a real EEXIST as an attach failure and the
        // supervisor falls back to nflog — non-fatal.
        if !msg.contains("File exists") && !msg.contains("EEXIST") {
            return Err(format!("qdisc_add_clsact({iface}) failed: {msg}"));
        }
    }

    // Step 2 — locate the program. Missing-program is the most common
    // outcome when the BPF object was built without this classifier name
    // exported; surface a stable string so the caller's loop can move on.
    let program_handle = bpf
        .program_mut(program_name)
        .ok_or_else(|| format!("program {program_name:?} not found in BPF object"))?;
    let program: &mut SchedClassifier =
        program_handle
            .try_into()
            .map_err(|e: aya::programs::ProgramError| {
                format!("program {program_name:?} is not a SchedClassifier: {e}")
            })?;

    // Step 3 — verifier-side load. Idempotent inside aya: calling load()
    // twice on the same program returns `AlreadyLoaded`, which we treat
    // as benign so callers retrying after a transient attach error
    // don't have to track per-program state.
    if let Err(e) = program.load() {
        let msg = format!("{e}");
        if !msg.contains("AlreadyLoaded") && !msg.contains("already loaded") {
            return Err(format!("SchedClassifier::load failed: {msg}"));
        }
    }

    // Step 4 — attach to egress. The LinkId is owned by `program` (i.e.
    // by the parent `aya::Ebpf` object) and is dropped when the monitor
    // is stopped.
    program
        .attach(iface, TcAttachType::Egress)
        .map_err(|e| format!("SchedClassifier::attach({iface}, egress) failed: {e}"))?;

    Ok(())
}

/// Wire layout of one BPF-side `FlowEvent` (must mirror
/// `crates/cellos-supervisor-ebpf/src/main.rs`).
///
/// Layout summary (40 bytes total, `#[repr(C)]` natural alignment):
///   - bytes 0..4   src_ip   (u32, host order — BPF program converts from
///                  big-endian before writing)
///   - bytes 4..8   dst_ip   (u32, host order)
///   - bytes 8..10  src_port (u16, host order)
///   - bytes 10..12 dst_port (u16, host order)
///   - byte  12     proto    (u8, IANA proto byte)
///   - bytes 13..16 _pad     (3 bytes, ignored)
///   - bytes 16..24 pkt_count    (u64)
///   - bytes 24..32 first_seen_ns (u64)
///   - byte  32     verdict  (u8)
///   - bytes 33..40 _pad     (7 bytes, ignored)
///
/// Total: 40 bytes. This constant pins the parser against accidental
/// drift between the BPF crate's `FlowEvent` and the host-side decoder.
#[cfg(all(target_os = "linux", feature = "ebpf-aya"))]
const BPF_FLOW_EVENT_SIZE: usize = 40;

/// Spawn the tokio task that drains the `FLOW_EVENTS` ring buffer.
///
/// We use `take_map` rather than `map_mut` so the resulting `RingBuf`
/// owns its `MapData` and can move into a `spawn_blocking` closure
/// without holding a `&mut aya::Ebpf` across the await point. The
/// supervisor's `aya::Ebpf` retains every OTHER map (notably `FLOWS`)
/// and the loaded program; only the ring buffer migrates ownership.
///
/// The drainer loop:
///   1. Calls `ring_buf.next()`. On `Some(item)`, parses the bytes via
///      `parse_ring_buf_event` and forwards the resulting `FlowEvent`
///      down `tx`. If the receiver has been dropped (caller stopped the
///      monitor), the send fails and the task exits cleanly.
///   2. On `None`, sleeps 100 µs and retries. This is intentionally
///      coarse: ring-buffer polling is a hot loop and a tight spin would
///      waste a worker thread, while a longer sleep would inflate
///      first-flow latency. 100 µs gives ~10k poll iterations/sec which
///      comfortably tracks any realistic per-cell flow rate.
///
/// The returned `JoinHandle` is held by `EbpfFlowMonitor::drainer` so
/// `stop()` can `abort()` the task deterministically.
#[cfg(all(target_os = "linux", feature = "ebpf-aya"))]
fn spawn_ring_buffer_drainer(
    bpf: &mut aya::Ebpf,
    tx: tokio::sync::mpsc::UnboundedSender<FlowEvent>,
) -> Result<tokio::task::JoinHandle<()>, String> {
    use aya::maps::RingBuf;

    // The BPF crate names the map `FLOW_EVENTS`. aya 0.13 exposes both
    // the original lowercase name and the macro-stamped uppercase name
    // through the symbol table; we try the canonical name first, then
    // fall back to the legacy lowercase form so a BPF crate rebuild
    // with a different naming convention doesn't break the supervisor.
    let map = bpf
        .take_map("FLOW_EVENTS")
        .or_else(|| bpf.take_map("flow_events"))
        .ok_or_else(|| "FLOW_EVENTS map not found in BPF object".to_string())?;
    let mut ring_buf: RingBuf<aya::maps::MapData> =
        RingBuf::try_from(map).map_err(|e| format!("FLOW_EVENTS map is not a ring buffer: {e}"))?;

    let handle = tokio::task::spawn_blocking(move || {
        // Tight-but-yielding poll loop. We deliberately do NOT register
        // the ring-buffer FD with epoll/AsyncFd here: the supervisor's
        // tokio runtime is a multi-threaded reactor and the cost of one
        // dedicated blocking thread per cell is far smaller than the
        // contract surface of an AsyncFd-based wakeup path (which would
        // require holding `&mut RingBuf` across `.await`). 100 µs sleeps
        // bound the busy-poll overhead while keeping new-flow latency
        // well under 1 ms.
        loop {
            match ring_buf.next() {
                Some(item) => {
                    if let Some(event) = parse_ring_buf_event(&item) {
                        if tx.send(event).is_err() {
                            // Receiver dropped — monitor stopped. Exit.
                            return;
                        }
                    }
                    // Item drops here, advancing the consumer position.
                }
                None => {
                    std::thread::sleep(std::time::Duration::from_micros(100));
                }
            }
        }
    });

    Ok(handle)
}

/// Decode the BPF-side `FlowEvent` byte layout into the host-side
/// [`FlowEvent`].
///
/// Returns `None` if the slice is shorter than the expected 40 bytes
/// (a truncated ring-buffer write should never happen in practice, but
/// we treat it as a soft failure rather than a panic so a single
/// malformed event cannot crash the supervisor).
///
/// The BPF program writes `src_ip` / `dst_ip` / ports in HOST byte order
/// (it applies `u32::from_be` / `u16::from_be` before writing into the
/// map key — see `try_track` in the BPF crate), so we read them with
/// native-endian decoding here.
///
/// ## Relationship to [`decode_ring_buf_event`]
///
/// This function is the LIVE-PATH adapter the Linux+aya drainer calls
/// per ring-buffer slot. It returns `Option<FlowEvent>` (silent drop on
/// short reads) because the drainer cannot afford a typed error per
/// slot on the hot path.
///
/// The platform-independent decoder [`decode_ring_buf_event`] returns a
/// typed [`RingBufFlowEvent`] + [`RingBufParseError`] and is used by
/// the unit-test suite (which runs on every host, including macOS),
/// plus any future caller that needs the full kernel-side layout
/// (pkt_count, first_seen_ns, verdict — fields this adapter discards).
/// Both decoders agree on the wire layout; the layout-pin test
/// `ring_buf_event_layout_constant_matches_producer` keeps them in
/// lockstep against the BPF crate's `#[repr(C)]` producer struct.
#[cfg(all(target_os = "linux", feature = "ebpf-aya"))]
fn parse_ring_buf_event(bytes: &[u8]) -> Option<FlowEvent> {
    use std::net::Ipv4Addr;

    if bytes.len() < BPF_FLOW_EVENT_SIZE {
        return None;
    }

    // FlowKey — bytes 0..16.
    let src_ip = u32::from_ne_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
    let dst_ip = u32::from_ne_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]);
    let _src_port = u16::from_ne_bytes([bytes[8], bytes[9]]);
    let dst_port = u16::from_ne_bytes([bytes[10], bytes[11]]);
    let proto = bytes[12];
    // bytes 13..16: padding, ignored.

    // FlowEntry — bytes 16..40. Decoded but currently unused:
    //   - pkt_count (bytes 16..24): observability metadata; the host-side
    //     `FlowEvent` shape carries `byte_count`, not `packet_count`, and
    //     the BPF program does not track bytes today. Leaving both `None`
    //     beats fabricating a value.
    //   - first_seen_ns (bytes 24..32): the FlowEvent wire shape has no
    //     timestamp field today.
    //   - verdict (byte 32): always 0 from the BPF program (userspace
    //     will write it for analytics later).
    let _pkt_count = u64::from_ne_bytes([
        bytes[16], bytes[17], bytes[18], bytes[19], bytes[20], bytes[21], bytes[22], bytes[23],
    ]);

    // Map IANA proto byte to the protocol string the
    // `network_flow_decision` schema expects. Anything outside the BPF
    // program's filter (TCP/UDP only) becomes `None` defensively.
    let protocol = match proto {
        6 => Some("tcp".to_string()),
        17 => Some("udp".to_string()),
        1 => Some("icmp".to_string()),
        58 => Some("icmpv6".to_string()),
        _ => None,
    };

    let _src_addr = Ipv4Addr::from(src_ip); // not exposed on FlowEvent; kept
                                            // for future src-side surfacing
    let dst_addr = Ipv4Addr::from(dst_ip);

    Some(FlowEvent {
        direction: NetworkFlowDirection::Egress,
        // The BPF classifier is OBSERVABILITY ONLY (always returns
        // TC_ACT_OK) — every event it emits represents a packet the
        // kernel actually let through. Map that to `Allow`. Real
        // nftables drops surface via the separate nflog path with
        // `REASON_NFLOG_MATCH`.
        decision: NetworkFlowDecisionOutcome::Allow,
        reason_code: REASON_EBPF_XDP_DROP,
        dst_addr: Some(dst_addr.to_string()),
        dst_port: Some(dst_port),
        protocol,
        // byte_count: the BPF program counts packets, not bytes, so we
        // surface neither today. A future BPF-side change can add a
        // bytes_total field to FlowEntry and we'd thread it through
        // here.
        byte_count: None,
    })
}

// ────────────────────────────────────────────────────────────────
// Platform-independent ring-buffer event decoder
// ────────────────────────────────────────────────────────────────
//
// The Linux+aya path above ([`parse_ring_buf_event`]) is the
// production adapter the drainer calls per slot. It is cfg-gated so it
// only exists on Linux builds with the `ebpf-aya` feature on, and it
// returns `Option<FlowEvent>` shaped for the live channel.
//
// This second decoder is the unit-testable, cross-platform companion.
// It decodes the SAME 40-byte `#[repr(C)]` record from the BPF
// producer crate (`cellos-supervisor-ebpf/src/main.rs`) but:
//
//   - Compiles on every target (no cfg gate, no aya dependency).
//   - Returns a typed `RingBufFlowEvent` exposing every field the
//     kernel writes (including `pkt_count`, `first_seen_ns`, `verdict`,
//     `src_port` — fields the live adapter discards today).
//   - Surfaces malformed slots as a typed `RingBufParseError` rather
//     than a silent `None`, so future call sites that care about
//     layout-drift detection can act on it.
//   - Has a layout-pin test (`ring_buf_event_layout_constant_matches_producer`)
//     that fails loudly if the producer's 40-byte assumption shifts —
//     catching the drift on every CI host, not just Linux runners with
//     the aya feature enabled.
//
// The two decoders share the wire layout; only the call shape differs.
// A future refactor can collapse them once the live adapter wants
// access to the full FlowEntry surface (timestamps, verdict, pkt
// counts).
//
// Byte layout (native endian — the kernel program normalises wire
// bytes via `u32::from_be` / `u16::from_be` before insert, so the
// ring-buffer values are already host-order):
//
//   FlowKey   (16 bytes)
//     0..4   : src_ip   (u32, host order)
//     4..8   : dst_ip   (u32, host order)
//     8..10  : src_port (u16, host order)
//     10..12 : dst_port (u16, host order)
//     12     : proto    (u8; IANA: 6=TCP, 17=UDP)
//     13..16 : _pad     ([u8; 3], zeroed by the producer)
//   FlowEntry (24 bytes)
//     16..24 : pkt_count     (u64, host order)
//     24..32 : first_seen_ns (u64, host order)
//     32     : verdict       (u8; 0=unknown, 1=accept, 2=drop)
//     33..40 : _pad          ([u8; 7], zeroed)

/// Size in bytes of the on-the-ring-buffer `FlowEvent` record. Tied to
/// the `#[repr(C)]` layout in `cellos-supervisor-ebpf/src/main.rs`. If
/// the producer ever changes shape, this constant + the decoder must
/// move together — the unit tests below pin the exact byte offsets.
pub const RING_BUF_EVENT_LEN: usize = 40;

/// IANA protocol numbers we know how to label. Anything else is
/// surfaced as `None` so the downstream event keeps its `protocol`
/// field empty (rather than lying about a TCP/UDP that isn't either).
const IPPROTO_TCP: u8 = 6;
const IPPROTO_UDP: u8 = 17;
const IPPROTO_ICMP: u8 = 1;
const IPPROTO_ICMPV6: u8 = 58;

/// Decoded ring-buffer record. Mirrors the kernel-side `FlowEvent`
/// 1:1 but as plain Rust types: callers that don't care about the eBPF
/// shape get a portable, copy-only struct.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RingBufFlowEvent {
    pub src_ip: u32,
    pub dst_ip: u32,
    pub src_port: u16,
    pub dst_port: u16,
    pub proto: u8,
    pub pkt_count: u64,
    pub first_seen_ns: u64,
    pub verdict: u8,
}

/// Decode failure: the slice didn't have the canonical 40-byte shape.
/// The caller (drainer task) logs once and drops the malformed record —
/// surfacing a kernel-protocol mismatch is more useful than letting a
/// torn read silently inflate the flow count.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum RingBufParseError {
    #[error(
        "ring-buffer record is {got} bytes, expected exactly {RING_BUF_EVENT_LEN}; \
         kernel producer ABI drift or torn read"
    )]
    WrongLength { got: usize },
}

/// Decode a single 40-byte ring-buffer slot into a [`RingBufFlowEvent`].
///
/// The slot is treated as the contiguous `#[repr(C)]` byte image of
/// the eBPF `FlowEvent { key: FlowKey, entry: FlowEntry }`. All
/// multi-byte ints are host-endian (the eBPF program byte-swaps wire
/// values before inserting).
///
/// This is a pure function with no IO and no aya dependency, so it
/// builds and tests identically on Linux and macOS. The live-path
/// drainer uses the cfg-gated [`parse_ring_buf_event`] adapter (above)
/// for its hot path; this decoder is for unit tests and any future
/// caller that wants the full kernel field surface.
pub fn decode_ring_buf_event(bytes: &[u8]) -> Result<RingBufFlowEvent, RingBufParseError> {
    if bytes.len() != RING_BUF_EVENT_LEN {
        return Err(RingBufParseError::WrongLength { got: bytes.len() });
    }
    // Safe: slice length is exactly RING_BUF_EVENT_LEN, every fixed
    // offset below is bounded by it.
    let src_ip = u32::from_ne_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
    let dst_ip = u32::from_ne_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]);
    let src_port = u16::from_ne_bytes([bytes[8], bytes[9]]);
    let dst_port = u16::from_ne_bytes([bytes[10], bytes[11]]);
    let proto = bytes[12];
    // bytes[13..16] are explicit padding; we ignore by contract.
    let pkt_count = u64::from_ne_bytes([
        bytes[16], bytes[17], bytes[18], bytes[19], bytes[20], bytes[21], bytes[22], bytes[23],
    ]);
    let first_seen_ns = u64::from_ne_bytes([
        bytes[24], bytes[25], bytes[26], bytes[27], bytes[28], bytes[29], bytes[30], bytes[31],
    ]);
    let verdict = bytes[32];
    // bytes[33..40] are explicit padding.
    Ok(RingBufFlowEvent {
        src_ip,
        dst_ip,
        src_port,
        dst_port,
        proto,
        pkt_count,
        first_seen_ns,
        verdict,
    })
}

/// Translate a decoded [`RingBufFlowEvent`] into the supervisor-facing
/// [`FlowEvent`] shape consumed by the [`EbpfFlowMonitor`] channel.
///
/// IPv4 is rendered as dotted-quad. The kernel-side program filters out
/// non-IPv4 traffic, so we can format `src_ip`/`dst_ip` as `Ipv4Addr`
/// without losing information.
pub fn ring_buf_event_to_flow_event(decoded: &RingBufFlowEvent) -> FlowEvent {
    use std::net::Ipv4Addr;

    let protocol = match decoded.proto {
        IPPROTO_TCP => Some("tcp".to_string()),
        IPPROTO_UDP => Some("udp".to_string()),
        IPPROTO_ICMP => Some("icmp".to_string()),
        IPPROTO_ICMPV6 => Some("icmpv6".to_string()),
        _ => None,
    };
    // The kernel program is observability-only; verdict==2 ("drop") is
    // operator-written analytics, not enforcement. We map the three
    // documented values to `Allow` / `Deny`; the default (0 = unknown)
    // mirrors what the proxy + nft sees on the wire (allowed by
    // default, denials are nft-driven elsewhere).
    let decision = match decoded.verdict {
        2 => NetworkFlowDecisionOutcome::Deny,
        _ => NetworkFlowDecisionOutcome::Allow,
    };
    FlowEvent {
        direction: NetworkFlowDirection::Egress,
        decision,
        reason_code: REASON_EBPF_XDP_DROP,
        dst_addr: Some(Ipv4Addr::from(decoded.dst_ip).to_string()),
        dst_port: Some(decoded.dst_port),
        protocol,
        byte_count: None, // ebpf program tracks pkt_count, not bytes
    }
}

// ────────────────────────────────────────────────────────────────
// Test-only shim: a constructible EbpfFlowMonitor for unit tests
// that need to exercise the `drain` / `stop` surface without
// depending on a real BPF program. NOT for production use.
// ────────────────────────────────────────────────────────────

#[cfg(test)]
impl EbpfFlowMonitor {
    /// Construct a monitor from a pre-built receiver. Test-only.
    pub(crate) fn from_parts_for_test(
        rx: tokio::sync::mpsc::UnboundedReceiver<FlowEvent>,
        iface: &str,
    ) -> Self {
        Self {
            #[cfg(all(target_os = "linux", feature = "ebpf-aya"))]
            _bpf: None,
            rx,
            drainer: None,
            iface: iface.to_string(),
        }
    }
}

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

    fn sample_event() -> FlowEvent {
        FlowEvent {
            direction: NetworkFlowDirection::Egress,
            decision: NetworkFlowDecisionOutcome::Deny,
            reason_code: REASON_EBPF_XDP_DROP,
            dst_addr: Some("10.0.0.1".into()),
            dst_port: Some(443),
            protocol: Some("tcp".into()),
            byte_count: Some(74),
        }
    }

    // ── original scaffold tests (preserved) ───────────────────────────

    #[test]
    fn env_flag_default_off() {
        let prev = std::env::var(PER_FLOW_EBPF_ENV).ok();
        std::env::remove_var(PER_FLOW_EBPF_ENV);
        assert!(!is_per_flow_ebpf_enabled());
        if let Some(v) = prev {
            std::env::set_var(PER_FLOW_EBPF_ENV, v);
        }
    }

    #[test]
    fn env_flag_recognises_one() {
        let prev = std::env::var(PER_FLOW_EBPF_ENV).ok();
        std::env::set_var(PER_FLOW_EBPF_ENV, "1");
        assert!(is_per_flow_ebpf_enabled());
        match prev {
            Some(v) => std::env::set_var(PER_FLOW_EBPF_ENV, v),
            None => std::env::remove_var(PER_FLOW_EBPF_ENV),
        }
    }

    #[test]
    fn env_flag_rejects_truthy_lookalikes() {
        let prev = std::env::var(PER_FLOW_EBPF_ENV).ok();
        for bad in ["true", "yes", "on", "TRUE", "0", "", "2"] {
            std::env::set_var(PER_FLOW_EBPF_ENV, bad);
            assert!(
                !is_per_flow_ebpf_enabled(),
                "value {bad:?} must not enable Phase 2"
            );
        }
        match prev {
            Some(v) => std::env::set_var(PER_FLOW_EBPF_ENV, v),
            None => std::env::remove_var(PER_FLOW_EBPF_ENV),
        }
    }

    #[test]
    fn noop_listener_yields_no_events() {
        let mut listener = NoopFlowListener;
        let events = listener.drain_events();
        assert!(events.is_empty());
    }

    #[test]
    fn noop_listener_drain_is_idempotent() {
        let mut listener = NoopFlowListener;
        for _ in 0..5 {
            assert!(listener.drain_events().is_empty());
        }
    }

    #[test]
    fn noop_listener_backend_name_is_noop() {
        let listener = NoopFlowListener;
        assert_eq!(listener.backend_name(), "noop");
    }

    #[test]
    fn build_default_listener_is_noop() {
        let mut listener = build_default_listener();
        assert_eq!(listener.backend_name(), "noop");
        assert!(listener.drain_events().is_empty());
    }

    #[test]
    fn translator_preserves_direction_and_outcome() {
        let event = sample_event();
        let payload = flow_event_to_network_flow_decision(
            &event,
            "cell-test",
            "run-test",
            "dec-test",
            None,
            None,
            None,
            None,
            "2026-05-06T00:00:00Z",
        );
        assert_eq!(payload.direction, NetworkFlowDirection::Egress);
        assert_eq!(payload.decision, NetworkFlowDecisionOutcome::Deny);
        assert_eq!(payload.reason_code, REASON_EBPF_XDP_DROP);
    }

    #[test]
    fn translator_stamps_caller_metadata() {
        let event = sample_event();
        let payload = flow_event_to_network_flow_decision(
            &event,
            "cell-1",
            "run-1",
            "dec-1",
            Some("sha256:abc"),
            Some("kid-1"),
            Some("issuer-1"),
            Some("corr-1"),
            "2026-05-06T12:00:00Z",
        );
        assert_eq!(payload.cell_id, "cell-1");
        assert_eq!(payload.run_id, "run-1");
        assert_eq!(payload.decision_id, "dec-1");
        assert_eq!(payload.policy_digest.as_deref(), Some("sha256:abc"));
        assert_eq!(payload.keyset_id.as_deref(), Some("kid-1"));
        assert_eq!(payload.issuer_kid.as_deref(), Some("issuer-1"));
        assert_eq!(payload.correlation_id.as_deref(), Some("corr-1"));
        assert_eq!(payload.observed_at, "2026-05-06T12:00:00Z");
    }

    #[test]
    fn translator_omits_metadata_when_caller_passes_none() {
        let event = sample_event();
        let payload = flow_event_to_network_flow_decision(
            &event,
            "cell-1",
            "run-1",
            "dec-1",
            None,
            None,
            None,
            None,
            "2026-05-06T12:00:00Z",
        );
        assert!(payload.policy_digest.is_none());
        assert!(payload.keyset_id.is_none());
        assert!(payload.issuer_kid.is_none());
        assert!(payload.correlation_id.is_none());
        assert!(payload.nft_rule_ref.is_none());
    }

    #[test]
    fn translator_carries_l3_l4_fields() {
        let event = sample_event();
        let payload = flow_event_to_network_flow_decision(
            &event,
            "cell-1",
            "run-1",
            "dec-1",
            None,
            None,
            None,
            None,
            "2026-05-06T12:00:00Z",
        );
        assert_eq!(payload.dst_addr.as_deref(), Some("10.0.0.1"));
        assert_eq!(payload.dst_port, Some(443));
        assert_eq!(payload.protocol.as_deref(), Some("tcp"));
        assert_eq!(payload.byte_count, Some(74));
        assert_eq!(payload.packet_count, Some(1));
    }

    #[test]
    fn translator_handles_event_without_byte_count() {
        let mut event = sample_event();
        event.byte_count = None;
        let payload = flow_event_to_network_flow_decision(
            &event,
            "cell-1",
            "run-1",
            "dec-1",
            None,
            None,
            None,
            None,
            "2026-05-06T12:00:00Z",
        );
        assert!(payload.byte_count.is_none());
        assert!(payload.packet_count.is_none());
    }

    #[test]
    fn translator_handles_nflog_reason_code() {
        let mut event = sample_event();
        event.reason_code = REASON_NFLOG_MATCH;
        event.decision = NetworkFlowDecisionOutcome::Allow;
        let payload = flow_event_to_network_flow_decision(
            &event,
            "cell-1",
            "run-1",
            "dec-1",
            None,
            None,
            None,
            None,
            "2026-05-06T12:00:00Z",
        );
        assert_eq!(payload.reason_code, REASON_NFLOG_MATCH);
        assert_eq!(payload.decision, NetworkFlowDecisionOutcome::Allow);
    }

    #[test]
    fn translator_payload_round_trips_through_serde_json() {
        let event = sample_event();
        let payload = flow_event_to_network_flow_decision(
            &event,
            "cell-1",
            "run-1",
            "dec-1",
            Some("sha256:abc"),
            Some("kid-1"),
            Some("issuer-1"),
            Some("corr-1"),
            "2026-05-06T12:00:00Z",
        );
        let json = serde_json::to_string(&payload).expect("serialize");
        let round_tripped: NetworkFlowDecision = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(round_tripped, payload);
    }

    #[test]
    fn reason_codes_are_distinct_from_phase1_codes() {
        for code in [REASON_EBPF_XDP_DROP, REASON_NFLOG_MATCH] {
            assert!(
                !code.starts_with("nft_"),
                "Phase 2 code {code} must not collide with Phase 1 nft_* prefix"
            );
        }
    }

    // ── E7-4 EbpfFlowMonitor tests ───────────────────────────────────

    #[cfg(not(target_os = "linux"))]
    #[tokio::test]
    async fn start_is_unsupported_on_non_linux() {
        // Pass a sentinel fd; the macOS code path never touches it.
        let res = EbpfFlowMonitor::start(-1, "tap0").await;
        match res {
            Err(EbpfMonitorError::Unsupported { host }) => {
                assert!(
                    !host.is_empty(),
                    "Unsupported error must name the host platform"
                );
            }
            Err(other) => panic!("expected Unsupported, got {other:?}"),
            Ok(_) => panic!("monitor must not construct on non-Linux"),
        }
    }

    #[cfg(all(target_os = "linux", not(feature = "ebpf-aya")))]
    #[tokio::test]
    async fn start_returns_feature_disabled_when_feature_off() {
        let res = EbpfFlowMonitor::start(-1, "tap0").await;
        assert!(
            matches!(res, Err(EbpfMonitorError::FeatureDisabled)),
            "must report FeatureDisabled when ebpf-aya cargo feature is off"
        );
    }

    #[tokio::test]
    async fn drain_yields_empty_when_no_events_buffered() {
        let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<FlowEvent>();
        let mut monitor = EbpfFlowMonitor::from_parts_for_test(rx, "tap0");
        let batch = monitor.drain();
        assert!(batch.is_empty(), "fresh monitor has no buffered events");
    }

    #[tokio::test]
    async fn drain_yields_events_pushed_through_channel() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<FlowEvent>();
        let mut monitor = EbpfFlowMonitor::from_parts_for_test(rx, "tap0");
        tx.send(sample_event()).expect("send ok");
        tx.send(sample_event()).expect("send ok");
        let batch = monitor.drain();
        assert_eq!(batch.len(), 2, "drain must pull both events synchronously");
    }

    #[tokio::test]
    async fn drain_is_non_blocking() {
        // No sender pushes; drain must return immediately, not park.
        let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<FlowEvent>();
        let mut monitor = EbpfFlowMonitor::from_parts_for_test(rx, "tap0");
        let start = std::time::Instant::now();
        let batch = monitor.drain();
        let elapsed = start.elapsed();
        assert!(batch.is_empty());
        assert!(
            elapsed < std::time::Duration::from_millis(50),
            "drain must be non-blocking; took {elapsed:?}"
        );
    }

    #[tokio::test]
    async fn stop_is_idempotent_when_no_drainer_present() {
        let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<FlowEvent>();
        let monitor = EbpfFlowMonitor::from_parts_for_test(rx, "tap0");
        // Should complete without panicking even though `drainer` is None.
        monitor.stop().await;
    }

    // ── decode_ring_buf_event tests ────────────────────────────────────
    //
    // Pin the exact byte offsets the kernel-side
    // `cellos-supervisor-ebpf/src/main.rs` producer emits. If the
    // producer's `#[repr(C)]` layout ever shifts, these tests fail in
    // place rather than letting a torn record silently inflate the
    // supervisor's flow count.
    //
    // These tests target the platform-independent
    // [`decode_ring_buf_event`] decoder, which exists on every host. The
    // cfg-gated live-path adapter [`parse_ring_buf_event`] is exercised
    // implicitly: both decoders share the layout pinned by
    // `ring_buf_event_layout_constant_matches_producer`.

    fn canonical_record() -> [u8; RING_BUF_EVENT_LEN] {
        let mut buf = [0u8; RING_BUF_EVENT_LEN];
        // src_ip = 10.0.0.1 (host order u32 = 0x0A000001 LE-bytes; we
        // round-trip through `to_ne_bytes` so the test passes on any
        // host endianness).
        let src_ip: u32 = u32::from(std::net::Ipv4Addr::new(10, 0, 0, 1));
        let dst_ip: u32 = u32::from(std::net::Ipv4Addr::new(192, 0, 2, 5));
        buf[0..4].copy_from_slice(&src_ip.to_ne_bytes());
        buf[4..8].copy_from_slice(&dst_ip.to_ne_bytes());
        buf[8..10].copy_from_slice(&44_321u16.to_ne_bytes()); // src_port
        buf[10..12].copy_from_slice(&443u16.to_ne_bytes()); // dst_port
        buf[12] = 6; // proto = TCP
                     // bytes[13..16] = padding, left zero
        buf[16..24].copy_from_slice(&7_777u64.to_ne_bytes()); // pkt_count
        buf[24..32].copy_from_slice(&123_456_789u64.to_ne_bytes()); // first_seen_ns
        buf[32] = 1; // verdict = accept
                     // bytes[33..40] = padding, left zero
        buf
    }

    #[test]
    fn decode_ring_buf_event_round_trips_canonical_record() {
        let buf = canonical_record();
        let decoded = decode_ring_buf_event(&buf).expect("canonical 40-byte record decodes");
        assert_eq!(
            decoded.src_ip,
            u32::from(std::net::Ipv4Addr::new(10, 0, 0, 1))
        );
        assert_eq!(
            decoded.dst_ip,
            u32::from(std::net::Ipv4Addr::new(192, 0, 2, 5))
        );
        assert_eq!(decoded.src_port, 44_321);
        assert_eq!(decoded.dst_port, 443);
        assert_eq!(decoded.proto, 6);
        assert_eq!(decoded.pkt_count, 7_777);
        assert_eq!(decoded.first_seen_ns, 123_456_789);
        assert_eq!(decoded.verdict, 1);
    }

    #[test]
    fn decode_ring_buf_event_rejects_short_record() {
        let buf = [0u8; RING_BUF_EVENT_LEN - 1];
        match decode_ring_buf_event(&buf) {
            Err(RingBufParseError::WrongLength { got }) => {
                assert_eq!(got, RING_BUF_EVENT_LEN - 1);
            }
            other => panic!("expected WrongLength, got {other:?}"),
        }
    }

    #[test]
    fn decode_ring_buf_event_rejects_long_record() {
        let buf = [0u8; RING_BUF_EVENT_LEN + 8];
        assert!(matches!(
            decode_ring_buf_event(&buf),
            Err(RingBufParseError::WrongLength { got }) if got == RING_BUF_EVENT_LEN + 8
        ));
    }

    #[test]
    fn decode_ring_buf_event_ignores_padding_bytes() {
        // Padding bytes in [13..16) and [33..40) are explicitly defined
        // as "zero by producer" but the decoder must not error on
        // non-zero padding — the eBPF verifier zeros them, but a future
        // producer that doesn't might still surface useful records.
        let mut buf = canonical_record();
        buf[13..16].copy_from_slice(&[0xff, 0xff, 0xff]);
        buf[33..40].copy_from_slice(&[0xaa; 7]);
        let decoded = decode_ring_buf_event(&buf).expect("non-zero padding must decode");
        // Non-padding fields unaffected.
        assert_eq!(decoded.proto, 6);
        assert_eq!(decoded.verdict, 1);
    }

    #[test]
    fn ring_buf_event_to_flow_event_renders_tcp_dst() {
        let decoded = decode_ring_buf_event(&canonical_record()).unwrap();
        let ev = ring_buf_event_to_flow_event(&decoded);
        assert_eq!(ev.direction, NetworkFlowDirection::Egress);
        assert_eq!(ev.decision, NetworkFlowDecisionOutcome::Allow);
        assert_eq!(ev.protocol.as_deref(), Some("tcp"));
        assert_eq!(ev.dst_addr.as_deref(), Some("192.0.2.5"));
        assert_eq!(ev.dst_port, Some(443));
    }

    #[test]
    fn ring_buf_event_to_flow_event_maps_verdict_drop_to_deny() {
        let mut buf = canonical_record();
        buf[32] = 2; // verdict = drop
        let decoded = decode_ring_buf_event(&buf).unwrap();
        let ev = ring_buf_event_to_flow_event(&decoded);
        assert_eq!(ev.decision, NetworkFlowDecisionOutcome::Deny);
    }

    #[test]
    fn ring_buf_event_to_flow_event_maps_unknown_proto_to_none() {
        let mut buf = canonical_record();
        buf[12] = 99; // unknown IANA proto
        let decoded = decode_ring_buf_event(&buf).unwrap();
        let ev = ring_buf_event_to_flow_event(&decoded);
        assert!(ev.protocol.is_none(), "unknown proto must not lie");
    }

    #[test]
    fn ring_buf_event_to_flow_event_maps_udp() {
        let mut buf = canonical_record();
        buf[12] = 17; // UDP
        let decoded = decode_ring_buf_event(&buf).unwrap();
        let ev = ring_buf_event_to_flow_event(&decoded);
        assert_eq!(ev.protocol.as_deref(), Some("udp"));
    }

    #[test]
    fn ring_buf_event_layout_constant_matches_producer() {
        // Defensive: if anyone widens the kernel-side FlowEvent struct
        // without updating either host-side parser, this assertion makes
        // the mismatch loud at compile-test time rather than at runtime
        // on a Linux deployment.
        //
        // Producer (cellos-supervisor-ebpf): 16 (FlowKey) + 24 (FlowEntry) = 40.
        assert_eq!(RING_BUF_EVENT_LEN, 40);
        // Linux+aya path uses a parallel `BPF_FLOW_EVENT_SIZE` constant.
        // The two MUST agree — the only difference is which platforms
        // they compile on, not what they describe.
        #[cfg(all(target_os = "linux", feature = "ebpf-aya"))]
        assert_eq!(RING_BUF_EVENT_LEN, BPF_FLOW_EVENT_SIZE);
    }

    #[test]
    fn ebpf_monitor_error_display_strings_are_useful() {
        // Defensive: operators see these strings in their logs when a
        // cell falls back to nflog. They must name the failure mode
        // unambiguously so paging on "eBPF flow monitor error" is
        // actionable rather than mysterious.
        let unsupported = EbpfMonitorError::Unsupported { host: "macos" };
        assert!(format!("{unsupported}").contains("Linux-only"));

        let disabled = EbpfMonitorError::FeatureDisabled;
        assert!(format!("{disabled}").contains("ebpf-aya"));

        let missing = EbpfMonitorError::ObjectMissing {
            path: "/tmp/x.o".to_string(),
        };
        assert!(format!("{missing}").contains("/tmp/x.o"));

        let attach = EbpfMonitorError::AttachFailed {
            iface: "tap0".to_string(),
            message: "ENODEV".to_string(),
        };
        let s = format!("{attach}");
        assert!(s.contains("tap0"));
        assert!(s.contains("ENODEV"));
    }
}

// ────────────────────────────────────────────────────────────────────────
// L5-15 — Connection-tracking flow accumulator
// ────────────────────────────────────────────────────────────────────────
//
// The types above (`FlowEvent`, `FlowEventListener`, `NoopFlowListener`,
// `flow_event_to_network_flow_decision`) are the Phase 2 nflog-shaped
// scaffolding — one event = one observed packet/decision, designed to ride
// the existing `network_flow_decision` v1 wire schema.
//
// L5-15 needs a *different* shape: open/close signals per 5-tuple so the
// supervisor can count UNIQUE connections that were actually exercised by
// the workload, and stamp that count into the `HomeostasisSignal`'s
// `exercised_egress_connections` field. This sub-module ships the new
// types under their own namespace to avoid colliding with the Phase 2
// scaffold's `FlowEvent`, which has additive `reason_code` semantics this
// counter doesn't need.
//
// The accumulator is intentionally minimal: a `HashSet<FlowKey>` of opened
// flows. When `CELLOS_PER_FLOW_REALTIME=1` is on and a real backend (eBPF
// cgroup_skb or nflog connection tracker) feeds it Opened events, the
// homeostasis emit site reads `unique_flow_count()` and stamps it into the
// signal. When the env var is off OR no backend is wired, the supervisor
// falls back to `None` + `telemetry_pending_L2-04` reason, preserving the
// E2-06 contract.
pub mod connection_tracking {
    use std::collections::HashSet;
    use std::net::IpAddr;

    /// 5-tuple uniquely identifying an L4 flow. Used as the hash key for
    /// counting unique connections exercised by the workload.
    ///
    /// `protocol` is a small numeric (IANA proto byte: 6 = TCP, 17 = UDP,
    /// 1 = ICMP, 58 = ICMPv6) rather than a `String` so the hash key stays
    /// cheap to compute and copy. Source addr/port are included so a
    /// workload that opens N parallel connections to the same `(dst, port)`
    /// is counted as N flows, matching what `exercised_egress_connections`
    /// promises to consumers ("connections", not "destinations").
    #[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
    pub struct FlowKey {
        pub src_addr: IpAddr,
        pub src_port: u16,
        pub dst_addr: IpAddr,
        pub dst_port: u16,
        pub protocol: u8,
    }

    /// Open/close phase of a connection-tracked flow.
    ///
    /// `Closed` carries the observed packet + byte counts so future taudit
    /// passes can correlate flow lifespan with declared egress rules.
    /// The accumulator below only consumes `Opened` today; `Closed` is
    /// kept on the type so a backend implementation can emit both phases
    /// without us churning the enum later.
    #[derive(Debug, Clone)]
    pub enum FlowEventKind {
        Opened,
        Closed { pkt_count: u64, byte_count: u64 },
    }

    /// One observation a connection-tracking backend (eBPF cgroup_skb or
    /// nflog with conntrack) surfaces to the supervisor.
    ///
    /// Distinct from [`super::FlowEvent`] (the Phase 2 nflog scaffold's
    /// packet-decision shape). The two are deliberately namespaced so the
    /// supervisor can wire one or both without ambiguity.
    #[derive(Debug, Clone)]
    pub struct FlowEvent {
        pub key: FlowKey,
        pub kind: FlowEventKind,
        /// Monotonic-clock nanoseconds when the event was observed.
        /// Kept as `u64` to match `clock_gettime(CLOCK_MONOTONIC)` /
        /// `bpf_ktime_get_ns()` semantics — kernel-side clock, not wall.
        pub timestamp_ns: u64,
    }

    /// Accumulates unique opened flows for the lifetime of one cell run.
    ///
    /// `record()` is idempotent on the `FlowKey` — the same 5-tuple
    /// observed twice counts as one flow, matching what the homeostasis
    /// `exercised_egress_connections` field promises consumers (a count of
    /// distinct connections actually exercised, not packet count).
    ///
    /// Constructed once per `Supervisor::run` when
    /// `CELLOS_PER_FLOW_REALTIME=1` is on; consumed at the homeostasis
    /// emit site after `child.wait()` returns.
    #[derive(Debug, Default)]
    pub struct FlowAccumulator {
        opened: HashSet<FlowKey>,
    }

    impl FlowAccumulator {
        pub fn new() -> Self {
            Self::default()
        }

        /// Record a flow event. Only `Opened` events affect the unique
        /// count today; `Closed` is silently consumed so a future
        /// implementation can stamp byte/packet aggregates without a
        /// breaking API change.
        pub fn record(&mut self, event: &FlowEvent) {
            if matches!(event.kind, FlowEventKind::Opened) {
                self.opened.insert(event.key);
            }
        }

        /// Number of unique 5-tuples observed in the `Opened` phase so far.
        ///
        /// Returns `u64` because real workloads can plausibly open more
        /// than `u32::MAX / 2` flows over a long-lived run; the supervisor
        /// saturates to `u32::MAX` at the homeostasis emit site since the
        /// wire schema's field is `u32`. Saturation, not truncation, is
        /// the safe choice: a count claiming "exactly u32::MAX" is closer
        /// to truth than wrapping to 0.
        pub fn unique_flow_count(&self) -> u64 {
            self.opened.len() as u64
        }
    }

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

        fn key(src_port: u16, dst_octet: u8, dst_port: u16) -> FlowKey {
            FlowKey {
                src_addr: IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)),
                src_port,
                dst_addr: IpAddr::V4(Ipv4Addr::new(192, 0, 2, dst_octet)),
                dst_port,
                protocol: 6, // TCP
            }
        }

        fn opened_event(k: FlowKey) -> FlowEvent {
            FlowEvent {
                key: k,
                kind: FlowEventKind::Opened,
                timestamp_ns: 0,
            }
        }

        #[test]
        fn empty_accumulator_reports_zero() {
            let acc = FlowAccumulator::new();
            assert_eq!(acc.unique_flow_count(), 0);
        }

        #[test]
        fn opened_event_increments_count() {
            let mut acc = FlowAccumulator::new();
            acc.record(&opened_event(key(40000, 1, 443)));
            assert_eq!(acc.unique_flow_count(), 1);
        }

        #[test]
        fn duplicate_opened_is_idempotent() {
            // Same 5-tuple observed twice MUST count as one flow — the
            // homeostasis semantics are "distinct connections," not "open
            // events." A retransmitted SYN or a backend re-emitting an
            // event MUST NOT inflate the count.
            let mut acc = FlowAccumulator::new();
            let k = key(40000, 1, 443);
            acc.record(&opened_event(k));
            acc.record(&opened_event(k));
            assert_eq!(acc.unique_flow_count(), 1);
        }

        #[test]
        fn distinct_flows_count_independently() {
            let mut acc = FlowAccumulator::new();
            acc.record(&opened_event(key(40000, 1, 443)));
            acc.record(&opened_event(key(40001, 1, 443))); // different src_port
            acc.record(&opened_event(key(40000, 2, 443))); // different dst_addr
            acc.record(&opened_event(key(40000, 1, 80))); // different dst_port
            assert_eq!(acc.unique_flow_count(), 4);
        }

        #[test]
        fn closed_event_does_not_increment_count() {
            // `Closed` is observed for symmetry/future use but MUST NOT
            // affect the unique-flow count — the count is keyed off the
            // `Opened` phase.
            let mut acc = FlowAccumulator::new();
            let k = key(40000, 1, 443);
            acc.record(&FlowEvent {
                key: k,
                kind: FlowEventKind::Closed {
                    pkt_count: 10,
                    byte_count: 1500,
                },
                timestamp_ns: 1,
            });
            assert_eq!(acc.unique_flow_count(), 0);
        }

        #[test]
        fn closed_after_opened_preserves_count() {
            let mut acc = FlowAccumulator::new();
            let k = key(40000, 1, 443);
            acc.record(&opened_event(k));
            acc.record(&FlowEvent {
                key: k,
                kind: FlowEventKind::Closed {
                    pkt_count: 5,
                    byte_count: 500,
                },
                timestamp_ns: 2,
            });
            assert_eq!(acc.unique_flow_count(), 1);
        }
    }
}