Add alert rules with FilterBar-based triggers and ntfy delivery (tickets 017-019)

Adds AlertRule/AlertCondition/AlertSink models with CRUD persisted on
Settings (mirroring CssEntry's shape), per-line trigger evaluation hooked
into the engine's ingest path (Immediate/Threshold/NewValue/Ratio
conditions), and best-effort ntfy delivery spawned off the ingest hot path.

Design notes:
- Alert queries reuse FilterBar syntax but only support the
  regex()/sequence/timestamp subset (no metric[param] comparisons), since
  resolving those would require a full bucket rescan per alert per ingested
  line rather than cost proportional to new logs.
- Ratio's zero-denominator case skips evaluation for that line rather than
  treating the ratio as infinite.
- NewValue's seen-value set is in-memory only and does not survive a
  restart, matching the engine's other non-persisted runtime state.

engine::run now takes a SettingsHandle (moved from api.rs to settings.rs so
engine.rs can depend on it without a circular import) to read the current
alert rules on each ingest.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o

src/alerts.rs

@@ -0,0 +1,654 @@
+//! Alert rules: definitions (ticket 017), trigger evaluation (ticket 018),
+//! and ntfy delivery (ticket 019).
+//!
+//! ## Design: why alert queries only support regex/sequence/timestamp
+//!
+//! `Immediate`/`Threshold`/`Ratio` conditions reuse FilterBar query strings
+//! (see `crate::filter_parser`), evaluated against a *single newly-ingested
+//! line* on `EngineMessage::Ingest` (see `evaluate_alerts`, called from
+//! `engine::run`). `engine.rs`'s own `compile_filter_query` resolves
+//! `metric[param]` comparisons by scanning that metric's *entire* bucket
+//! history up front (`compile_metric_comparison`) — reusing that per alert
+//! check, per ingested line, would make alert evaluation cost proportional to
+//! total log volume rather than to new logs ingested, which this ticket
+//! requires. So alert queries here only support the subset of FilterBar that
+//! evaluates against a single line's own fields — `regex(...)`, `sequence`,
+//! `timestamp`, and `and`/`or` — via `eval_alert_expr` below. A `metric[param]`
+//! comparison in an alert query is rejected with a clear error (logged, the
+//! rule just never fires) rather than silently misbehaving.
+
+use std::collections::hash_map::DefaultHasher;
+use std::collections::{HashMap, HashSet, VecDeque};
+use std::fmt;
+use std::hash::{Hash, Hasher};
+use std::io;
+use std::path::Path;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use regex::Regex;
+use serde::{Deserialize, Serialize};
+
+use crate::engine::apply_compare_op;
+use crate::filter_datetime;
+use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
+use crate::settings::Settings;
+use crate::timestamp::Timestamp;
+
+/// A single alert rule definition (ticket 017).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct AlertRule {
+    pub id: String,
+    pub name: String,
+    pub enabled: bool,
+    pub condition: AlertCondition,
+    pub sink: AlertSink,
+}
+
+/// What triggers an alert rule (ticket 018).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "lowercase")]
+pub enum AlertCondition {
+    /// Fires once per log line matching `query`.
+    Immediate { query: String },
+    /// Fires when `query`'s match count within a rolling `window_ms` crosses
+    /// `threshold` (via `op`).
+    Threshold {
+        query: String,
+        op: CompareOp,
+        threshold: f64,
+        window_ms: i64,
+    },
+    /// Fires the first time `field_regex` matches a line and `capture_group`
+    /// captures a value not seen before *for this rule*. Seen values live
+    /// only in memory (`AlertRuntimeState::seen_values`) for the process's
+    /// lifetime — a restart forgets them, so previously-seen values can fire
+    /// again. This is a deliberate simplicity tradeoff (matches the
+    /// codebase's general "no restart persistence for in-memory engine
+    /// state" pattern), not an oversight.
+    NewValue {
+        field_regex: String,
+        capture_group: String,
+    },
+    /// Fires when `count(query_a) / count(query_b)` (rolling over
+    /// `window_ms`) crosses `ratio` (via `op`). A zero `count_b` skips
+    /// evaluation entirely for that line (rather than treating the ratio as
+    /// infinite) — there's no meaningful "actual vs threshold" comparison to
+    /// report yet.
+    Ratio {
+        query_a: String,
+        query_b: String,
+        op: CompareOp,
+        ratio: f64,
+        window_ms: i64,
+    },
+}
+
+/// Where a fired alert is delivered (ticket 019). Modeled as an enum (rather
+/// than `AlertRule` holding a bare topic/server) so more sink kinds can be
+/// added later without reshaping `AlertRule`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "lowercase")]
+pub enum AlertSink {
+    Ntfy {
+        topic: String,
+        /// Defaults to `https://ntfy.sh` when absent.
+        server: Option<String>,
+    },
+}
+
+/// Error returned by alert-rule mutations/lookups. Mirrors `CssError`'s shape.
+#[derive(Debug)]
+pub enum AlertError {
+    NotFound,
+    Io(io::Error),
+}
+
+impl fmt::Display for AlertError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            AlertError::NotFound => write!(f, "no alert rule with that id"),
+            AlertError::Io(err) => write!(f, "I/O error: {err}"),
+        }
+    }
+}
+
+impl std::error::Error for AlertError {}
+
+impl From<io::Error> for AlertError {
+    fn from(err: io::Error) -> Self {
+        AlertError::Io(err)
+    }
+}
+
+/// Generates an id for a newly-created alert rule. Same approach as
+/// `settings::generate_imported_css_id`: hash the name and current time
+/// rather than pulling in a `uuid` dependency.
+fn generate_alert_rule_id(name: &str) -> String {
+    let mut hasher = DefaultHasher::new();
+    name.hash(&mut hasher);
+    let nanos = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_nanos())
+        .unwrap_or(0);
+    nanos.hash(&mut hasher);
+    format!("alert-{:016x}", hasher.finish())
+}
+
+impl Settings {
+    /// Creates a new alert rule, persisting settings to `settings_path`.
+    pub fn create_alert_rule(
+        &mut self,
+        settings_path: &Path,
+        name: String,
+        enabled: bool,
+        condition: AlertCondition,
+        sink: AlertSink,
+    ) -> Result<AlertRule, AlertError> {
+        let rule = AlertRule { id: generate_alert_rule_id(&name), name, enabled, condition, sink };
+        self.alert_rules.push(rule.clone());
+        self.save(settings_path)?;
+        Ok(rule)
+    }
+
+    /// Replaces an existing alert rule's fields (id stays fixed), persisting
+    /// settings to `settings_path`. Returns `AlertError::NotFound` if no rule
+    /// with `id` exists.
+    pub fn update_alert_rule(
+        &mut self,
+        settings_path: &Path,
+        id: &str,
+        name: String,
+        enabled: bool,
+        condition: AlertCondition,
+        sink: AlertSink,
+    ) -> Result<AlertRule, AlertError> {
+        let rule = self
+            .alert_rules
+            .iter_mut()
+            .find(|rule| rule.id == id)
+            .ok_or(AlertError::NotFound)?;
+        rule.name = name;
+        rule.enabled = enabled;
+        rule.condition = condition;
+        rule.sink = sink;
+        let updated = rule.clone();
+        self.save(settings_path)?;
+        Ok(updated)
+    }
+
+    /// Removes an alert rule by id, persisting settings to `settings_path`.
+    /// Returns `AlertError::NotFound` if no rule with `id` exists.
+    pub fn remove_alert_rule(&mut self, settings_path: &Path, id: &str) -> Result<(), AlertError> {
+        let index = self
+            .alert_rules
+            .iter()
+            .position(|rule| rule.id == id)
+            .ok_or(AlertError::NotFound)?;
+        self.alert_rules.remove(index);
+        self.save(settings_path)?;
+        Ok(())
+    }
+}
+
+/// Per-rule in-memory evaluation state (ticket 018): rolling match-timestamp
+/// windows for `Threshold`/`Ratio`, and seen-values for `NewValue`. Lives
+/// only for the engine process's lifetime (see `AlertCondition::NewValue`'s
+/// doc comment) — not persisted, not backfilled against already-ingested
+/// logs.
+#[derive(Debug, Default)]
+pub struct AlertRuntimeState {
+    threshold_matches: VecDeque<i64>,
+    ratio_a_matches: VecDeque<i64>,
+    ratio_b_matches: VecDeque<i64>,
+    seen_values: HashSet<String>,
+}
+
+/// Drops every entry older than `window_ms` relative to `now_ms` from the
+/// front of a rolling window. `timestamps` is push_back-ordered, so once a
+/// front entry is within the window every entry behind it is too.
+fn prune_window(timestamps: &mut VecDeque<i64>, now_ms: i64, window_ms: i64) {
+    while let Some(&front) = timestamps.front() {
+        if now_ms - front > window_ms {
+            timestamps.pop_front();
+        } else {
+            break;
+        }
+    }
+}
+
+/// Evaluates a FilterBar query string against a single log line's own fields
+/// only — see this module's doc comment for why `metric[param]` comparisons
+/// aren't supported here.
+fn eval_alert_query(query: &str, sequence: u64, timestamp: Timestamp, text: &str) -> Result<bool, String> {
+    let expr = filter_parser::parse(query).map_err(|err| err.to_string())?;
+    eval_alert_expr(&expr, sequence, timestamp, text)
+}
+
+fn eval_alert_expr(expr: &FilterExpr, sequence: u64, timestamp: Timestamp, text: &str) -> Result<bool, String> {
+    match expr {
+        FilterExpr::Regex(pattern) => {
+            let re = Regex::new(pattern).map_err(|err| format!("invalid regex in alert query: {err}"))?;
+            Ok(re.is_match(text))
+        }
+        FilterExpr::Comparison { metric, param: None, op, value } if metric == "sequence" => {
+            let target = match value {
+                FilterValue::Number(n) => *n,
+                FilterValue::Raw(raw) => raw
+                    .parse::<f64>()
+                    .map_err(|_| format!("'sequence' comparisons require a numeric value, got {raw:?}"))?,
+            };
+            Ok(apply_compare_op((sequence as f64).total_cmp(&target), *op))
+        }
+        FilterExpr::Comparison { metric, param: None, op, value } if metric == "timestamp" => {
+            let target_ms = match value {
+                FilterValue::Number(n) => *n as i64,
+                FilterValue::Raw(raw) => filter_datetime::parse_datetime(raw, Timestamp::now())
+                    .map_err(|err| err.to_string())?
+                    .as_millis(),
+            };
+            Ok(apply_compare_op(timestamp.as_millis().cmp(&target_ms), *op))
+        }
+        FilterExpr::Comparison { metric, .. } => Err(format!(
+            "alert queries only support regex()/sequence/timestamp (not metric comparisons like \
+             '{metric}[param]'): metric-based alert conditions aren't supported yet"
+        )),
+        FilterExpr::And(lhs, rhs) => {
+            Ok(eval_alert_expr(lhs, sequence, timestamp, text)? && eval_alert_expr(rhs, sequence, timestamp, text)?)
+        }
+        FilterExpr::Or(lhs, rhs) => {
+            Ok(eval_alert_expr(lhs, sequence, timestamp, text)? || eval_alert_expr(rhs, sequence, timestamp, text)?)
+        }
+    }
+}
+
+/// Evaluates every enabled rule's condition against one newly-ingested line,
+/// mutating each rule's runtime state (rolling windows / seen-values) along
+/// the way. Returns `(rule, message)` for every rule that fired — the caller
+/// (`engine::run`) hands each off to `deliver_for_rule`. A malformed query
+/// (bad regex, unsupported metric comparison) is logged to stderr and simply
+/// never matches, rather than panicking or propagating.
+pub fn evaluate_alerts(
+    rules: &[AlertRule],
+    states: &mut HashMap<String, AlertRuntimeState>,
+    sequence: u64,
+    timestamp: Timestamp,
+    text: &str,
+) -> Vec<(AlertRule, String)> {
+    let mut fired = Vec::new();
+    for rule in rules {
+        if !rule.enabled {
+            continue;
+        }
+        let state = states.entry(rule.id.clone()).or_default();
+        if let Some(message) = evaluate_condition(&rule.condition, state, sequence, timestamp, text, &rule.name) {
+            fired.push((rule.clone(), message));
+        }
+    }
+    fired
+}
+
+fn evaluate_condition(
+    condition: &AlertCondition,
+    state: &mut AlertRuntimeState,
+    sequence: u64,
+    timestamp: Timestamp,
+    text: &str,
+    rule_name: &str,
+) -> Option<String> {
+    match condition {
+        AlertCondition::Immediate { query } => match eval_alert_query(query, sequence, timestamp, text) {
+            Ok(true) => Some(format!("{rule_name}: matched line: {text}")),
+            Ok(false) => None,
+            Err(err) => {
+                eprintln!("warning: alert '{rule_name}' query error: {err}");
+                None
+            }
+        },
+        AlertCondition::Threshold { query, op, threshold, window_ms } => {
+            match eval_alert_query(query, sequence, timestamp, text) {
+                Ok(true) => {
+                    let now_ms = timestamp.as_millis();
+                    state.threshold_matches.push_back(now_ms);
+                    prune_window(&mut state.threshold_matches, now_ms, *window_ms);
+                    let count = state.threshold_matches.len() as f64;
+                    apply_compare_op(count.total_cmp(threshold), *op).then(|| {
+                        format!("{rule_name}: match count {count} {op:?} threshold {threshold} (window {window_ms}ms)")
+                    })
+                }
+                Ok(false) => None,
+                Err(err) => {
+                    eprintln!("warning: alert '{rule_name}' query error: {err}");
+                    None
+                }
+            }
+        }
+        AlertCondition::NewValue { field_regex, capture_group } => {
+            let re = match Regex::new(field_regex) {
+                Ok(re) => re,
+                Err(err) => {
+                    eprintln!("warning: alert '{rule_name}' has an invalid field_regex: {err}");
+                    return None;
+                }
+            };
+            let captures = re.captures(text)?;
+            let value = captures.name(capture_group)?.as_str().to_string();
+            state
+                .seen_values
+                .insert(value.clone())
+                .then(|| format!("{rule_name}: new value seen: {value} (line: {text})"))
+        }
+        AlertCondition::Ratio { query_a, query_b, op, ratio, window_ms } => {
+            let matched_a = eval_alert_query(query_a, sequence, timestamp, text);
+            let matched_b = eval_alert_query(query_b, sequence, timestamp, text);
+            let (matched_a, matched_b) = match (matched_a, matched_b) {
+                (Ok(a), Ok(b)) => (a, b),
+                (Err(err), _) | (_, Err(err)) => {
+                    eprintln!("warning: alert '{rule_name}' query error: {err}");
+                    return None;
+                }
+            };
+            if !matched_a && !matched_b {
+                return None;
+            }
+            let now_ms = timestamp.as_millis();
+            if matched_a {
+                state.ratio_a_matches.push_back(now_ms);
+                prune_window(&mut state.ratio_a_matches, now_ms, *window_ms);
+            }
+            if matched_b {
+                state.ratio_b_matches.push_back(now_ms);
+                prune_window(&mut state.ratio_b_matches, now_ms, *window_ms);
+            }
+            let count_a = state.ratio_a_matches.len() as f64;
+            let count_b = state.ratio_b_matches.len() as f64;
+            if count_b == 0.0 {
+                // Zero denominator: skip evaluation rather than treat the
+                // ratio as infinite — there's nothing meaningful to compare
+                // against `ratio` yet.
+                return None;
+            }
+            let observed = count_a / count_b;
+            apply_compare_op(observed.total_cmp(ratio), *op).then(|| {
+                format!(
+                    "{rule_name}: ratio {observed:.3} ({count_a}/{count_b}) {op:?} {ratio} (window {window_ms}ms)"
+                )
+            })
+        }
+    }
+}
+
+/// Delivers one fired alert to its rule's configured sink. Errors are logged
+/// to stderr and dropped — no retry queue, matching `remove_css`'s
+/// best-effort style. Called via `tokio::spawn` from the ingest hot path
+/// (`engine::run`) so a slow/unreachable sink never blocks ingestion.
+pub async fn deliver_for_rule(rule: AlertRule, message: String) {
+    match rule.sink {
+        AlertSink::Ntfy { topic, server } => {
+            let server = server.unwrap_or_else(|| "https://ntfy.sh".to_string());
+            if let Err(err) = deliver_ntfy(&server, &topic, &message).await {
+                eprintln!("warning: failed to deliver alert '{}' via ntfy: {err}", rule.name);
+            }
+        }
+    }
+}
+
+/// Publishes `message` to `{server}/{topic}` via ntfy's plain-text publish
+/// API (no auth, raw body). Standalone and `AlertRule`-free so a future
+/// "send test notification" endpoint can call it directly.
+pub async fn deliver_ntfy(server: &str, topic: &str, message: &str) -> Result<(), String> {
+    let url = format!("{}/{}", server.trim_end_matches('/'), topic);
+    let response = reqwest::Client::new()
+        .post(&url)
+        .body(message.to_string())
+        .send()
+        .await
+        .map_err(|err| err.to_string())?;
+    if response.status().is_success() {
+        Ok(())
+    } else {
+        Err(format!("ntfy responded with status {}", response.status()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn line(text: &str, sequence: u64, timestamp_ms: i64) -> (u64, Timestamp, &str) {
+        (sequence, Timestamp::from_millis(timestamp_ms), text)
+    }
+
+    fn immediate_rule(query: &str) -> AlertRule {
+        AlertRule {
+            id: "r1".to_string(),
+            name: "test rule".to_string(),
+            enabled: true,
+            condition: AlertCondition::Immediate { query: query.to_string() },
+            sink: AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
+        }
+    }
+
+    #[test]
+    fn alert_rule_crud_round_trips_through_save_and_load() {
+        let dir = tempfile::tempdir().unwrap();
+        let settings_path = dir.path().join("settings.json");
+        let mut settings = Settings::load(&settings_path);
+
+        let created = settings
+            .create_alert_rule(
+                &settings_path,
+                "spike".to_string(),
+                true,
+                AlertCondition::Immediate { query: r#"regex("error")"#.to_string() },
+                AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
+            )
+            .unwrap();
+        assert_eq!(settings.alert_rules.len(), 1);
+
+        let reloaded = Settings::load(&settings_path);
+        assert_eq!(reloaded.alert_rules, settings.alert_rules);
+
+        let updated = settings
+            .update_alert_rule(
+                &settings_path,
+                &created.id,
+                "renamed".to_string(),
+                false,
+                AlertCondition::Immediate { query: r#"regex("warn")"#.to_string() },
+                AlertSink::Ntfy { topic: "other".to_string(), server: Some("https://example.com".to_string()) },
+            )
+            .unwrap();
+        assert_eq!(updated.id, created.id);
+        assert_eq!(updated.name, "renamed");
+        assert!(!updated.enabled);
+
+        settings.remove_alert_rule(&settings_path, &created.id).unwrap();
+        assert!(settings.alert_rules.is_empty());
+        let reloaded = Settings::load(&settings_path);
+        assert!(reloaded.alert_rules.is_empty());
+    }
+
+    #[test]
+    fn alert_rule_mutations_reject_unknown_id() {
+        let dir = tempfile::tempdir().unwrap();
+        let settings_path = dir.path().join("settings.json");
+        let mut settings = Settings::load(&settings_path);
+
+        let result = settings.update_alert_rule(
+            &settings_path,
+            "does-not-exist",
+            "x".to_string(),
+            true,
+            AlertCondition::Immediate { query: r#"regex("x")"#.to_string() },
+            AlertSink::Ntfy { topic: "x".to_string(), server: None },
+        );
+        assert!(matches!(result, Err(AlertError::NotFound)));
+
+        let result = settings.remove_alert_rule(&settings_path, "does-not-exist");
+        assert!(matches!(result, Err(AlertError::NotFound)));
+    }
+
+    #[test]
+    fn immediate_fires_on_matching_line_only() {
+        let rules = vec![immediate_rule(r#"regex("error")"#)];
+        let mut states = HashMap::new();
+
+        let (seq, ts, text) = line("all good", 0, 0);
+        assert!(evaluate_alerts(&rules, &mut states, seq, ts, text).is_empty());
+
+        let (seq, ts, text) = line("an error happened", 1, 1);
+        let fired = evaluate_alerts(&rules, &mut states, seq, ts, text);
+        assert_eq!(fired.len(), 1);
+        assert!(fired[0].1.contains("an error happened"));
+    }
+
+    #[test]
+    fn disabled_rule_never_fires() {
+        let mut rule = immediate_rule(r#"regex("error")"#);
+        rule.enabled = false;
+        let mut states = HashMap::new();
+        let (seq, ts, text) = line("error!", 0, 0);
+        assert!(evaluate_alerts(&[rule], &mut states, seq, ts, text).is_empty());
+    }
+
+    #[test]
+    fn threshold_fires_once_count_crosses_and_not_before() {
+        let rule = AlertRule {
+            id: "r1".to_string(),
+            name: "spike".to_string(),
+            enabled: true,
+            condition: AlertCondition::Threshold {
+                query: r#"regex("error")"#.to_string(),
+                op: CompareOp::Ge,
+                threshold: 3.0,
+                window_ms: 10_000,
+            },
+            sink: AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
+        };
+        let mut states = HashMap::new();
+        for (seq, ms) in [(0u64, 0i64), (1, 1000)] {
+            let fired = evaluate_alerts(&[rule.clone()], &mut states, seq, Timestamp::from_millis(ms), "error");
+            assert!(fired.is_empty(), "should not fire before threshold is crossed");
+        }
+        let fired = evaluate_alerts(&[rule.clone()], &mut states, 2, Timestamp::from_millis(2000), "error");
+        assert_eq!(fired.len(), 1);
+        // Stays quiet for a non-matching line even though the window is full.
+        let fired = evaluate_alerts(&[rule], &mut states, 3, Timestamp::from_millis(2100), "fine");
+        assert!(fired.is_empty());
+    }
+
+    #[test]
+    fn threshold_window_expires_old_matches() {
+        let rule = AlertRule {
+            id: "r1".to_string(),
+            name: "spike".to_string(),
+            enabled: true,
+            condition: AlertCondition::Threshold {
+                query: r#"regex("error")"#.to_string(),
+                op: CompareOp::Ge,
+                threshold: 2.0,
+                window_ms: 1_000,
+            },
+            sink: AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
+        };
+        let mut states = HashMap::new();
+        assert!(evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "error").is_empty());
+        // Second match arrives well after the first has aged out of the window.
+        assert!(evaluate_alerts(&[rule], &mut states, 1, Timestamp::from_millis(5_000), "error").is_empty());
+    }
+
+    #[test]
+    fn new_value_fires_only_on_first_sighting() {
+        let rule = AlertRule {
+            id: "r1".to_string(),
+            name: "new user agent".to_string(),
+            enabled: true,
+            condition: AlertCondition::NewValue {
+                field_regex: r"agent=(?P<agent>\S+)".to_string(),
+                capture_group: "agent".to_string(),
+            },
+            sink: AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
+        };
+        let mut states = HashMap::new();
+        let fired = evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "agent=curl");
+        assert_eq!(fired.len(), 1);
+        let fired = evaluate_alerts(&[rule.clone()], &mut states, 1, Timestamp::from_millis(1), "agent=curl");
+        assert!(fired.is_empty());
+        let fired = evaluate_alerts(&[rule], &mut states, 2, Timestamp::from_millis(2), "agent=wget");
+        assert_eq!(fired.len(), 1);
+    }
+
+    #[test]
+    fn ratio_fires_when_crossing_and_skips_zero_denominator() {
+        let rule = AlertRule {
+            id: "r1".to_string(),
+            name: "error ratio".to_string(),
+            enabled: true,
+            condition: AlertCondition::Ratio {
+                query_a: r#"regex("error")"#.to_string(),
+                query_b: r#"regex("request")"#.to_string(),
+                op: CompareOp::Ge,
+                ratio: 0.5,
+                window_ms: 10_000,
+            },
+            sink: AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
+        };
+        let mut states = HashMap::new();
+        // Only an "error" so far; no "request" yet, so denominator is 0 -> skip.
+        assert!(evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "error").is_empty());
+        // One request: ratio 1/1 = 1.0 >= 0.5 -> fires.
+        let fired = evaluate_alerts(&[rule.clone()], &mut states, 1, Timestamp::from_millis(1), "request");
+        assert_eq!(fired.len(), 1);
+        // A second request: ratio 1/2 = 0.5 still >= 0.5 -> fires again.
+        let fired = evaluate_alerts(&[rule.clone()], &mut states, 2, Timestamp::from_millis(2), "request");
+        assert_eq!(fired.len(), 1);
+        // A third request with no more errors: ratio drops to 1/3 -> no longer fires.
+        let fired = evaluate_alerts(&[rule], &mut states, 3, Timestamp::from_millis(3), "request");
+        assert!(fired.is_empty());
+    }
+
+    #[test]
+    fn metric_comparison_in_alert_query_is_rejected_not_panicking() {
+        let rule = immediate_rule("some_metric[param]==5");
+        let mut states = HashMap::new();
+        // Should log a warning and simply not fire, never panic.
+        let fired = evaluate_alerts(&[rule], &mut states, 0, Timestamp::from_millis(0), "anything");
+        assert!(fired.is_empty());
+    }
+
+    #[tokio::test]
+    async fn deliver_ntfy_posts_message_to_server_topic_path() {
+        use axum::extract::State;
+        use axum::routing::post;
+        use axum::Router;
+        use std::sync::{Arc, Mutex};
+        use tokio::net::TcpListener;
+
+        #[derive(Clone, Default)]
+        struct Captured(Arc<Mutex<Option<(String, String)>>>);
+
+        async fn capture(
+            State(captured): State<Captured>,
+            axum::extract::Path(topic): axum::extract::Path<String>,
+            body: String,
+        ) -> &'static str {
+            *captured.0.lock().unwrap() = Some((topic, body));
+            "ok"
+        }
+
+        let captured = Captured::default();
+        let app = Router::new().route("/{topic}", post(capture)).with_state(captured.clone());
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        let server = format!("http://{addr}");
+        deliver_ntfy(&server, "alerts", "hello world").await.unwrap();
+
+        let (topic, body) = captured.0.lock().unwrap().clone().unwrap();
+        assert_eq!(topic, "alerts");
+        assert_eq!(body, "hello world");
+    }
+}

src/api.rs

@@ -1,33 +1,30 @@
 use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 
 use crate::dto::{
-    BucketValuesParams, CreateDashboardRequest, CreateProfileRequest, CssEntryDto, DashboardDto,
-    ErrorDto, ImportCssRequest, LinkCssRequest, MetricDeclarationDto, ParamDeclarationDto,
-    LogsParams, ProfileSummaryDto, PromqlParams, SaveDashboardRequest, SettingsDto,
-    TimelineParams, UpdateSettingsRequest,
+    BucketValuesParams, CreateAlertRuleRequest, CreateDashboardRequest, CreateProfileRequest,
+    CssEntryDto, DashboardDto, ErrorDto, ImportCssRequest, LinkCssRequest, MetricDeclarationDto,
+    ParamDeclarationDto, LogsParams, ProfileSummaryDto, PromqlParams, SaveDashboardRequest,
+    SettingsDto, TimelineParams, UpdateAlertRuleRequest, UpdateSettingsRequest,
 };
 use crate::engine::{
     DeclareMetricRequest, DeclareParamRequest, EngineHandle, EngineMessage, EngineQuery,
     EngineResponse,
 };
+use crate::alerts::AlertError;
 use crate::profile::{self, Profile, ProfileError, ProfileStore};
-use crate::settings::{CssError, Settings};
+use crate::settings::{CssError, SettingsHandle};
 use axum::extract::{Path as AxumPath, Query, State};
 use axum::http::{header, StatusCode};
 use axum::response::{IntoResponse, Json, Response};
-use axum::routing::{delete, get, post};
+use axum::routing::{delete, get, post, put};
 use axum::Router;
 use fs_embed::Dir as EmbedDir;
 use tokio::sync::oneshot;
 
 static WEBAPP_DIST: EmbedDir = fs_embed::fs_embed!("webapp/dist");
 
-/// Shared, process-wide settings handle. Cheap to clone; lock is only held
-/// briefly since settings changes are rare (unlike the engine's hot path).
-pub type SettingsHandle = Arc<Mutex<Settings>>;
-
 #[derive(Clone)]
 struct ApiState {
     engine: EngineHandle,
@@ -108,6 +105,11 @@ pub fn router(engine: EngineHandle, settings: SettingsHandle, settings_path: Pat
         .route("/api/settings/css/link", post(link_css))
         .route("/api/settings/css/{id}", delete(remove_css))
         .route("/api/settings/css/{id}/content", get(get_css_content))
+        .route("/api/settings/alerts", get(list_alert_rules).post(create_alert_rule))
+        .route(
+            "/api/settings/alerts/{id}",
+            put(update_alert_rule).delete(delete_alert_rule),
+        )
         .route("/api/settings/profiles", get(list_profiles).post(create_profile))
         .route("/api/settings/profiles/{name}", delete(delete_profile))
         .route("/api/settings/profiles/{name}/activate", post(activate_profile))
@@ -464,6 +466,63 @@ async fn get_css_content(
     }
 }
 
+/// Maps an `AlertError` to a response status/message: `NotFound` -> 404,
+/// `Io` -> 500. Mirrors `css_error_response` above.
+fn alert_error_response(err: AlertError) -> axum::response::Response {
+    match err {
+        AlertError::NotFound => (StatusCode::NOT_FOUND, err.to_string()).into_response(),
+        AlertError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
+    }
+}
+
+/// `GET /api/settings/alerts` — lists all configured alert rules as
+/// `Vec<AlertRule>`.
+async fn list_alert_rules(State(state): State<ApiState>) -> impl IntoResponse {
+    let settings = state.settings.lock().unwrap();
+    Json(settings.alert_rules.clone()).into_response()
+}
+
+/// `POST /api/settings/alerts` — creates an alert rule. Body:
+/// `{ "name": "...", "enabled": true, "condition": { "kind": "immediate" | \
+/// "threshold" | "newvalue" | "ratio", ... }, "sink": { "kind": "ntfy", \
+/// "topic": "...", "server": "..." | null } }`. Returns `201 Created` with
+/// the created rule (including its generated `id`).
+async fn create_alert_rule(
+    State(state): State<ApiState>,
+    Json(body): Json<CreateAlertRuleRequest>,
+) -> impl IntoResponse {
+    let mut settings = state.settings.lock().unwrap();
+    match settings.create_alert_rule(&state.settings_path, body.name, body.enabled, body.condition, body.sink) {
+        Ok(rule) => (StatusCode::CREATED, Json(rule)).into_response(),
+        Err(err) => alert_error_response(err),
+    }
+}
+
+/// `PUT /api/settings/alerts/:id` — replaces an alert rule's fields (id in
+/// the URL is authoritative). Returns the updated rule, or `404 Not Found`
+/// if no rule with `id` exists.
+async fn update_alert_rule(
+    State(state): State<ApiState>,
+    AxumPath(id): AxumPath<String>,
+    Json(body): Json<UpdateAlertRuleRequest>,
+) -> impl IntoResponse {
+    let mut settings = state.settings.lock().unwrap();
+    match settings.update_alert_rule(&state.settings_path, &id, body.name, body.enabled, body.condition, body.sink) {
+        Ok(rule) => Json(rule).into_response(),
+        Err(err) => alert_error_response(err),
+    }
+}
+
+/// `DELETE /api/settings/alerts/:id` — removes an alert rule. Returns
+/// `204 No Content`, or `404 Not Found` if no rule with `id` exists.
+async fn delete_alert_rule(State(state): State<ApiState>, AxumPath(id): AxumPath<String>) -> impl IntoResponse {
+    let mut settings = state.settings.lock().unwrap();
+    match settings.remove_alert_rule(&state.settings_path, &id) {
+        Ok(()) => StatusCode::NO_CONTENT.into_response(),
+        Err(err) => alert_error_response(err),
+    }
+}
+
 /// `GET /api/settings` — returns the subset of `Settings` the NavMenu UI
 /// needs (colour scheme, active profile name) as `SettingsDto`. CSS entries
 /// have their own dedicated list endpoint (`GET /api/settings/css`).
@@ -744,11 +803,13 @@ async fn delete_dashboard(
 mod tests {
     use super::*;
     use crate::engine::{self, IngestedLine};
+    use crate::settings::Settings;
     use crate::timestamp::Timestamp;
     use axum::body::Body;
     use axum::http::Request;
     use http_body_util::BodyExt;
     use serde_json::{json, Value};
+    use std::sync::Mutex;
     use tokio::sync::mpsc;
     use tower::ServiceExt;
 
@@ -758,12 +819,13 @@ mod tests {
     /// `TempDir` guard (kept alive for the test's duration).
     fn test_app() -> (Router, EngineHandle, SettingsHandle, tempfile::TempDir) {
         let (engine_tx, engine_rx) = mpsc::channel(128);
-        tokio::spawn(engine::run(engine_rx));
 
         let dir = tempfile::tempdir().unwrap();
         let settings_path = dir.path().join("settings.json");
         let settings = Arc::new(Mutex::new(Settings::load(&settings_path)));
 
+        tokio::spawn(engine::run(engine_rx, settings.clone()));
+
         let app = router(engine_tx.clone(), settings.clone(), settings_path);
         (app, engine_tx, settings, dir)
     }
@@ -799,6 +861,80 @@ mod tests {
         (status, value)
     }
 
+    #[tokio::test]
+    async fn alert_rule_crud_round_trips_through_http() {
+        let (app, _engine, _settings, _dir) = test_app();
+
+        let (status, body) = send_json(
+            &app,
+            "POST",
+            "/api/settings/alerts",
+            json!({
+                "name": "error spike",
+                "enabled": true,
+                "condition": { "kind": "immediate", "query": "regex(\"error\")" },
+                "sink": { "kind": "ntfy", "topic": "alerts", "server": null }
+            }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::CREATED);
+        assert_eq!(body["name"], "error spike");
+        let id = body["id"].as_str().unwrap().to_string();
+
+        let (status, body) = get_json(&app, "/api/settings/alerts").await;
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(body.as_array().unwrap().len(), 1);
+
+        let (status, body) = send_json(
+            &app,
+            "PUT",
+            &format!("/api/settings/alerts/{id}"),
+            json!({
+                "name": "renamed",
+                "enabled": false,
+                "condition": { "kind": "immediate", "query": "regex(\"warn\")" },
+                "sink": { "kind": "ntfy", "topic": "alerts", "server": null }
+            }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(body["name"], "renamed");
+        assert_eq!(body["enabled"], false);
+
+        let status = delete_request(&app, &format!("/api/settings/alerts/{id}")).await;
+        assert_eq!(status, StatusCode::NO_CONTENT);
+        let (_, body) = get_json(&app, "/api/settings/alerts").await;
+        assert!(body.as_array().unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn alert_rule_routes_404_for_unknown_id() {
+        let (app, _engine, _settings, _dir) = test_app();
+
+        let status = delete_request(&app, "/api/settings/alerts/does-not-exist").await;
+        assert_eq!(status, StatusCode::NOT_FOUND);
+
+        // `update_alert_rule`'s `AlertError::NotFound` maps to a plain-text
+        // 404 body (mirrors `css_error_response`), not JSON, so this checks
+        // the status directly rather than via `send_json`'s JSON parsing.
+        let request = Request::builder()
+            .method("PUT")
+            .uri("/api/settings/alerts/does-not-exist")
+            .header(header::CONTENT_TYPE, "application/json")
+            .body(Body::from(
+                json!({
+                    "name": "x",
+                    "enabled": true,
+                    "condition": { "kind": "immediate", "query": "regex(\"x\")" },
+                    "sink": { "kind": "ntfy", "topic": "x", "server": null }
+                })
+                .to_string(),
+            ))
+            .unwrap();
+        let response = app.clone().oneshot(request).await.unwrap();
+        assert_eq!(response.status(), StatusCode::NOT_FOUND);
+    }
+
     #[tokio::test]
     async fn declares_param_and_returns_201() {
         let (app, _engine, _settings, _dir) = test_app();

src/dto.rs

@@ -1,3 +1,4 @@
+use crate::alerts::{AlertCondition, AlertSink};
 use crate::data_value::{MetricPrimitiveKind, RegexToMetricPrimitiveMatcher};
 use crate::metric_primitive::MetricPrimitive;
 use crate::profile::{ChartType, Dashboard, Panel, Profile};
@@ -236,6 +237,28 @@ where
     Option::<T>::deserialize(deserializer).map(Some)
 }
 
+/// Request body for `POST /api/settings/alerts` (ticket 017): creates a new
+/// alert rule (see `crate::alerts::Settings::create_alert_rule`). `AlertRule`
+/// itself is returned directly as the response body (it already derives
+/// `Serialize`), so there's no separate response DTO here.
+#[derive(Debug, Deserialize)]
+pub struct CreateAlertRuleRequest {
+    pub name: String,
+    pub enabled: bool,
+    pub condition: AlertCondition,
+    pub sink: AlertSink,
+}
+
+/// Request body for `PUT /api/settings/alerts/:id`: replaces every field
+/// except `id` (taken from the URL, see `Settings::update_alert_rule`).
+#[derive(Debug, Deserialize)]
+pub struct UpdateAlertRuleRequest {
+    pub name: String,
+    pub enabled: bool,
+    pub condition: AlertCondition,
+    pub sink: AlertSink,
+}
+
 /// Summary of one profile for `GET /api/settings/profiles` (ticket 011):
 /// just the name and declaration/dashboard counts the Presets panel needs to
 /// render a list, rather than the full param/metric declaration bodies.

src/engine.rs

@@ -1,3 +1,4 @@
+use crate::alerts::{self, AlertRuntimeState};
 use crate::data_value::{DataValue, MetricPrimitiveKind, ParamRole, RegexToMetricPrimitiveMatcher};
 use crate::dto::{
     BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto,
@@ -8,11 +9,12 @@ use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
 use crate::index_key::IndexKey;
 use crate::metric_primitive::{MetricPrimitive, Number};
 use crate::promql::{self, AggOp, LabelMatcher, PromqlQuery};
+use crate::settings::SettingsHandle;
 use crate::timestamp::Timestamp;
 use regex::Regex;
 use serde::Deserialize;
 use std::cmp::Ordering;
-use std::collections::{BTreeMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use tokio::sync::{mpsc, oneshot};
 
 pub struct IngestedLine {
@@ -157,10 +159,14 @@ pub type EngineHandle = mpsc::Sender<EngineMessage>;
 /// Sole owner of the primary index. Runs single-threaded so the map never needs a
 /// lock, and does its own response serialization/DTO construction so callers never
 /// have to clone the index to answer a query.
-pub async fn run(mut rx: mpsc::Receiver<EngineMessage>) {
+pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle) {
     let mut index: BTreeMap<IndexKey, DataValue> = BTreeMap::new();
     let mut max_logs: Option<usize> = None;
     let mut log_count: usize = 0;
+    // Ticket 018: per-rule rolling windows / seen-value sets, keyed by rule
+    // id. Lives only as long as this loop (no restart persistence — see
+    // `alerts::AlertCondition::NewValue`'s doc comment).
+    let mut alert_states: HashMap<String, AlertRuntimeState> = HashMap::new();
 
     while let Some(message) = rx.recv().await {
         match message {
@@ -173,6 +179,15 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>) {
                 log_count += 1;
                 apply_all_metrics(&mut index, sequence, timestamp, &text);
                 evict_excess_logs(&mut index, &mut log_count, max_logs);
+
+                // Ticket 018: evaluate alert rules against just this line
+                // (not the whole log range), then hand any fired rule off to
+                // ticket 019's delivery, off the hot path via `tokio::spawn`.
+                let rules = settings.lock().unwrap().alert_rules.clone();
+                let fired = alerts::evaluate_alerts(&rules, &mut alert_states, sequence, timestamp, &text);
+                for (rule, message) in fired {
+                    tokio::spawn(alerts::deliver_for_rule(rule, message));
+                }
             }
             EngineMessage::SetMaxLogs(new_max) => {
                 max_logs = new_max;
@@ -739,7 +754,9 @@ fn compare_metric_primitive(
     Ok(ordering.is_some_and(|ordering| apply_compare_op(ordering, op)))
 }
 
-fn apply_compare_op(ordering: Ordering, op: CompareOp) -> bool {
+/// `pub(crate)` so `crate::alerts` can reuse it for `Threshold`/`Ratio`
+/// comparisons instead of duplicating the match.
+pub(crate) fn apply_compare_op(ordering: Ordering, op: CompareOp) -> bool {
     match op {
         CompareOp::Lt => ordering == Ordering::Less,
         CompareOp::Gt => ordering == Ordering::Greater,

src/filter_parser.rs

@@ -26,6 +26,7 @@
 use pest::Parser;
 use pest::iterators::Pair;
 use pest_derive::Parser as PestParser;
+use serde::{Deserialize, Serialize};
 
 #[derive(PestParser)]
 #[grammar = "filter_parser.pest"]
@@ -47,8 +48,11 @@ pub enum FilterExpr {
     Or(Box<FilterExpr>, Box<FilterExpr>),
 }
 
-/// Comparison operators supported by metric comparisons.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+/// Comparison operators supported by metric comparisons. Also reused by
+/// `crate::alerts::AlertCondition::Threshold`/`Ratio` for their own
+/// threshold/ratio comparisons (hence `Serialize`/`Deserialize`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
 pub enum CompareOp {
     Lt,
     Gt,

src/main.rs

@@ -1,3 +1,4 @@
+mod alerts;
 mod api;
 mod data_value;
 mod dto;
@@ -53,7 +54,7 @@ async fn main() {
     let settings = Arc::new(Mutex::new(Settings::load(&settings_path)));
 
     let (engine_tx, engine_rx) = mpsc::channel(1024);
-    tokio::spawn(engine::run(engine_rx));
+    tokio::spawn(engine::run(engine_rx, settings.clone()));
     {
         let max_logs = settings.lock().unwrap().max_logs;
         let _ = engine_tx.send(engine::EngineMessage::SetMaxLogs(max_logs)).await;

src/profile.rs

@@ -932,7 +932,8 @@ mod tests {
         let default_profile = store.get(DEFAULT_PROFILE_NAME).unwrap().clone();
 
         let (engine_tx, engine_rx) = mpsc::channel(128);
-        tokio::spawn(engine::run(engine_rx));
+        let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone()));
+        tokio::spawn(engine::run(engine_rx, settings_handle));
         load_into_engine(&engine_tx, &default_profile).await;
 
         for (sequence, text) in [(0u64, "all good"), (1, "an ERROR happened"), (2, "fine again")] {
@@ -1130,7 +1131,8 @@ mod tests {
 
         // Spawn a fresh engine and replay the profile's declarations into it.
         let (engine_tx, engine_rx) = mpsc::channel(16);
-        tokio::spawn(crate::engine::run(engine_rx));
+        let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone()));
+        tokio::spawn(crate::engine::run(engine_rx, settings_handle));
         load_into_engine(&engine_tx, &profile).await;
 
         // Ingest a line and confirm the engine is functioning with the replayed

src/settings.rs

@@ -11,11 +11,18 @@ use std::fs;
 use std::hash::{Hash, Hasher};
 use std::io;
 use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex};
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use directories::ProjectDirs;
 use serde::{Deserialize, Serialize};
 
+use crate::alerts::AlertRule;
+
+/// Shared, process-wide settings handle. Cheap to clone; lock is only held
+/// briefly since settings changes are rare (unlike the engine's hot path).
+pub type SettingsHandle = Arc<Mutex<Settings>>;
+
 /// User's preferred colour scheme for the webapp.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
 #[serde(rename_all = "lowercase")]
@@ -63,7 +70,10 @@ pub struct Profiles {
 }
 
 /// The single, process-wide settings instance.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+///
+/// Only `PartialEq` (not `Eq`): `alert_rules` carries `f64` fields
+/// (`AlertCondition::Threshold`/`Ratio`), which aren't `Eq`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct Settings {
     pub color_scheme: ColorScheme,
     pub css_entries: Vec<CssEntry>,
@@ -72,6 +82,10 @@ pub struct Settings {
     /// means unlimited — matches today's unbounded retention.
     #[serde(default)]
     pub max_logs: Option<usize>,
+    /// Alert rule definitions (ticket 017). `#[serde(default)]` so existing
+    /// settings.json files without this field still load.
+    #[serde(default)]
+    pub alert_rules: Vec<AlertRule>,
 }
 
 impl Default for Settings {
@@ -85,6 +99,7 @@ impl Default for Settings {
                 profiles_dir: "profiles".to_string(),
             },
             max_logs: None,
+            alert_rules: Vec::new(),
         }
     }
 }
@@ -355,6 +370,18 @@ mod tests {
         });
         settings.profiles.names.push("default".to_string());
         settings.profiles.active_profile = Some("default".to_string());
+        settings.alert_rules.push(crate::alerts::AlertRule {
+            id: "rule1".to_string(),
+            name: "error spike".to_string(),
+            enabled: true,
+            condition: crate::alerts::AlertCondition::Immediate {
+                query: r#"regex("error")"#.to_string(),
+            },
+            sink: crate::alerts::AlertSink::Ntfy {
+                topic: "alerts".to_string(),
+                server: None,
+            },
+        });
 
         settings.save(&path).unwrap();