car-eventlog 0.47.0

//! Live metrics aggregation + threshold alerting (EPIC G / G1).
//!
//! `metrics_totals` sums token/cost/latency and `harness_metrics` scores the
//! six operational dimensions, but neither gives an operator a single live
//! rollup (success rate, cost, latency, rejections, approvals) or fires an
//! alert when a threshold is crossed. This module supplies both as pure folds
//! over the event stream, so the daemon's `metrics.*` surface can render live
//! state and raise operational alerts. Deterministic and side-effect-free, like
//! `harness_metrics` and `cost_by_agent_of`.

use crate::{cost_by_agent_of, metrics_totals_of, AgentCost, Event, EventKind};
use serde::{Deserialize, Serialize};

/// A live operational rollup of the event stream (EPIC G / G1).
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct MetricsSummary {
    /// Total events considered.
    pub total_events: usize,
    /// Actions that succeeded / failed / were rejected (validator/policy/gate).
    pub actions_succeeded: u64,
    pub actions_failed: u64,
    pub actions_rejected: u64,
    /// succeeded / (succeeded + failed + rejected). 1.0 when nothing ran.
    pub success_rate: f64,
    /// error rate = 1 - success_rate (the alertable complement).
    pub error_rate: f64,
    /// Summed cost + tokens across the **retained** metered events — a
    /// windowed fold that shrinks when retention trims metered events.
    pub cost_usd: f64,
    /// Monotonic cumulative cost over the log's lifetime (G1). Unlike
    /// `cost_usd`, this survives retention trims: [`summarize_log`] reads it
    /// from [`crate::EventLog::cumulative_cost_usd`], so the
    /// `max_cost_usd` budget alert can never un-fire because old metered
    /// events were evicted. When folded from a bare slice via [`summarize`]
    /// it equals `cost_usd` (a slice carries no trim history).
    #[serde(default)]
    pub cumulative_cost_usd: f64,
    pub tokens_in: u64,
    pub tokens_out: u64,
    /// Mean latency across events that carried a duration metric (0 if none).
    pub avg_latency_ms: f64,
    /// HITL approvals recorded and permission decisions taken.
    pub approvals_recorded: u64,
    pub permission_decisions: u64,
    /// Admission-gate rejections (information-flow / concurrency / policy) and
    /// hard policy violations — the safety-refusal counters.
    pub gate_rejections: u64,
    pub policy_violations: u64,
    /// Goal-loop verifier outcomes. `goal_evaluations` counts deterministic
    /// verifier passes; `goals_met` counts passes with `met: true`; and
    /// `goals_ungrounded` counts passes where the verifier could not treat the
    /// completion evidence as grounded.
    #[serde(default)]
    pub goal_evaluations: u64,
    #[serde(default)]
    pub goals_met: u64,
    #[serde(default)]
    pub goals_ungrounded: u64,
    /// Per-agent cost breakdown (G3), folded in so one call renders the whole
    /// live picture.
    pub cost_by_agent: Vec<AgentCost>,
}

/// Fold a slice of events into a [`MetricsSummary`] (EPIC G / G1).
pub fn summarize(events: &[Event]) -> MetricsSummary {
    let totals = metrics_totals_of(events);
    let mut s = MetricsSummary {
        total_events: events.len(),
        cost_usd: totals.cost_usd,
        // A bare slice has no trim history; the live counter is applied by
        // `summarize_log`.
        cumulative_cost_usd: totals.cost_usd,
        tokens_in: totals.tokens_in,
        tokens_out: totals.tokens_out,
        cost_by_agent: cost_by_agent_of(events),
        ..Default::default()
    };
    for e in events {
        match e.kind {
            EventKind::ActionSucceeded => s.actions_succeeded += 1,
            EventKind::ActionFailed => s.actions_failed += 1,
            EventKind::ActionRejected => s.actions_rejected += 1,
            EventKind::ApprovalRecorded => s.approvals_recorded += 1,
            EventKind::PermissionDecision => s.permission_decisions += 1,
            EventKind::PolicyViolation => s.policy_violations += 1,
            EventKind::AdmissionGateDecision => {
                // A rejection/approval-escalation is a refusal to run as-is.
                let decision = e.data.get("decision").and_then(|v| v.as_str());
                if matches!(decision, Some("reject") | Some("needs_approval")) {
                    s.gate_rejections += 1;
                }
            }
            EventKind::GoalEvaluated => {
                s.goal_evaluations += 1;
                if e.data.get("met").and_then(|v| v.as_bool()) == Some(true) {
                    s.goals_met += 1;
                }
                if e.data.get("grounded").and_then(|v| v.as_bool()) == Some(false) {
                    s.goals_ungrounded += 1;
                }
            }
            _ => {}
        }
    }
    let attempted = s.actions_succeeded + s.actions_failed + s.actions_rejected;
    s.success_rate = if attempted == 0 {
        1.0
    } else {
        s.actions_succeeded as f64 / attempted as f64
    };
    s.error_rate = 1.0 - s.success_rate;
    s.avg_latency_ms = if totals.metered_events == 0 {
        0.0
    } else {
        totals.duration_ms / totals.metered_events as f64
    };
    s
}

