feat(retire-falco): F5 — per-node agent-liveness contract + honest corroboration-coverage UX (JEF-308)

agent/protector-agent/src/coalesce/tests.rs

@@ -9,6 +9,7 @@ fn obs(pod_uid: &str, behavior: Behavior, at_ms: u64) -> RuntimeObservation {
         attribution: Attribution::by_pod_uid(pod_uid),
         source: Some("protector-agent".into()),
         observed_at_ms: Some(at_ms),
+        node: None,
         behavior,
     }
 }

agent/protector-agent/src/main.rs

@@ -13,14 +13,62 @@ mod pod;
 mod reporter;
 
 use std::io::IsTerminal;
-use std::time::Duration;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
-use protector_behavior::RuntimeObservation;
+use protector_behavior::{AgentReport, RuntimeObservation};
 use tokio::sync::mpsc;
 
 use coalesce::Coalescer;
 use reporter::Reporter;
 
+/// Shared **probe-attach status** (JEF-308): the observer sets it once its eBPF probes attach, and
+/// the per-node liveness beacon reads it each window. The default no-eBPF build never sets it (stays
+/// `0/0`), so the agent honestly reports itself BLIND — signal-flow liveness, not pod-Ready.
+#[derive(Default)]
+pub struct ProbeStatus {
+    loaded: AtomicU32,
+    total: AtomicU32,
+}
+
+impl ProbeStatus {
+    /// Record how many probes attached out of how many were tried (called by the observer).
+    pub fn set(&self, loaded: u32, total: u32) {
+        self.loaded.store(loaded, Ordering::Relaxed);
+        self.total.store(total, Ordering::Relaxed);
+    }
+
+    /// `(loaded, total)` as of now — read by the liveness beacon.
+    pub fn snapshot(&self) -> (u32, u32) {
+        (
+            self.loaded.load(Ordering::Relaxed),
+            self.total.load(Ordering::Relaxed),
+        )
+    }
+}
+
+/// Wall-clock now as Unix epoch millis, for the beacon's freshness stamp. `None` only if the clock
+/// is before the epoch — the engine then stamps ingest time.
+fn now_ms() -> Option<u64> {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .ok()
+        .map(|d| d.as_millis() as u64)
+}
+
+/// Build a per-node liveness beacon (JEF-308) from the node, the probe-attach status, and the
+/// signals emitted this window. Pure over its inputs so it's unit-testable without the runtime.
+fn build_agent_report(node: &str, probes: (u32, u32), signals: u64) -> AgentReport {
+    AgentReport {
+        node: node.to_string(),
+        probes_loaded: probes.0,
+        probes_total: probes.1,
+        signals_emitted: signals,
+        observed_at_ms: now_ms(),
+    }
+}
+
 /// Max distinct coalesced keys the debounce buffer holds before a forced flush (JEF-296).
 /// Bounds memory and keeps a flushed batch well under the engine's 1024 per-batch cap, so
 /// the "behavior batch exceeds the per-batch cap" WARN stays quiet under normal load.
