Skip to content

fix(runtime-ingest): unify /behavior + /agent-liveness into one report envelope (JEF-336) #183

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
thejefflarson merged 1 commit into from
Jul 6, 2026
Merged

fix(runtime-ingest): unify /behavior + /agent-liveness into one report envelope (JEF-336) #183

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 30 additions & 6 deletions agent/protector-agent/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -162,19 +162,35 @@ 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;
}
},
_ = window.tick() => {
// 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() => {
Expand All @@ -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;
}
Expand Down
138 changes: 65 additions & 73 deletions agent/protector-agent/src/reporter.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 <token>` (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
Expand All @@ -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
Expand All @@ -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<dyn FnMut() -> Option<String> + 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.
Expand Down Expand Up @@ -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,
Expand All @@ -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) {
Expand Down Expand Up @@ -220,30 +193,37 @@ 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,
"behavior ingest accepted a batch again — ingest auth recovered"
);
}
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);
Expand All @@ -268,15 +248,16 @@ 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);
let for_source = owned.clone();
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,
Expand All @@ -285,8 +266,8 @@ mod tests {
}
}

fn sample_batch() -> Vec<RuntimeObservation> {
vec![RuntimeObservation {
fn sample_observation() -> RuntimeObservation {
RuntimeObservation {
attribution: Attribution::by_namespaced_name("app", "web"),
source: Some("agent".into()),
observed_at_ms: None,
Expand All @@ -295,15 +276,22 @@ 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 <token>`.
#[test]
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
Expand All @@ -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"
Expand All @@ -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");
}

Expand Down Expand Up @@ -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!(
Expand Down
Loading
Loading