/// Fold a live [`crate::EventLog`] into a [`MetricsSummary`], carrying the
/// log's **monotonic** cumulative cost counter into `cumulative_cost_usd`
/// (G1). This is the summary the budget alert must be evaluated against: a
/// plain [`summarize`] over `log.events()` re-folds only the retained window,
/// so a retention trim would slide the cost backward and un-fire a
/// `max_cost_usd` alert.
pub fn summarize_log(log: &crate::EventLog) -> MetricsSummary {
    let mut s = summarize(log.events());
    s.cumulative_cost_usd = log.cumulative_cost_usd();
    s
}

/// Operational alert thresholds (EPIC G / G1). Each is optional; an unset
/// threshold never fires. `min_actions` suppresses noisy error-rate alerts on
/// tiny samples.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct AlertThresholds {
    /// Fire when cumulative cost exceeds this many USD. Evaluated against
    /// `MetricsSummary::cumulative_cost_usd` — the monotonic, trim-proof
    /// counter — not the windowed `cost_usd` fold.
    #[serde(default)]
    pub max_cost_usd: Option<f64>,
    /// Fire when the error rate (0..1) exceeds this.
    #[serde(default)]
    pub max_error_rate: Option<f64>,
    /// Fire when average latency exceeds this many milliseconds.
    #[serde(default)]
    pub max_avg_latency_ms: Option<f64>,
    /// Fire when ungrounded goal-verifier passes exceed this count.
    #[serde(default)]
    pub max_goals_ungrounded: Option<u64>,
    /// Don't fire the error-rate alert until at least this many actions ran
    /// (default 5), so one early failure doesn't trip a 100%-error alert.
    #[serde(default)]
    pub min_actions: Option<u64>,
}

/// The class of an operational alert.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AlertKind {
    CostOverage,
    ErrorRate,
    Latency,
    GoalUngrounded,
}

/// A fired operational alert: what tripped, the observed value, and the
/// threshold it crossed.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Alert {
    pub kind: AlertKind,
    pub message: String,
    pub observed: f64,
    pub threshold: f64,
}

