From 923fd11bbab4d17963d25f69d346587cf3d4bfdb Mon Sep 17 00:00:00 2001 From: Jeff Larson Date: Mon, 6 Jul 2026 01:35:34 -0700 Subject: [PATCH] fix(runtime-ingest): unify /behavior + /agent-liveness into one report envelope (JEF-336) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The agent posted the JEF-308 liveness beacon as a single AgentReport to /agent-liveness, but that handler expected Json> — so every beacon 422'd and no liveness was ever recorded, which is why the dashboard showed "no agents connected". Observations (/behavior) were unaffected. Collapse the two endpoints into ONE per-window RuntimeReport envelope { observations, liveness: Option } in the shared behavior crate, so there is one schema and one auth path and liveness always travels with the report (a quiet node still reports → HEALTHY-quiet, not blind). - behavior: add RuntimeReport (liveness is Option so a third-party sensor may send observations only, per ADR-0003); round-trip + omission tests. - engine: the single /behavior handler ingests observations (semantics unchanged, incl. Behavior::Alert → corroboration) AND records envelope liveness; /agent-liveness route removed. Backward-compat route (a): a serde untagged BehaviorIngest accepts EITHER the envelope OR a bare [...] array (legacy/third-party), so nothing breaks. Bearer auth + body cap + rate limit are unchanged Router layers. - agent: fold liveness into the /behavior envelope; drop the separate beacon POST + liveness_url. Quiet-window heartbeat POSTs empty observations + liveness. Compat route (a) chosen over envelope-only: the per-PR e2e itself POSTs a bare array to /behavior, and it stays truest to the ADR-0003 tool-agnostic port. Closes JEF-336 Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01VtjoJttCvBY4dzCoE4f9vP --- agent/protector-agent/src/main.rs | 36 ++- agent/protector-agent/src/reporter.rs | 138 +++++------ behavior/src/lib.rs | 98 ++++++++ .../protector/templates/agent-daemonset.yaml | 3 +- engine/src/engine/observe/mod.rs | 2 +- engine/src/engine/observe/runtime.rs | 226 +++++++++++++++--- engine/src/engine/run_loop.rs | 5 +- 7 files changed, 390 insertions(+), 118 deletions(-) diff --git a/agent/protector-agent/src/main.rs b/agent/protector-agent/src/main.rs index 611761e..df07d87 100644 --- a/agent/protector-agent/src/main.rs +++ b/agent/protector-agent/src/main.rs @@ -17,7 +17,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicU32, Ordering}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use protector_behavior::{AgentReport, RuntimeObservation}; +use protector_behavior::{AgentReport, RuntimeObservation, RuntimeReport}; use tokio::sync::mpsc; use coalesce::Coalescer; @@ -162,11 +162,22 @@ async fn main() -> anyhow::Result<()> { // or the drained buffer if this new distinct key hit the max-size cap. let immediate = coalescer.offer(obs); if !immediate.is_empty() { - reported_since_tick += reporter.send(&immediate).await; + reported_since_tick += reporter + .send(&RuntimeReport { + observations: immediate, + liveness: None, + }) + .await; } } None => { - reporter.send(&coalescer.drain()).await; // drain on shutdown + // drain on shutdown + reporter + .send(&RuntimeReport { + observations: coalescer.drain(), + liveness: None, + }) + .await; break; } }, @@ -174,7 +185,12 @@ async fn main() -> anyhow::Result<()> { // Window elapsed: flush the coalesced batch. Skip the round-trip when // nothing accumulated (a quiet window stays silent). if !coalescer.is_empty() { - reported_since_tick += reporter.send(&coalescer.drain()).await; + reported_since_tick += reporter + .send(&RuntimeReport { + observations: coalescer.drain(), + liveness: None, + }) + .await; } } _ = heartbeat.tick() => { @@ -196,12 +212,20 @@ async fn main() -> anyhow::Result<()> { // 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( + let liveness = build_agent_report( node, beacon_probes.snapshot(), reported_since_tick as u64, ); - reporter.send_report(&report).await; + // Liveness rides the unified envelope (JEF-336): a quiet node still POSTs + // — empty observations, liveness present — so it reads HEALTHY-quiet, not + // blind, instead of the old single-beacon POST the engine 422-rejected. + reporter + .send(&RuntimeReport { + observations: Vec::new(), + liveness: Some(liveness), + }) + .await; } reported_since_tick = 0; } diff --git a/agent/protector-agent/src/reporter.rs b/agent/protector-agent/src/reporter.rs index 143f74f..e8946f4 100644 --- a/agent/protector-agent/src/reporter.rs +++ b/agent/protector-agent/src/reporter.rs @@ -1,8 +1,14 @@ -//! The reporter: batches observations and POSTs them to the engine's behavioral -//! ingest (`/behavior`, ADR-0014). In-cluster, mesh-protected hop; the agent never -//! sends behavioral data anywhere else (the data is a map of the cluster — it stays +//! The reporter: batches a window's observations and (when this node has one) its per-node +//! liveness beacon into ONE [`RuntimeReport`] envelope and POSTs it to the engine's unified +//! behavioral ingest (`/behavior`, ADR-0014 / JEF-336). In-cluster, mesh-protected hop; the agent +//! never sends behavioral data anywhere else (the data is a map of the cluster — it stays //! in-cluster, per VISION's local-first conviction). //! +//! One endpoint, one envelope (JEF-336): liveness always travels with the report, so a quiet node +//! still POSTs (empty observations, liveness present) and the engine reads it HEALTHY-quiet rather +//! than blind. This replaced a separate `/agent-liveness` beacon POST that shipped a single +//! `AgentReport` the engine's array-typed handler 422-rejected — the "no agents connected" bug. +//! //! The POST carries an `Authorization: Bearer ` (Fix A) so the engine can //! reject forged observations from any other caller that can reach :9999. The token //! is the shared secret the engine also reads; authentication (this header) is @@ -23,7 +29,7 @@ use std::time::Duration; -use protector_behavior::{AgentReport, RuntimeObservation}; +use protector_behavior::RuntimeReport; /// Re-resolve the token after this many consecutive 401s. Small so a genuine skew heals /// fast, but >1 so a single transient 401 (e.g. an engine mid-restart that hasn't loaded @@ -39,13 +45,11 @@ const ERROR_EVERY_N_REJECTIONS: u64 = 20; /// (a stale-then-fresh token) without touching the filesystem or sleeping. type TokenSource = Box Option + Send>; -/// POSTs batches of [`RuntimeObservation`]s to `{base}/behavior`, and per-node liveness beacons -/// (JEF-308) to `{base}/agent-liveness`. +/// POSTs per-window [`RuntimeReport`] envelopes (observations + optional per-node liveness beacon, +/// JEF-336) to `{base}/behavior`. pub struct Reporter { client: reqwest::Client, url: String, - /// The per-node agent-liveness beacon endpoint (`{base}/agent-liveness`, JEF-308). - liveness_url: String, /// Shared-secret bearer for the engine's ingest authn (Fix A). `None` = send no /// `Authorization` header (the engine then runs the ingest unauthenticated, which /// it warns about); set it once the Secret has rolled out. @@ -122,7 +126,6 @@ impl Reporter { Self { client, url: format!("{base}/behavior"), - liveness_url: format!("{base}/agent-liveness"), token, token_source: source, consecutive_401s: 0, @@ -131,46 +134,16 @@ impl Reporter { } } - /// Build the POST for `batch`, attaching the bearer header when a token is - /// configured. Split out so the header wiring is unit-testable without a server. - fn build_request(&self, batch: &[RuntimeObservation]) -> reqwest::RequestBuilder { - let mut req = self.client.post(&self.url).json(batch); - if let Some(token) = &self.token { - req = req.bearer_auth(token); - } - req - } - - /// Build the POST for a per-node liveness beacon (JEF-308), attaching the same bearer as the - /// observation path. Split out so the header/URL wiring is unit-testable without a server. - fn build_report_request(&self, report: &AgentReport) -> reqwest::RequestBuilder { - let mut req = self.client.post(&self.liveness_url).json(report); + /// Build the POST for one [`RuntimeReport`] envelope, attaching the bearer header when a token + /// is configured. Split out so the header/URL wiring is unit-testable without a server. + fn build_request(&self, report: &RuntimeReport) -> reqwest::RequestBuilder { + let mut req = self.client.post(&self.url).json(report); if let Some(token) = &self.token { req = req.bearer_auth(token); } req } - /// POST one per-node liveness beacon (JEF-308) to `{base}/agent-liveness`. Best-effort like - /// [`send`](Self::send): a failed beacon is logged and dropped — a missed beacon costs a little - /// freshness, and the engine's TTL means a node that genuinely stops beaconing reads blind, - /// which is the honest outcome. Sent every window even when the node is quiet, so a quiet node - /// reads healthy-quiet, not blind. `&mut self` (like [`send`](Self::send)) so the future stays - /// `Send` for `tokio::spawn` — the boxed token source makes a shared `&Reporter` non-`Send`. - pub async fn send_report(&mut self, report: &AgentReport) { - match self.build_report_request(report).send().await { - Ok(resp) if resp.status().is_success() => { - tracing::debug!(node = %report.node, "reported agent liveness"); - } - Ok(resp) => { - tracing::warn!(status = %resp.status(), "agent-liveness ingest rejected beacon"); - } - Err(error) => { - tracing::warn!(%error, "agent-liveness ingest unreachable"); - } - } - } - /// Cumulative (delivered, rejected) tallies for the periodic heartbeat (JEF-240). /// `delivered` counts accepted observations; `rejected` counts rejected batches. pub fn counters(&self) -> (u64, u64) { @@ -220,21 +193,28 @@ impl Reporter { } } - /// Send one batch; returns how many observations were accepted (0 on failure or an - /// empty batch). Best-effort: a failed POST is logged and dropped — behavioral - /// evidence is additive, so a lost batch costs a little freshness, never correctness, - /// and must never wedge the agent. The caller rolls the count into an interval - /// heartbeat; per-send detail stays at debug. + /// Send one per-window [`RuntimeReport`] envelope (observations + optional liveness, JEF-336); + /// returns how many observations were accepted (0 on failure, or when the envelope carries + /// neither observations nor liveness). An envelope with empty observations but a liveness beacon + /// IS sent — that is the quiet-node path that keeps a silent node reading HEALTHY-quiet, not + /// blind. Best-effort: a failed POST is logged and dropped — behavioral evidence is additive, so + /// a lost report costs a little freshness, never correctness, and must never wedge the agent. + /// The caller rolls the count into an interval heartbeat; per-send detail stays at debug. /// /// On a run of 401s the token is re-resolved (JEF-240) so a secret rotation self-heals /// without a pod restart; the run-length resets on the first 2xx. - pub async fn send(&mut self, batch: &[RuntimeObservation]) -> usize { - if batch.is_empty() { + pub async fn send(&mut self, report: &RuntimeReport) -> usize { + if report.observations.is_empty() && report.liveness.is_none() { return 0; } - match self.build_request(batch).send().await { + let n = report.observations.len(); + match self.build_request(report).send().await { Ok(resp) if resp.status().is_success() => { - tracing::debug!(n = batch.len(), "reported behavioral observations"); + tracing::debug!( + observations = n, + liveness = report.liveness.is_some(), + "reported runtime envelope" + ); if self.consecutive_401s > 0 { tracing::info!( after_401s = self.consecutive_401s, @@ -242,8 +222,8 @@ impl Reporter { ); } self.consecutive_401s = 0; - self.delivered_total = self.delivered_total.saturating_add(batch.len() as u64); - batch.len() + self.delivered_total = self.delivered_total.saturating_add(n as u64); + n } Ok(resp) if resp.status() == reqwest::StatusCode::UNAUTHORIZED => { self.rejected_total = self.rejected_total.saturating_add(1); @@ -268,7 +248,9 @@ impl Reporter { #[cfg(test)] mod tests { use super::*; - use protector_behavior::{Attribution, Behavior, SecretReadSource}; + use protector_behavior::{ + AgentReport, Attribution, Behavior, RuntimeObservation, SecretReadSource, + }; fn reporter_with(token: Option<&str>) -> Reporter { let owned = token.map(str::to_string); @@ -276,7 +258,6 @@ mod tests { Reporter { client: reqwest::Client::new(), url: "http://engine.svc:9999/behavior".to_string(), - liveness_url: "http://engine.svc:9999/agent-liveness".to_string(), token: owned, token_source: Box::new(move || for_source.clone()), consecutive_401s: 0, @@ -285,8 +266,8 @@ mod tests { } } - fn sample_batch() -> Vec { - vec![RuntimeObservation { + fn sample_observation() -> RuntimeObservation { + RuntimeObservation { attribution: Attribution::by_namespaced_name("app", "web"), source: Some("agent".into()), observed_at_ms: None, @@ -295,7 +276,14 @@ mod tests { secret: "app/session-key".into(), source: SecretReadSource::Mounted, }, - }] + } + } + + fn sample_report() -> RuntimeReport { + RuntimeReport { + observations: vec![sample_observation()], + liveness: None, + } } /// When a token is configured the POST carries `Authorization: Bearer `. @@ -303,7 +291,7 @@ mod tests { fn attaches_bearer_header_when_token_set() { let reporter = reporter_with(Some("s3cr3t")); let req = reporter - .build_request(&sample_batch()) + .build_request(&sample_report()) .build() .expect("request builds"); let auth = req @@ -313,22 +301,26 @@ mod tests { assert_eq!(auth, "Bearer s3cr3t"); } - /// JEF-308: a liveness beacon POSTs to `{base}/agent-liveness` carrying the same bearer. + /// JEF-336: the unified envelope — including a quiet-node liveness-only report — POSTs to the + /// single `{base}/behavior` route carrying the same bearer (no separate `/agent-liveness`). #[test] - fn liveness_beacon_targets_agent_liveness_with_bearer() { + fn liveness_rides_the_behavior_envelope_with_bearer() { let reporter = reporter_with(Some("s3cr3t")); - let report = AgentReport { - node: "node-a".into(), - probes_loaded: 6, - probes_total: 6, - signals_emitted: 3, - observed_at_ms: None, + let report = RuntimeReport { + observations: Vec::new(), + liveness: Some(AgentReport { + node: "node-a".into(), + probes_loaded: 6, + probes_total: 6, + signals_emitted: 3, + observed_at_ms: None, + }), }; let req = reporter - .build_report_request(&report) + .build_request(&report) .build() .expect("request builds"); - assert_eq!(req.url().as_str(), "http://engine.svc:9999/agent-liveness"); + assert_eq!(req.url().as_str(), "http://engine.svc:9999/behavior"); assert_eq!( req.headers().get(reqwest::header::AUTHORIZATION).unwrap(), "Bearer s3cr3t" @@ -341,14 +333,14 @@ mod tests { fn omits_bearer_header_when_token_unset() { let reporter = reporter_with(None); let req = reporter - .build_request(&sample_batch()) + .build_request(&sample_report()) .build() .expect("request builds"); assert!( req.headers().get(reqwest::header::AUTHORIZATION).is_none(), "no Authorization header when token unset" ); - // The body is still the JSON batch — auth is the only thing that changed. + // The body is still the JSON envelope — auth is the only thing that changed. assert_eq!(req.url().as_str(), "http://engine.svc:9999/behavior"); } @@ -403,7 +395,7 @@ mod tests { // The fresh bearer is what subsequent posts attach. let req = reporter - .build_request(&sample_batch()) + .build_request(&sample_report()) .build() .expect("request builds"); assert_eq!( diff --git a/behavior/src/lib.rs b/behavior/src/lib.rs index b0b612e..07af847 100644 --- a/behavior/src/lib.rs +++ b/behavior/src/lib.rs @@ -358,6 +358,29 @@ impl AgentReport { } } +/// A per-window **runtime report** (JEF-336): the single envelope every sensor POSTs to the +/// engine's unified runtime ingest (`/behavior`). It carries the window's normalized +/// [`RuntimeObservation`]s AND — for a sensor that has one — its per-node liveness +/// [`AgentReport`], so liveness ALWAYS travels with the report. That is what keeps the JEF-308 +/// "quiet ≠ blind" guarantee honest: a node that saw nothing still POSTs an envelope with empty +/// `observations` and its `liveness` present, so the engine records it HEALTHY-quiet instead of +/// reading it blind for want of a beacon. +/// +/// `liveness` is [`Option`] so the ADR-0003 tool-agnostic port still accepts a third-party sensor +/// that sends only observations (it has no agent-specific `probes_loaded` to report). Both fields +/// are defaulted and skip-if-empty on the wire, so an observations-only envelope is just +/// `{"observations":[...]}` and a quiet liveness-only one is `{"liveness":{...}}`. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct RuntimeReport { + /// The normalized observations seen this window — possibly empty (a quiet node still reports). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub observations: Vec, + /// This sensor's per-node liveness beacon (JEF-308), when it has one. Absent for a + /// node-agnostic third-party sensor with no agent-specific liveness to report. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub liveness: Option, +} + #[cfg(test)] mod tests { use super::*; @@ -702,6 +725,81 @@ mod tests { assert_eq!(serde_json::from_value::(v).unwrap(), report); } + #[test] + fn runtime_report_round_trips_with_observations_and_liveness() { + // JEF-336: the unified envelope carries the window's observations AND the per-node + // liveness beacon in one shape, and round-trips byte-for-byte. + let report = RuntimeReport { + observations: vec![RuntimeObservation { + attribution: Attribution::by_pod_uid("uid"), + source: Some("protector-agent".into()), + observed_at_ms: None, + node: Some("node-a".into()), + behavior: Behavior::Alert { + rule: "Terminal shell in container".into(), + }, + }], + liveness: Some(AgentReport { + node: "node-a".into(), + probes_loaded: 6, + probes_total: 6, + signals_emitted: 1, + observed_at_ms: None, + }), + }; + let v = serde_json::to_value(&report).unwrap(); + assert!(v.get("observations").is_some()); + assert_eq!(v["liveness"]["node"], serde_json::json!("node-a")); + assert_eq!(serde_json::from_value::(v).unwrap(), report); + } + + #[test] + fn runtime_report_omits_empty_observations_and_absent_liveness() { + // A quiet node's envelope: no observations, liveness present — `observations` is omitted + // (skip_serializing_if empty) so the wire is just `{"liveness":{...}}`. + let quiet = RuntimeReport { + observations: Vec::new(), + liveness: Some(AgentReport { + node: "node-a".into(), + probes_loaded: 6, + probes_total: 6, + signals_emitted: 0, + observed_at_ms: None, + }), + }; + let v = serde_json::to_value(&quiet).unwrap(); + assert!( + v.get("observations").is_none(), + "empty observations omitted from the wire" + ); + assert!(v.get("liveness").is_some()); + assert_eq!(serde_json::from_value::(v).unwrap(), quiet); + + // A third-party observations-only envelope: liveness absent → `liveness` omitted, and it + // deserializes back with `liveness: None` (the ADR-0003 tool-agnostic path). + let obs_only = RuntimeReport { + observations: vec![RuntimeObservation { + attribution: Attribution::by_namespaced_name("app", "web"), + source: None, + observed_at_ms: None, + node: None, + behavior: Behavior::LibraryLoaded { + name: "openssl".into(), + }, + }], + liveness: None, + }; + let v = serde_json::to_value(&obs_only).unwrap(); + assert!( + v.get("liveness").is_none(), + "absent liveness omitted from the wire" + ); + assert_eq!( + serde_json::from_value::(v).unwrap(), + obs_only + ); + } + #[test] fn only_alert_corroborates() { assert!(Behavior::Alert { rule: "x".into() }.is_alert()); diff --git a/charts/protector/templates/agent-daemonset.yaml b/charts/protector/templates/agent-daemonset.yaml index 9705e41..617b1d7 100644 --- a/charts/protector/templates/agent-daemonset.yaml +++ b/charts/protector/templates/agent-daemonset.yaml @@ -69,7 +69,8 @@ spec: {{- end }} env: # The node this agent runs on (JEF-308), via the downward API. The agent stamps it - # onto every observation AND onto its per-node liveness beacon (/agent-liveness), so + # onto every observation AND onto the per-node liveness beacon it folds into the same + # /behavior report envelope (JEF-336), so # the engine can report runtime-corroboration coverage PER NODE ("blind on node X") # rather than fleet-aggregate — and a Ready-but-blind agent (probes failed to load) # still reads blind. Signal-flow liveness, not pod-Ready. diff --git a/engine/src/engine/observe/mod.rs b/engine/src/engine/observe/mod.rs index 84b7836..a1e4346 100644 --- a/engine/src/engine/observe/mod.rs +++ b/engine/src/engine/observe/mod.rs @@ -214,7 +214,7 @@ pub(crate) fn report_resource(object: &DynamicObject) -> Option { /// definition rather than a hand-synced duplicate. Re-exported here because the Observer /// and adapters refer to it as `observe::RuntimeObservation`. [`Attribution`] (how an /// observation is attributed to a workload) is re-exported alongside it for the same reason. -pub use protector_behavior::{AgentReport, Attribution, RuntimeObservation}; +pub use protector_behavior::{AgentReport, Attribution, RuntimeObservation, RuntimeReport}; /// The normalized API secret-read lifted from the apiserver audit log (JEF-269) — the /// corroborating runtime signal the eBPF agent can't observe. Re-exported here because the diff --git a/engine/src/engine/observe/runtime.rs b/engine/src/engine/observe/runtime.rs index 88da90f..9047f8c 100644 --- a/engine/src/engine/observe/runtime.rs +++ b/engine/src/engine/observe/runtime.rs @@ -1,8 +1,14 @@ //! The RuntimeEvidence ingest: live behavioral signals that supply the action bar's //! `corroborated-now` predicate. The first-party eBPF agent (and any sensor with a -//! translation adapter) POSTs normalized [`RuntimeObservation`]s to the tool-agnostic -//! behavioral port (ADR-0003), the `/behavior` route the engine exposes; [`RuntimeEvents`] -//! holds the recent ones. +//! translation adapter) POSTs a per-window [`RuntimeReport`] envelope to the tool-agnostic +//! behavioral port (ADR-0003), the single `/behavior` route the engine exposes; [`RuntimeEvents`] +//! holds the recent observations. +//! +//! One envelope, one endpoint (JEF-336): the report carries the window's observations AND — when +//! the sensor has one — its per-node liveness beacon (JEF-308), so liveness ALWAYS travels with the +//! report. A quiet node still POSTs (empty observations, liveness present) and reads HEALTHY-quiet +//! rather than blind. For backward compatibility the route also accepts a *bare* `[...]` array of +//! observations (a legacy / third-party observations-only poster) — see [`BehaviorIngest`]. //! //! A runtime signal is a *stream*, not a Kubernetes object, so it can't be reflected like //! the rest of the graph — hence the HTTP ingest. @@ -27,7 +33,7 @@ use tokio::sync::mpsc::Sender; use super::ingest_guard::{ DEFAULT_BURST, DEFAULT_RATE_PER_SEC, IngestToken, RateLimit, bearer_auth, rate_limit, }; -use super::{AgentReport, RuntimeObservation}; +use super::{RuntimeObservation, RuntimeReport}; use crate::engine::state::AgentLivenessStore; /// A time-windowed store of recent runtime observations. Thread-safe so the HTTP @@ -104,39 +110,72 @@ fn same_signal(a: &RuntimeObservation, b: &RuntimeObservation) -> bool { a.behavior == b.behavior && a.attribution == b.attribution } -/// Shared state for the ingest handlers: the event store, a wake channel, and the per-node -/// agent-liveness store (JEF-308) the `/agent-liveness` beacon feeds. +/// Shared state for the ingest handler: the event store, a wake channel, and the per-node +/// agent-liveness store (JEF-308) the report envelope's `liveness` feeds. type IngestState = (Arc, Sender<()>, Arc); -/// Receive a batch of normalized [`RuntimeObservation`]s on the tool-agnostic -/// behavioral port (ADR-0014) — the shape the first-party eBPF agent (and any sensor -/// with a translation adapter) POSTs. Each is recorded; the engine is woken once, and -/// only if the batch actually changed the store. The agent re-reports the same -/// connections continuously, so most batches are pure repeats — those refresh TTLs but -/// must not churn a process pass, which is the whole point of gating the wake here. +/// What the `/behavior` route accepts (JEF-336). The first-party agent POSTs a +/// [`RuntimeReport`] envelope (`{observations, liveness}`); a legacy or third-party sensor may +/// still POST a *bare* `[...]` array of observations. Untagged, envelope-first: an object matches +/// [`Self::Envelope`], a JSON array matches [`Self::Bare`], so the two wire forms never collide. +/// +/// Accepting both is the zero-break path that stays truest to the ADR-0003 tool-agnostic port — a +/// third-party poster that only speaks the pre-existing bare-array contract keeps working, while +/// the first-party agent gets liveness-with-observations in one envelope. +#[derive(serde::Deserialize)] +#[serde(untagged)] +enum BehaviorIngest { + /// The JEF-336 envelope: observations plus optional per-node liveness. + Envelope(RuntimeReport), + /// The legacy / third-party bare array of observations (no liveness). + Bare(Vec), +} + +impl BehaviorIngest { + /// Normalize either accepted wire form into a [`RuntimeReport`]. A bare array becomes an + /// envelope with no liveness. + fn into_report(self) -> RuntimeReport { + match self { + BehaviorIngest::Envelope(report) => report, + BehaviorIngest::Bare(observations) => RuntimeReport { + observations, + liveness: None, + }, + } + } +} + +/// Receive one per-window [`RuntimeReport`] on the tool-agnostic behavioral port (ADR-0014) — the +/// envelope the first-party eBPF agent (and any sensor with a translation adapter) POSTs, or a bare +/// legacy array (see [`BehaviorIngest`]). Observations are recorded and the engine is woken once, +/// and only if the batch actually changed the store — the agent re-reports the same connections +/// continuously, so most batches are pure repeats that refresh TTLs but must not churn a process +/// pass. Any `liveness` in the envelope is recorded into the agent-liveness store; unlike a +/// behavior it does NOT wake the engine (liveness is read at pass time). Because the envelope +/// arrives every window even when the node saw nothing, a quiet node reads HEALTHY-quiet, not blind. async fn ingest_behavior( - State((events, notify, _liveness)): State, - Json(observations): Json>, + State((events, notify, liveness)): State, + Json(payload): Json, ) -> StatusCode { - if record_batch(&events, observations) { + if ingest_report(&events, &liveness, payload.into_report()) { let _ = notify.try_send(()); } StatusCode::OK } -/// Receive a batch of per-node [`AgentReport`] liveness beacons (JEF-308) and record them into the -/// liveness store. Unlike a behavior, a beacon does NOT wake the engine — liveness is read at pass -/// time; a beacon only refreshes freshness. A beacon arrives every window even when the node saw -/// nothing, which is exactly what lets a quiet node read HEALTHY-quiet instead of blind. Same body -/// cap / authn / rate limit as the sibling routes; over-cap batches are truncated defensively. -async fn ingest_agent_liveness( - State((_events, _notify, liveness)): State, - Json(reports): Json>, -) -> StatusCode { - for report in reports.into_iter().take(MAX_BATCH) { - liveness.record(report); +/// Apply one [`RuntimeReport`] to the stores: record its liveness beacon (if any) and its +/// observations, returning whether the observation store actually **changed** (so the caller wakes +/// the engine only on a real change — liveness alone never wakes it). Split out and pure-over-the- +/// stores so the envelope's dual effect is unit-testable without an HTTP server. +fn ingest_report( + events: &RuntimeEvents, + liveness: &AgentLivenessStore, + report: RuntimeReport, +) -> bool { + if let Some(beacon) = report.liveness { + liveness.record(beacon); } - StatusCode::OK + record_batch(events, report.observations) } /// Upper bound on observations accepted per `/behavior` batch. Each `record()` is an @@ -168,10 +207,11 @@ fn record_batch(events: &RuntimeEvents, observations: Vec) - changed } -/// Serve the runtime-evidence ingest. `/behavior` accepts a batch of normalized observations -/// on the tool-agnostic behavioral port (ADR-0003) from the first-party eBPF agent or any -/// sensor with a translation adapter; `/agent-liveness` accepts the per-node liveness beacon. -/// This is the cluster-facing glue; the store it drives is what the tests cover. +/// Serve the runtime-evidence ingest. The single `/behavior` route accepts a per-window +/// [`RuntimeReport`] envelope (observations + optional per-node liveness) on the tool-agnostic +/// behavioral port (ADR-0003) from the first-party eBPF agent or any sensor with a translation +/// adapter — or a bare legacy `[...]` array of observations (JEF-336). This is the cluster-facing +/// glue; the store it drives is what the tests cover. pub async fn serve_runtime( addr: SocketAddr, events: Arc, @@ -190,9 +230,9 @@ pub async fn serve_runtime( let limiter = RateLimit::new(DEFAULT_RATE_PER_SEC, DEFAULT_BURST); let mut app = Router::new() + // One endpoint (JEF-336): the report envelope carries observations AND the per-node + // agent-liveness beacon (JEF-308) — signal-flow liveness, not pod-Ready. .route("/behavior", post(ingest_behavior)) - // The per-node agent-liveness beacon (JEF-308) — signal-flow liveness, not pod-Ready. - .route("/agent-liveness", post(ingest_agent_liveness)) // A real alert/batch is small; cap the body so a client can't OOM the engine // with a giant POST (mirrors the webhook server). The body cap, MAX_EVENTS, and // the per-batch MAX_BATCH all remain in force alongside authn + rate limiting. @@ -205,7 +245,7 @@ pub async fn serve_runtime( match token { Some(token) => { app = app.layer(axum::middleware::from_fn_with_state(token, bearer_auth)); - tracing::info!(%addr, "runtime-evidence ingest listening (/behavior, /agent-liveness) — bearer-authenticated"); + tracing::info!(%addr, "runtime-evidence ingest listening (/behavior) — bearer-authenticated"); } None => { tracing::warn!( @@ -229,7 +269,7 @@ pub async fn serve_runtime( #[cfg(test)] mod tests { - use super::super::Attribution; + use super::super::{AgentReport, Attribution}; use super::*; use crate::engine::graph::Behavior; use serde_json::json; @@ -331,6 +371,122 @@ mod tests { assert!(!obs[1].behavior.is_alert()); } + fn liveness_store() -> Arc { + Arc::new(AgentLivenessStore::new(Duration::from_secs(300))) + } + + fn beacon(node: &str, signals: u64) -> AgentReport { + AgentReport { + node: node.into(), + probes_loaded: 6, + probes_total: 6, + signals_emitted: signals, + observed_at_ms: None, + } + } + + /// JEF-336: both accepted wire forms deserialize — the `{observations, liveness}` envelope and + /// a bare legacy `[...]` array — and normalize to the same [`RuntimeReport`] shape. + #[test] + fn behavior_ingest_accepts_both_the_envelope_and_a_bare_array() { + // The first-party agent's envelope: observations + a per-node liveness beacon. + let envelope: BehaviorIngest = serde_json::from_value(json!({ + "observations": [ + {"namespace": "app", "pod": "web", + "behavior": {"kind": "alert", "rule": "Terminal shell in container"}} + ], + "liveness": {"node": "node-a", "probes_loaded": 6, "probes_total": 6, "signals_emitted": 1} + })) + .expect("envelope deserializes"); + let report = envelope.into_report(); + assert_eq!(report.observations.len(), 1); + assert_eq!(report.liveness.as_ref().unwrap().node, "node-a"); + + // A legacy / third-party bare array: observations only, no liveness. + let bare: BehaviorIngest = serde_json::from_value(json!([ + {"namespace": "app", "pod": "web", + "behavior": {"kind": "network_connection", "peer": "1.2.3.4:443", "internet": true}} + ])) + .expect("bare array deserializes"); + let report = bare.into_report(); + assert_eq!(report.observations.len(), 1); + assert!( + report.liveness.is_none(), + "a bare array carries no liveness (third-party path)" + ); + } + + /// The envelope's observations still ingest exactly as before (incl. the Alert path), and a + /// fresh batch wakes the engine (returns changed). + #[test] + fn ingest_report_records_observations_and_signals_a_change() { + let events = RuntimeEvents::new(Duration::from_secs(300)); + let liveness = liveness_store(); + let report = RuntimeReport { + observations: vec![obs("Terminal shell in container")], + liveness: None, + }; + assert!( + ingest_report(&events, &liveness, report), + "a fresh observation changes the store → wake the engine" + ); + let current = events.current(); + assert_eq!(current.len(), 1); + assert!(current[0].behavior.is_alert()); + // No liveness in the envelope → the liveness store stays empty. + assert!(liveness.snapshot().is_empty()); + } + + /// The engine records liveness carried in the envelope — the JEF-336 fix for the 422 bug. + #[test] + fn ingest_report_records_liveness_from_the_envelope() { + let events = RuntimeEvents::new(Duration::from_secs(300)); + let liveness = liveness_store(); + let report = RuntimeReport { + observations: vec![obs("shell")], + liveness: Some(beacon("node-a", 1)), + }; + ingest_report(&events, &liveness, report); + let snap = liveness.snapshot(); + assert_eq!(snap.len(), 1, "the node's liveness was recorded"); + assert!(snap.contains_key("node-a")); + } + + /// A quiet node: empty observations + liveness present. The liveness is recorded, but with no + /// observation change the engine is NOT woken — liveness alone never churns a pass. + #[test] + fn quiet_node_records_liveness_without_waking_the_engine() { + let events = RuntimeEvents::new(Duration::from_secs(300)); + let liveness = liveness_store(); + let report = RuntimeReport { + observations: Vec::new(), + liveness: Some(beacon("node-a", 0)), + }; + assert!( + !ingest_report(&events, &liveness, report), + "no observation change → no wake" + ); + assert!(events.current().is_empty()); + assert!( + liveness.snapshot().contains_key("node-a"), + "a quiet node still records liveness (quiet ≠ blind)" + ); + } + + /// A third-party liveness-less payload still ingests its observations, leaving liveness empty. + #[test] + fn liveness_less_payload_still_ingests_observations() { + let events = RuntimeEvents::new(Duration::from_secs(300)); + let liveness = liveness_store(); + let report = BehaviorIngest::Bare(vec![obs("shell")]).into_report(); + assert!(ingest_report(&events, &liveness, report)); + assert_eq!(events.current().len(), 1); + assert!( + liveness.snapshot().is_empty(), + "a third-party observations-only poster records no liveness" + ); + } + #[test] fn connection_fingerprints_are_coarse_so_peer_churn_does_not_rejudge() { // Different peers collapse to the same coarse key, so mundane connection churn diff --git a/engine/src/engine/run_loop.rs b/engine/src/engine/run_loop.rs index 97dc63d..5a94352 100644 --- a/engine/src/engine/run_loop.rs +++ b/engine/src/engine/run_loop.rs @@ -290,8 +290,9 @@ pub async fn run_watch( // Diagnostic judgement log: the full prompt + raw reply + verdict per judgement, // written by the adjudicator for later inspection. let journal = std::sync::Arc::new(state::JudgementLog::new()); - // Per-node agent-liveness (JEF-308): the TTL'd store the `/agent-liveness` beacon feeds and - // the engine reads each pass to classify runtime-corroboration coverage per node. Same 300s + // Per-node agent-liveness (JEF-308): the TTL'd store the `/behavior` report envelope's liveness + // feeds (JEF-336) and the engine reads each pass to classify runtime-corroboration coverage per + // node. Same 300s // freshness window as the runtime feed — a node whose agent stopped beaconing goes stale and // reads blind. Shared (`Arc`) between the ingest task and the engine loop. let agent_liveness = std::sync::Arc::new(state::AgentLivenessStore::new(