use crate::session::{ActivityState, Session, SubagentStatus};
use crate::util;
use chrono::{DateTime, Utc};
use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant};
pub const REARM: f64 = 0.8;
pub const ERROR_WINDOW: Duration = Duration::from_secs(10 * 60);
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Rules {
pub session_cost: f64,
pub burn_rate: f64,
pub daily_spend: f64,
pub error_rate: f64,
pub error_calls: u64,
pub stall: Duration,
}
impl Default for Rules {
fn default() -> Self {
Rules {
session_cost: 0.0,
burn_rate: 0.0,
daily_spend: 0.0,
error_rate: 0.0,
error_calls: 10,
stall: Duration::ZERO,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Kind {
Cost,
Burn,
Stall,
Errors,
}
impl Kind {
const ALL: [Kind; 4] = [Kind::Cost, Kind::Burn, Kind::Stall, Kind::Errors];
pub fn glyph(self) -> &'static str {
match self {
Kind::Cost | Kind::Burn => "$",
Kind::Errors => "!",
Kind::Stall => "◌",
}
}
fn name(self) -> &'static str {
match self {
Kind::Cost => "cost",
Kind::Burn => "burn",
Kind::Stall => "stall",
Kind::Errors => "errors",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Hooked {
Unknown,
Working,
InFlight,
Quiet,
}
#[derive(Debug, Clone, PartialEq)]
pub struct Fired {
pub key: Option<String>,
pub kind: Option<Kind>,
pub text: String,
}
impl Fired {
pub fn kind_name(&self) -> &'static str {
self.kind.map_or("today", Kind::name)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Level {
Above,
Between,
Clear,
}
fn level(value: f64, threshold: f64) -> Level {
if threshold <= 0.0 {
Level::Clear
} else if value >= threshold {
Level::Above
} else if value < threshold * REARM {
Level::Clear
} else {
Level::Between
}
}
#[derive(Debug, Clone, Copy, Default)]
struct Latch {
high: bool,
seen: bool,
}
impl Latch {
fn step(&mut self, level: Level) -> bool {
if !self.seen {
self.seen = true;
self.high = level == Level::Above;
return false;
}
match level {
Level::Above if !self.high => {
self.high = true;
true
}
Level::Clear => {
self.high = false;
false
}
_ => false,
}
}
}
#[derive(Debug, Default)]
struct Watch {
latches: [Latch; 4],
samples: VecDeque<(Instant, u64, u64)>,
}
impl Watch {
fn window(&mut self, now: Instant, calls: u64, errors: u64) -> (u64, u64) {
if self
.samples
.back()
.is_some_and(|&(_, c, e)| calls < c || errors < e)
{
self.samples.clear();
}
if self
.samples
.back()
.is_none_or(|&(_, c, e)| (c, e) != (calls, errors))
{
self.samples.push_back((now, calls, errors));
}
if let Some(start) = now.checked_sub(ERROR_WINDOW) {
while self.samples.len() > 1 && self.samples[1].0 <= start {
self.samples.pop_front();
}
}
let (_, c0, e0) = self.samples[0];
(calls - c0, errors - e0)
}
}
#[derive(Debug, Default)]
pub struct Alerts {
watched: HashMap<String, Watch>,
today: Latch,
rules: Option<Rules>,
}
impl Alerts {
pub fn observe(
&mut self,
rules: &Rules,
sessions: &[Session],
spend_today: f64,
hooked: impl Fn(&Session) -> Hooked,
now: Instant,
wall: DateTime<Utc>,
) -> Vec<Fired> {
if self.rules.as_ref() != Some(rules) {
self.rules = Some(*rules);
self.today.seen = false;
for watch in self.watched.values_mut() {
watch.latches = Default::default();
}
}
let mut fired = Vec::new();
let mut next = HashMap::with_capacity(self.watched.len());
for session in sessions.iter().filter(|s| s.is_running()) {
let key = session.key();
let mut watch = self.watched.remove(&key).unwrap_or_default();
let label = session.display_label();
let (calls, errors) = watch.window(now, session.tool_count, session.tool_errors);
for kind in Kind::ALL {
let (level, text) = reading(kind, rules, session, (calls, errors), &hooked, wall);
if watch.latches[kind as usize].step(level) {
fired.push(Fired {
key: Some(key.clone()),
kind: Some(kind),
text: format!("{label} {text}"),
});
}
}
next.insert(key, watch);
}
self.watched = next;
if self.today.step(level(spend_today, rules.daily_spend)) {
fired.push(Fired {
key: None,
kind: None,
text: format!(
"Today's spend is {}, past the {} alert",
util::compact_usd(spend_today),
util::compact_usd(rules.daily_spend)
),
});
}
fired
}
pub fn marker(&self, key: &str) -> Option<Kind> {
let watch = self.watched.get(key)?;
Kind::ALL
.into_iter()
.rev()
.find(|&k| watch.latches[k as usize].high)
}
}
fn reading(
kind: Kind,
rules: &Rules,
s: &Session,
(calls, errors): (u64, u64),
hooked: &impl Fn(&Session) -> Hooked,
wall: DateTime<Utc>,
) -> (Level, String) {
match kind {
Kind::Cost => match s.total_cost {
Some(cost) => (
level(cost, rules.session_cost),
format!(
"has cost {}, past the {} alert",
util::compact_usd(cost),
util::compact_usd(rules.session_cost)
),
),
None => (Level::Clear, String::new()),
},
Kind::Burn => match s.total_cost {
Some(_) => {
let per_hour = s.cost_per_min * 60.0;
(
level(per_hour, rules.burn_rate),
format!(
"is burning {}/h, past the {}/h alert",
util::compact_usd(per_hour),
util::compact_usd(rules.burn_rate)
),
)
}
None => (Level::Clear, String::new()),
},
Kind::Errors if rules.error_rate <= 0.0 => (Level::Clear, String::new()),
Kind::Errors => {
let rate = match calls {
0 => 0.0,
n => errors as f64 / n as f64,
};
let level = match level(rate, rules.error_rate) {
Level::Above if calls < rules.error_calls.max(1) => Level::Between,
Level::Clear if calls > 0 && rate >= rules.error_rate * REARM => Level::Between,
other => other,
};
(
level,
format!(
"is looping: {errors} of its last {calls} tool calls failed ({}m)",
ERROR_WINDOW.as_secs() / 60
),
)
}
Kind::Stall => stall(rules.stall, s, hooked(s), wall),
}
}
fn stall(threshold: Duration, s: &Session, hooked: Hooked, wall: DateTime<Utc>) -> (Level, String) {
if threshold.is_zero() || s.activity_state != ActivityState::Working {
return (Level::Clear, String::new());
}
let working = match hooked {
Hooked::Quiet => return (Level::Clear, String::new()),
Hooked::Unknown | Hooked::InFlight => false,
Hooked::Working => true,
};
let newest = std::iter::once(s.last_active.as_str())
.chain(
s.subagents
.iter()
.filter(|a| a.status == SubagentStatus::Running)
.filter_map(|a| a.last_active.as_deref()),
)
.filter_map(util::parse_ts)
.max();
let Some(newest) = newest else {
return (Level::Between, String::new());
};
let silent = (wall - newest).to_std().unwrap_or_default();
let level = match (silent >= threshold, working) {
(false, _) => Level::Clear,
(true, true) => Level::Above,
(true, false) => Level::Between,
};
(
level,
format!(
"has written nothing for {}m while working",
silent.as_secs() / 60
),
)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::pricing::Provider;
fn session(id: &str) -> Session {
let mut s = Session::new(Provider::Claude, id.into());
s.abbrev_label = id.into();
s.process = Some(crate::proc::ProcInfo::default());
s.total_cost = Some(0.0);
s.last_active = chrono::Utc::now().to_rfc3339();
s
}
struct Run {
alerts: Alerts,
rules: Rules,
start: Instant,
wall: DateTime<Utc>,
tick: u32,
}
impl Run {
fn new(rules: Rules) -> Run {
Run {
alerts: Alerts::default(),
rules,
start: Instant::now(),
wall: chrono::Utc::now(),
tick: 0,
}
}
fn step(&mut self, sessions: &[Session], today: f64, hooked: Hooked) -> Vec<Fired> {
self.tick += 1;
let at = Duration::from_secs(60 * self.tick as u64);
self.alerts.observe(
&self.rules,
sessions,
today,
|_| hooked,
self.start + at,
self.wall + chrono::Duration::from_std(at).unwrap(),
)
}
}
fn costing(dollars: f64) -> Vec<Session> {
let mut s = session("a");
s.total_cost = Some(dollars);
vec![s]
}
#[test]
fn a_threshold_fires_once_per_crossing_with_hysteresis() {
let mut run = Run::new(Rules {
session_cost: 10.0,
..Rules::default()
});
assert!(run.step(&costing(5.0), 0.0, Hooked::Unknown).is_empty());
let fired = run.step(&costing(10.5), 0.0, Hooked::Unknown);
assert_eq!(fired.len(), 1, "{fired:?}");
assert_eq!(fired[0].kind, Some(Kind::Cost));
assert!(fired[0].text.contains("$10.50"), "{}", fired[0].text);
assert!(run.step(&costing(11.0), 0.0, Hooked::Unknown).is_empty());
assert!(run.step(&costing(9.5), 0.0, Hooked::Unknown).is_empty());
assert!(run.step(&costing(10.1), 0.0, Hooked::Unknown).is_empty());
assert_eq!(run.alerts.marker("claude:a"), Some(Kind::Cost));
assert!(run.step(&costing(7.0), 0.0, Hooked::Unknown).is_empty());
assert_eq!(run.alerts.marker("claude:a"), None);
assert_eq!(run.step(&costing(12.0), 0.0, Hooked::Unknown).len(), 1);
}
#[test]
fn a_session_first_seen_over_budget_is_marked_not_rung() {
let mut run = Run::new(Rules {
session_cost: 10.0,
..Rules::default()
});
assert!(run.step(&costing(20.0), 0.0, Hooked::Unknown).is_empty());
assert_eq!(run.alerts.marker("claude:a"), Some(Kind::Cost));
run.rules.session_cost = 15.0;
assert!(run.step(&costing(20.0), 0.0, Hooked::Unknown).is_empty());
assert_eq!(run.alerts.marker("claude:a"), Some(Kind::Cost));
}
#[test]
fn a_rule_left_at_zero_never_fires() {
let mut run = Run::new(Rules::default());
let mut s = costing(1_000.0);
s[0].cost_per_min = 100.0;
s[0].tool_count = 50;
run.step(&s, 5_000.0, Hooked::Working);
s[0].tool_count = 100;
s[0].tool_errors = 50;
s[0].last_active = "2020-01-01T00:00:00Z".into();
assert!(run.step(&s, 9_000.0, Hooked::Working).is_empty());
assert_eq!(run.alerts.marker("claude:a"), None);
}
#[test]
fn a_burn_rate_is_judged_per_hour_and_an_included_plan_is_exempt() {
let mut run = Run::new(Rules {
burn_rate: 30.0,
..Rules::default()
});
let mut s = costing(1.0);
run.step(&s, 0.0, Hooked::Unknown);
s[0].cost_per_min = 0.6; let fired = run.step(&s, 0.0, Hooked::Unknown);
assert_eq!(fired.len(), 1);
assert!(fired[0].text.contains("$36.00/h"), "{}", fired[0].text);
let mut run = Run::new(run.rules);
let mut s = costing(1.0);
s[0].total_cost = None;
run.step(&s, 0.0, Hooked::Unknown);
s[0].cost_per_min = 5.0;
assert!(run.step(&s, 0.0, Hooked::Unknown).is_empty());
}
#[test]
fn the_days_spend_fires_once_and_again_the_next_day() {
let mut run = Run::new(Rules {
daily_spend: 50.0,
..Rules::default()
});
assert!(run.step(&[], 40.0, Hooked::Unknown).is_empty());
let fired = run.step(&[], 55.0, Hooked::Unknown);
assert_eq!(fired.len(), 1);
assert_eq!(
(fired[0].key.as_deref(), fired[0].kind_name()),
(None, "today")
);
assert!(run.step(&[], 60.0, Hooked::Unknown).is_empty());
assert!(run.step(&[], 0.0, Hooked::Unknown).is_empty());
assert_eq!(run.step(&[], 51.0, Hooked::Unknown).len(), 1);
}
fn calls(total: u64, failed: u64) -> Vec<Session> {
let mut s = session("a");
s.tool_count = total;
s.tool_errors = failed;
vec![s]
}
#[test]
fn an_error_loop_is_measured_over_the_recent_window() {
let mut run = Run::new(Rules {
error_rate: 0.25,
error_calls: 10,
..Rules::default()
});
run.step(&calls(200, 2), 0.0, Hooked::Unknown);
assert!(run.step(&calls(206, 7), 0.0, Hooked::Unknown).is_empty());
let fired = run.step(&calls(212, 10), 0.0, Hooked::Unknown);
assert_eq!(fired.len(), 1, "8 of 12 failing is a loop");
assert!(
fired[0].text.contains("8 of its last 12"),
"{}",
fired[0].text
);
assert!(
fired[0].text.contains(" failed"),
"worded as a failure, so the toast reads as one"
);
assert_eq!(run.alerts.marker("claude:a"), Some(Kind::Errors));
assert!(run.step(&calls(220, 14), 0.0, Hooked::Unknown).is_empty());
let mut total = 220;
for _ in 0..11 {
total += 5;
run.step(&calls(total, 14), 0.0, Hooked::Unknown);
}
assert_eq!(run.alerts.marker("claude:a"), None);
}
#[test]
fn a_counter_that_goes_backwards_starts_the_window_over() {
let mut w = Watch::default();
let t = Instant::now();
assert_eq!(w.window(t, 100, 10), (0, 0));
assert_eq!(w.window(t + Duration::from_secs(60), 110, 15), (10, 5));
assert_eq!(w.window(t + Duration::from_secs(120), 5, 1), (0, 0));
assert_eq!(
w.window(t + Duration::from_secs(120) + ERROR_WINDOW, 9, 1),
(4, 0)
);
}
fn silent_for(minutes: i64) -> Vec<Session> {
let mut s = session("a");
s.last_active = (chrono::Utc::now() - chrono::Duration::minutes(minutes)).to_rfc3339();
vec![s]
}
#[test]
fn a_stall_needs_the_hooks_to_say_working() {
let rules = Rules {
stall: Duration::from_secs(10 * 60),
..Rules::default()
};
for quiet in [Hooked::Unknown, Hooked::InFlight, Hooked::Quiet] {
let mut run = Run::new(rules);
run.step(&silent_for(0), 0.0, quiet);
assert!(
run.step(&silent_for(30), 0.0, quiet).is_empty(),
"{quiet:?}"
);
}
let mut run = Run::new(rules);
assert!(run.step(&silent_for(0), 0.0, Hooked::Working).is_empty());
assert!(run.step(&silent_for(5), 0.0, Hooked::Working).is_empty());
let fired = run.step(&silent_for(12), 0.0, Hooked::Working);
assert_eq!(fired.len(), 1);
assert_eq!(fired[0].kind, Some(Kind::Stall));
assert!(run.step(&silent_for(20), 0.0, Hooked::Working).is_empty());
assert_eq!(run.alerts.marker("claude:a"), Some(Kind::Stall));
assert!(run.step(&silent_for(25), 0.0, Hooked::InFlight).is_empty());
assert_eq!(run.alerts.marker("claude:a"), Some(Kind::Stall));
run.step(&silent_for(-10), 0.0, Hooked::Working);
assert_eq!(run.alerts.marker("claude:a"), None);
}
#[test]
fn a_running_subagent_counts_as_the_session_writing() {
let mut run = Run::new(Rules {
stall: Duration::from_secs(10 * 60),
..Rules::default()
});
let mut s = silent_for(30);
let now = (chrono::Utc::now() + chrono::Duration::minutes(5)).to_rfc3339();
s[0].subagents.push(crate::session::Subagent {
agent_id: "sub".into(),
agent_type: "general-purpose".into(),
description: String::new(),
model: String::new(),
started_at: None,
last_active: Some(now),
duration_ms: 0,
status: SubagentStatus::Running,
cost: 0.0,
tool_count: 0,
tool_use_id: None,
context: None,
ghost: false,
});
run.step(&s, 0.0, Hooked::Working);
assert!(run.step(&s, 0.0, Hooked::Working).is_empty());
assert_eq!(run.alerts.marker("claude:a"), None);
}
#[test]
fn the_marker_is_the_loudest_and_leaves_with_the_session() {
let mut run = Run::new(Rules {
session_cost: 1.0,
error_rate: 0.25,
error_calls: 2,
..Rules::default()
});
let mut s = calls(0, 0);
s[0].total_cost = Some(2.0);
run.step(&s, 0.0, Hooked::Unknown);
assert_eq!(run.alerts.marker("claude:a"), Some(Kind::Cost));
s[0].tool_count = 4;
s[0].tool_errors = 4;
assert_eq!(run.step(&s, 0.0, Hooked::Unknown).len(), 1);
assert_eq!(run.alerts.marker("claude:a"), Some(Kind::Errors));
s[0].process = None;
run.step(&s, 0.0, Hooked::Unknown);
assert_eq!(run.alerts.marker("claude:a"), None);
}
}