/// Evaluate a [`MetricsSummary`] against [`AlertThresholds`], returning every
/// alert that fired (EPIC G / G1). Pure — the caller decides how to deliver
/// (emit an event, push through car-messaging, etc.).
pub fn evaluate_alerts(summary: &MetricsSummary, thresholds: &AlertThresholds) -> Vec<Alert> {
    let mut alerts = Vec::new();
    if let Some(max) = thresholds.max_cost_usd {
        // Cumulative budget: read the monotonic counter, not the windowed
        // fold — retention trims must never un-fire a budget alert (G1).
        if summary.cumulative_cost_usd > max {
            alerts.push(Alert {
                kind: AlertKind::CostOverage,
                message: format!(
                    "cumulative cost ${:.4} exceeds budget ${:.4}",
                    summary.cumulative_cost_usd, max
                ),
                observed: summary.cumulative_cost_usd,
                threshold: max,
            });
        }
    }
    if let Some(max) = thresholds.max_error_rate {
        let attempted =
            summary.actions_succeeded + summary.actions_failed + summary.actions_rejected;
        let min = thresholds.min_actions.unwrap_or(5);
        if attempted >= min && summary.error_rate > max {
            alerts.push(Alert {
                kind: AlertKind::ErrorRate,
                message: format!(
                    "error rate {:.1}% exceeds {:.1}% over {attempted} actions",
                    summary.error_rate * 100.0,
                    max * 100.0
                ),
                observed: summary.error_rate,
                threshold: max,
            });
        }
    }
    if let Some(max) = thresholds.max_avg_latency_ms {
        if summary.avg_latency_ms > max {
            alerts.push(Alert {
                kind: AlertKind::Latency,
                message: format!(
                    "avg latency {:.0}ms exceeds {:.0}ms",
                    summary.avg_latency_ms, max
                ),
                observed: summary.avg_latency_ms,
                threshold: max,
            });
        }
    }
    if let Some(max) = thresholds.max_goals_ungrounded {
        if summary.goals_ungrounded > max {
            alerts.push(Alert {
                kind: AlertKind::GoalUngrounded,
                message: format!(
                    "{} ungrounded goal verifier pass{} exceed threshold {}",
                    summary.goals_ungrounded,
                    if summary.goals_ungrounded == 1 {
                        ""
                    } else {
                        "es"
                    },
                    max
                ),
                observed: summary.goals_ungrounded as f64,
                threshold: max as f64,
            });
        }
    }
    alerts
}

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

    fn ev(kind: EventKind) -> Event {
        Event {
            kind,
            action_id: None,
            proposal_id: None,
            data: Default::default(),
            timestamp: chrono::Utc::now(),
            prev_hash: None,
            hash: None,
        }
    }

    fn goal_ev(met: bool, grounded: bool) -> Event {
        let mut e = ev(EventKind::GoalEvaluated);
        e.data.insert("met".to_string(), serde_json::json!(met));
        e.data
            .insert("grounded".to_string(), serde_json::json!(grounded));
        e
    }

    #[test]
    fn summary_computes_rates_and_counts() {
        let mut events = vec![
            ev(EventKind::ActionSucceeded),
            ev(EventKind::ActionSucceeded),
            ev(EventKind::ActionFailed),
            ev(EventKind::ActionRejected),
            ev(EventKind::ApprovalRecorded),
            ev(EventKind::PolicyViolation),
            goal_ev(false, true),
            goal_ev(true, true),
            goal_ev(true, false),
        ];
        let mut gate = ev(EventKind::AdmissionGateDecision);
        gate.data
            .insert("decision".to_string(), serde_json::Value::from("reject"));
        events.push(gate);

        let s = summarize(&events);
        assert_eq!(s.actions_succeeded, 2);
        assert_eq!(s.actions_failed, 1);
        assert_eq!(s.actions_rejected, 1);
        // 2 / (2+1+1) = 0.5
        assert!((s.success_rate - 0.5).abs() < 1e-9);
        assert!((s.error_rate - 0.5).abs() < 1e-9);
        assert_eq!(s.approvals_recorded, 1);
        assert_eq!(s.policy_violations, 1);
        assert_eq!(s.gate_rejections, 1);
        assert_eq!(s.goal_evaluations, 3);
        assert_eq!(s.goals_met, 2);
        assert_eq!(s.goals_ungrounded, 1);
    }

    #[test]
    fn cost_overage_alert_fires() {
        // Build one metered event carrying a cost, via the real append path.
        let mut log = crate::EventLog::new();
        log.append_metered(
            EventKind::InferenceMetered,
            None,
            None,
            Default::default(),
            Metrics {
                cost_usd: Some(12.5),
                ..Default::default()
            },
        );
        let s = summarize(log.events());
        assert!((s.cost_usd - 12.5).abs() < 1e-9);
        let alerts = evaluate_alerts(
            &s,
            &AlertThresholds {
                max_cost_usd: Some(10.0),
                ..Default::default()
            },
        );
        assert_eq!(alerts.len(), 1);
        assert_eq!(alerts[0].kind, AlertKind::CostOverage);
        assert_eq!(alerts[0].observed, 12.5);
    }

    #[test]
    fn cost_budget_alert_survives_retention_trim() {
        // Regression (review G1): the budget check used to re-fold cost over
        // the retention-trimmed log, so trimming metered events slid the
        // counter backward and un-fired the alert. The cumulative counter is
        // monotonic — the alert must still read over-budget after a trim.
        let mut log = crate::EventLog::new();
        log.set_retention(Some(crate::RetentionPolicy {
            max_events: Some(1),
            max_age_secs: None,
        }));
        for _ in 0..3 {
            log.append_metered(
                EventKind::InferenceMetered,
                None,
                None,
                Default::default(),
                Metrics {
                    cost_usd: Some(6.0),
                    ..Default::default()
                },
            );
        }
        // Retention kept only the last metered event: the windowed fold sees
        // $6, under the $10 budget…
        assert_eq!(log.events().len(), 1);
        let s = summarize_log(&log);
        assert!((s.cost_usd - 6.0).abs() < 1e-9, "windowed fold trimmed");
        // …but the cumulative counter still carries the full $18 spend.
        assert!((s.cumulative_cost_usd - 18.0).abs() < 1e-9);
        let alerts = evaluate_alerts(
            &s,
            &AlertThresholds {
                max_cost_usd: Some(10.0),
                ..Default::default()
            },
        );
        assert_eq!(alerts.len(), 1, "budget alert must not un-fire on trim");
        assert_eq!(alerts[0].kind, AlertKind::CostOverage);
        assert_eq!(alerts[0].observed, 18.0);
    }

    #[test]
    fn ungrounded_goal_alert_fires() {
        let s = summarize(&[goal_ev(false, true), goal_ev(true, false)]);
        assert_eq!(s.goal_evaluations, 2);
        assert_eq!(s.goals_ungrounded, 1);

        let alerts = evaluate_alerts(
            &s,
            &AlertThresholds {
                max_goals_ungrounded: Some(0),
                ..Default::default()
            },
        );
        assert_eq!(alerts.len(), 1);
        assert_eq!(alerts[0].kind, AlertKind::GoalUngrounded);
        assert_eq!(alerts[0].observed, 1.0);
        assert_eq!(alerts[0].threshold, 0.0);
    }

    #[test]
    fn error_rate_alert_suppressed_below_min_actions() {
        // One failure only — below the default min_actions (5), so no alert
        // despite a 100% error rate.
        let s = summarize(&[ev(EventKind::ActionFailed)]);
        assert!((s.error_rate - 1.0).abs() < 1e-9);
        let alerts = evaluate_alerts(
            &s,
            &AlertThresholds {
                max_error_rate: Some(0.5),
                ..Default::default()
            },
        );
        assert!(alerts.is_empty(), "should not fire under min_actions");
    }

    #[test]
    fn no_thresholds_never_fires() {
        let s = summarize(&[ev(EventKind::ActionFailed)]);
        assert!(evaluate_alerts(&s, &AlertThresholds::default()).is_empty());
    }
}