@@ -63,12 +111,26 @@ async fn main() -> anyhow::Result<()> {
     let endpoint = std::env::var("PROTECTOR_AGENT_ENDPOINT")
         .unwrap_or_else(|_| "http://protector.protector.svc.cluster.local:9999".to_string());
     let debounce_window = parse_debounce_window(std::env::var("PROTECTOR_AGENT_DEBOUNCE_MS").ok());
+    // The node this agent runs on (JEF-308), from the downward API (`K8S_NODE = spec.nodeName`).
+    // Stamped onto every observation and onto the per-node liveness beacon. When unset (a dev run
+    // outside k8s) we can't attribute per node — observations go out node-less and no beacon is
+    // sent (an un-attributable beacon would be dishonest).
+    let node = std::env::var("K8S_NODE").ok().filter(|n| !n.is_empty());
+    if node.is_none() {
+        tracing::warn!(
+            "K8S_NODE is unset — observations are not node-attributed and no per-node liveness \
+             beacon will be sent (set it via the downward API `spec.nodeName`)"
+        );
+    }
     tracing::info!(
         %endpoint,
+        node = node.as_deref().unwrap_or("<unset>"),
         debounce_ms = debounce_window.as_millis(),
         "protector-agent starting"
     );
     let mut reporter = Reporter::new(&endpoint)?;
+    // Probe-attach status the observer updates and the liveness beacon reads (JEF-308).
+    let probes = Arc::new(ProbeStatus::default());
 
     let (tx, mut rx) = mpsc::channel::<RuntimeObservation>(4096);
 
@@ -79,6 +141,8 @@ async fn main() -> anyhow::Result<()> {
     // corroboration must stay low-latency). Best-effort sends — a lost batch costs freshness,
     // never correctness (behavioral evidence is additive). A running count is logged at info
     // once per HEARTBEAT_INTERVAL so an operator can confirm the agent is actually reporting.
+    let beacon_node = node.clone();
+    let beacon_probes = probes.clone();
     let flusher = tokio::spawn(async move {
         let mut coalescer = Coalescer::new(MAX_BATCH);
         let mut reported_since_tick: usize = 0;
@@ -91,7 +155,9 @@ async fn main() -> anyhow::Result<()> {
         loop {
             tokio::select! {
                 recv = rx.recv() => match recv {
-                    Some(obs) => {
+                    Some(mut obs) => {
+                        // Stamp this agent's node (JEF-308) so the observation is node-attributed.
+                        obs.node = beacon_node.clone();
                         // `offer` returns anything to POST NOW: an alert (never debounced),
                         // or the drained buffer if this new distinct key hit the max-size cap.
                         let immediate = coalescer.offer(obs);
@@ -124,20 +190,35 @@ async fn main() -> anyhow::Result<()> {
                         "behavioral observations reported (last {}s)",
                         HEARTBEAT_INTERVAL.as_secs(),
                     );
+                    // Per-node liveness beacon (JEF-308): sent EVERY window even when quiet
+                    // (reported_since_tick == 0) — a quiet node with probes loaded reads
+                    // healthy-quiet, not blind. Skipped only when the node is unknown (an
+                    // un-attributable beacon would be dishonest). probes==0/0 ⇒ blind (Ready
+                    // but no probe attached, or the no-eBPF build).
+                    if let Some(node) = &beacon_node {
+                        let report = build_agent_report(
+                            node,
+                            beacon_probes.snapshot(),
+                            reported_since_tick as u64,
+                        );
+                        reporter.send_report(&report).await;
+                    }
                     reported_since_tick = 0;
                 }
             }
         }
     });
 
-    // Collection. Default build is a no-op; `--features ebpf` loads the real probes.
+    // Collection. Default build is a no-op; `--features ebpf` loads the real probes. The observer
+    // updates `probes` with how many eBPF probes attached (JEF-308) — the no-op build leaves it at
+    // 0/0, honestly reporting itself blind.
     #[cfg(not(feature = "ebpf"))]
-    observer::NoopObserver.run(tx).await;
+    observer::NoopObserver.run(tx, probes).await;
     #[cfg(feature = "ebpf")]
     {
         // Events are attributed by pod UID (from the cgroup); the engine resolves UID →
         // namespace/pod via its watch, so the agent needs no cluster credentials.
-        if let Err(error) = observer::EbpfObserver.run(tx).await {
+        if let Err(error) = observer::EbpfObserver.run(tx, probes).await {
             // Degrade, don't crashloop (ADR-0014): a missing hook / failed attach should
             // leave the pod up for inspection, not hammer restarts.
             tracing::error!(%error, "ebpf observer exited; idling (no collection)");
@@ -166,6 +247,32 @@ mod tests {
         );
     }
 
+    #[test]
+    fn agent_report_carries_node_probes_and_window_signals() {
+        // JEF-308: a healthy window — probes loaded, some signals.
+        let r = build_agent_report("node-a", (6, 6), 12);
+        assert_eq!(r.node, "node-a");
+        assert_eq!(r.probes_loaded, 6);
+        assert_eq!(r.probes_total, 6);
+        assert_eq!(r.signals_emitted, 12);
+        assert!(!r.is_blind());
+        assert!(!r.is_partial());
+    }
+
+    #[test]
+    fn agent_report_is_blind_when_no_probe_attached() {
+        // The no-eBPF build (or a Ready-but-blind agent) reports 0 probes → blind, even quiet.
+        let r = build_agent_report("node-a", (0, 0), 0);
+        assert!(r.is_blind());
+        // A quiet window with probes loaded is NOT blind (quiet≠blind).
+        let quiet = build_agent_report("node-a", (6, 6), 0);
+        assert!(!quiet.is_blind());
+        // A partial load is degraded, not blind.
+        let partial = build_agent_report("node-a", (4, 6), 1);
+        assert!(partial.is_partial());
+        assert!(!partial.is_blind());
+    }
+
     #[test]
     fn debounce_window_falls_back_to_the_default() {
         // Unset, unparseable, and zero all fall back — a zero period would panic the