Move alert evaluation to the profile actor (ticket 105)

Alert rules are profile-scoped and read metric buckets, which now live
on the profile actor (ticket 104) rather than the primary. alert_states
and the per-line evaluate_alerts/deliver_for_rule calls move from the
primary's Ingest arm to the profile actor's per-line loop, evaluated
against its own snapshot right after that line's apply_all_metrics
publishes -- matching the primary's original per-message publish-then-
evaluate ordering. The primary's run() no longer needs a SettingsHandle
at all now that alerts are gone from its Ingest arm; settings (for
alert rules and ntfy config, read fresh per line) moves to
engine::profile::spawn instead.

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

src/engine/handle.rs

@@ -3,7 +3,6 @@
 
 use crate::config::SettingsHandle;
 use crate::consts::{INGEST_LAG_WATERMARK_MAX_ENTRIES, INGEST_LAG_WATERMARK_WINDOW_MS};
-use crate::engine::alerts::{self, AlertRuntimeState};
 use crate::engine::index_key::IndexKey;
 use crate::engine::message::EngineMessage;
 use crate::engine::metric_primitive::{MetricPrimitive, Number};
@@ -12,7 +11,7 @@ use crate::engine::snapshot::params::log_stream_header_transition;
 use crate::engine::timestamp::Timestamp;
 use crate::engine::{ClearParamsResult, IngestedLine};
 use left_right::Absorb;
-use std::collections::{BTreeMap, HashMap};
+use std::collections::BTreeMap;
 use std::sync::Arc;
 use tokio::sync::{mpsc, oneshot, watch};
 
@@ -146,13 +145,13 @@ fn new_channel_and_handle(
 
 pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle {
     let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity);
-    let profile = crate::engine::profile::spawn(reader.clone(), version_rx);
+    let profile = crate::engine::profile::spawn(reader.clone(), version_rx, settings);
     let handle = EngineHandle {
         tx,
         reader,
         profile_reader: profile.reader.clone(),
     };
-    tokio::spawn(run(rx, settings, write, profile));
+    tokio::spawn(run(rx, write, profile));
     handle
 }
 
@@ -179,25 +178,24 @@ pub(crate) fn spawn_with_profile_observer(
     channel_capacity: usize,
 ) -> (EngineHandle, tokio::sync::mpsc::UnboundedReceiver<u64>) {
     let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity);
-    let (profile, observed) = crate::engine::profile::spawn_observed(reader.clone(), version_rx);
+    let (profile, observed) =
+        crate::engine::profile::spawn_observed(reader.clone(), version_rx, settings);
     let handle = EngineHandle {
         tx,
         reader,
         profile_reader: profile.reader.clone(),
     };
-    tokio::spawn(run(rx, settings, write, profile));
+    tokio::spawn(run(rx, write, profile));
     (handle, observed)
 }
 
 async fn run(
     mut rx: mpsc::Receiver<EngineMessage>,
-    settings: SettingsHandle,
     mut write: IndexWriter,
     profile: crate::engine::profile::ProfileHandle,
 ) {
     let mut max_logs: Option<usize> = None;
     let mut metric_sample_sequence: u64 = 0;
-    let mut alert_states: HashMap<String, AlertRuntimeState> = HashMap::new();
     let mut log_stream_header: Option<crate::engine::logfmt::LogStreamHeader> = None;
     let mut ingest_lag_samples: Vec<(Timestamp, i64)> = Vec::new();
 
@@ -226,48 +224,18 @@ async fn run(
                     log_stream_header.clone()
                 };
 
-                let (rules, ntfy_topic, ntfy_server) = {
-                    let settings = settings.lock().unwrap();
-                    (
-                        settings.alert_rules.clone(),
-                        settings.ntfy_topic.clone(),
-                        settings.ntfy_server.clone(),
-                    )
-                };
-
                 let text_for_op = Arc::clone(&text);
                 write.mutate(move |snap: &mut EngineSnapshot| {
                     // Primary/param extraction only (ticket 101) -- metric computation
-                    // (ticket 104) now happens on the profile actor, which wakes on this
+                    // (ticket 104) and alert evaluation (ticket 105) now happen on the
-                    // publish and reads the newly-visible line itself.
+                    // profile actor, which wakes on this publish and reads the
+                    // newly-visible line itself.
                     snap.ingest_params(sequence, timestamp, &text_for_op, row_header.as_ref());
                     snap.update_sort_helper_indexes_on_ingest(sequence, timestamp, &text_for_op);
                     snap.purge_stale_sort_helper_indexes();
                     snap.evict_excess_logs(max_logs);
                     snap.ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms;
                 });
-
-                let fired = write
-                    .peek()
-                    .map(|guard| {
-                        alerts::evaluate_alerts(
-                            &rules,
-                            &mut alert_states,
-                            &guard,
-                            sequence,
-                            timestamp,
-                            &text,
-                        )
-                    })
-                    .unwrap_or_default();
-                for (rule, message) in fired {
-                    tokio::spawn(alerts::deliver_for_rule(
-                        rule.name,
-                        message,
-                        ntfy_topic.clone(),
-                        ntfy_server.clone(),
-                    ));
-                }
             }
             EngineMessage::IngestMetricSample(sample) => {
                 let labels: BTreeMap<String, MetricPrimitive> = sample

src/engine/profile.rs

@@ -1,18 +1,20 @@
-//! The profile actor (ticket 104, building on ticket 103's skeleton). Owns metric declarations
+//! The profile actor (tickets 103-105). Owns metric declarations, bucket computation, and alert
-//! and bucket computation, moved off the primary per `ideas/0001-actor-split.md`'s resolved
+//! evaluation, moved off the primary per `ideas/0001-actor-split.md`'s resolved decision that
-//! decision that the primary "extracts declared + auto params only... no metrics, no profile
+//! the primary "extracts declared + auto params only... no metrics, no profile logic" and that
-//! logic." Reuses `EngineSnapshot` itself as this actor's own snapshot type — `declare_metric`/
+//! alerts move with metric computation since they read the buckets it produces. Reuses
-//! `update_metric`/`remove_metric_declaration`/`apply_all_metrics` all work unchanged — rather
+//! `EngineSnapshot` itself as this actor's own snapshot type — `declare_metric`/`update_metric`/
-//! than inventing a parallel type, since metric evaluation needs param declarations present in
+//! `remove_metric_declaration`/`apply_all_metrics` all work unchanged — rather than inventing a
-//! the same snapshot it reads (`resolved_param_declarations`); this actor mirrors the primary's
+//! parallel type, since metric evaluation needs param declarations present in the same snapshot
-//! param declarations into its own snapshot on every wake so that stays true, but stores no raw
+//! it reads (`resolved_param_declarations`); this actor mirrors the primary's param declarations
-//! log text of its own, reading it through `primary_reader` (a clone of the primary's
+//! into its own snapshot on every wake so that stays true, but stores no raw log text of its
-//! `IndexReader`) instead. `EngineHandle`'s public API is unchanged by this ticket —
+//! own, reading it through `primary_reader` (a clone of the primary's `IndexReader`) instead.
-//! `DeclareMetric`/`UpdateMetric`/`RemoveMetricDeclaration` are still sent to the primary's
+//! `EngineHandle`'s public API is unchanged by this ticket — `DeclareMetric`/`UpdateMetric`/
-//! `mpsc` channel and proxied here by `handle.rs`'s `run()` loop; ticket 106 gives API handlers
+//! `RemoveMetricDeclaration` are still sent to the primary's `mpsc` channel and proxied here by
-//! a direct route.
+//! `handle.rs`'s `run()` loop; ticket 106 gives API handlers a direct route.
 
+use crate::config::SettingsHandle;
 use crate::engine::DeclareMetricRequest;
+use crate::engine::alerts::{self, AlertRuntimeState};
 use crate::engine::data_value::{DataValue, ParamDeclaration};
 use crate::engine::handle::IndexReader;
 use crate::engine::index_key::IndexKey;
@@ -20,6 +22,7 @@ use crate::engine::snapshot::EngineSnapshot;
 use crate::engine::timestamp::Timestamp;
 use left_right::Absorb;
 use regex::Regex;
+use std::collections::HashMap;
 use tokio::sync::{mpsc, oneshot, watch};
 
 struct ProfileOp(Box<dyn FnMut(&mut EngineSnapshot) + Send + 'static>);
@@ -76,16 +79,26 @@ pub(crate) fn inert_reader() -> IndexReader {
 /// this actor its own read-only view of the primary's index (raw log text, param declarations);
 /// `primary_version` is bumped by the primary's `IndexWriter::mutate` after every `left_right`
 /// publish, so this actor blocks on `changed()` between primary writes rather than polling.
+/// `settings` is used to read alert rules and ntfy config fresh for each evaluated line (ticket
+/// 105 -- alert evaluation moved here from the primary alongside the metric buckets it reads).
 pub(crate) fn spawn(
     primary_reader: IndexReader,
     primary_version: watch::Receiver<u64>,
+    settings: SettingsHandle,
 ) -> ProfileHandle {
     let (tx, rx) = mpsc::channel(crate::consts::ENGINE_CHANNEL_CAPACITY);
     let (write, read) = left_right::new::<EngineSnapshot, ProfileOp>();
     let reader = IndexReader {
         factory: read.factory(),
     };
-    tokio::spawn(run(primary_reader, primary_version, rx, write, None));
+    tokio::spawn(run(
+        primary_reader,
+        primary_version,
+        rx,
+        write,
+        settings,
+        None,
+    ));
     ProfileHandle { tx, reader }
 }
 
@@ -96,6 +109,7 @@ pub(crate) fn spawn(
 pub(crate) fn spawn_observed(
     primary_reader: IndexReader,
     primary_version: watch::Receiver<u64>,
+    settings: SettingsHandle,
 ) -> (ProfileHandle, mpsc::UnboundedReceiver<u64>) {
     let (tx, rx) = mpsc::channel(crate::consts::ENGINE_CHANNEL_CAPACITY);
     let (write, read) = left_right::new::<EngineSnapshot, ProfileOp>();
@@ -108,6 +122,7 @@ pub(crate) fn spawn_observed(
         primary_version,
         rx,
         write,
+        settings,
         Some(advance_tx),
     ));
     (ProfileHandle { tx, reader }, advance_rx)
@@ -118,9 +133,11 @@ async fn run(
     mut primary_version: watch::Receiver<u64>,
     mut rx: mpsc::Receiver<ProfileMessage>,
     mut write: left_right::WriteHandle<EngineSnapshot, ProfileOp>,
+    settings: SettingsHandle,
     on_advance: Option<mpsc::UnboundedSender<u64>>,
 ) {
     let mut last_seen_sequence: Option<u64> = None;
+    let mut alert_states: HashMap<String, AlertRuntimeState> = HashMap::new();
 
     loop {
         tokio::select! {
@@ -143,12 +160,43 @@ async fn run(
                 drop(primary);
 
                 for (sequence, timestamp, text) in new_lines {
+                    let text_for_op = text.clone();
                     write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| {
-                        snap.apply_all_metrics(sequence, timestamp, &text);
+                        snap.apply_all_metrics(sequence, timestamp, &text_for_op);
                     })));
-                    last_seen_sequence = Some(sequence);
-                }
                     write.publish();
+                    last_seen_sequence = Some(sequence);
+
+                    let (rules, ntfy_topic, ntfy_server) = {
+                        let settings = settings.lock().unwrap();
+                        (
+                            settings.alert_rules.clone(),
+                            settings.ntfy_topic.clone(),
+                            settings.ntfy_server.clone(),
+                        )
+                    };
+                    let fired = write
+                        .enter()
+                        .map(|guard| {
+                            alerts::evaluate_alerts(
+                                &rules,
+                                &mut alert_states,
+                                &guard,
+                                sequence,
+                                timestamp,
+                                &text,
+                            )
+                        })
+                        .unwrap_or_default();
+                    for (rule, message) in fired {
+                        tokio::spawn(alerts::deliver_for_rule(
+                            rule.name,
+                            message,
+                            ntfy_topic.clone(),
+                            ntfy_server.clone(),
+                        ));
+                    }
+                }
                 if let (Some(sequence), Some(tx)) = (last_seen_sequence, on_advance.as_ref()) {
                     let _ = tx.send(sequence);
                 }

tickets/105-move-alert-evaluation-to-profile-actor.md

@@ -1,52 +0,0 @@
-# 105 — Move alert evaluation onto the profile actor
-
-## Summary
-
-Alert rules are profile-scoped and read metric buckets
-(`alert_states: HashMap<String, AlertRuntimeState>` lives on the
-single actor's stack frame today), so per [[0001-actor-split]]'s
-resolved decision, alert evaluation moves with metric computation.
-Depends on [[104-move-metric-computation-to-profile-actor]] landing
-first — alert evaluation reads the metric buckets that ticket moves
-onto `ProfileSnapshot`.
-
-## Where
-
-- `src/engine/handle.rs:141` — `let mut alert_states:
-  HashMap<String, AlertRuntimeState> = HashMap::new();`, currently a
-  local in the single actor's `run()` loop.
-- `src/engine/handle.rs`'s `Ingest` arm — today, right after
-  `write.mutate(...)` publishes, the actor does `write.peek()` and
-  calls `alerts::evaluate_alerts(&rules, &mut alert_states, &guard,
-  sequence, timestamp, &text)`, then spawns
-  `alerts::deliver_for_rule` for each fired alert. This whole block
-  moves to wherever the profile actor's per-line processing lives
-  (the watch-wake loop from ticket 103/104).
-- `src/engine/alerts.rs:139` (`AlertRuntimeState`), `:241`
-  (`evaluate_alerts`), `:384` (`deliver_for_rule`) — signatures take
-  the rules list and a `&EngineSnapshot`-shaped guard; check whether
-  `evaluate_alerts` needs to change its snapshot parameter type to
-  `ProfileSnapshot` (likely yes, since it reads bucket data that now
-  lives there) or whether it can stay snapshot-shape-agnostic via
-  whatever trait/accessor bucket-reading already uses.
-- Settings access for alert rules/ntfy config
-  (`src/engine/handle.rs:170-177`, `settings.lock().unwrap()`) — this
-  stays available to the profile actor the same way (it's a
-  `SettingsHandle` clone, not primary-actor state), so no primary
-  dependency here.
-
-## Scope
-
-- Move `alert_states` ownership to the profile actor's task-local
-  state.
-- Move the post-publish `evaluate_alerts` call and `deliver_for_rule`
-  spawn to fire from the profile actor's per-line processing, reading
-  `ProfileSnapshot` (post-metric-computation) instead of the primary's
-  `EngineSnapshot`.
-- Update `evaluate_alerts`'s signature in `src/engine/alerts.rs` if it
-  needs to accept a different snapshot type; keep its actual
-  evaluation logic unchanged — this ticket is a relocation, not a
-  rewrite of alert semantics.
-- Out of scope: any change to alert rule semantics, ntfy delivery, or
-  the `AlertRuntimeState` model itself — purely which actor owns and
-  drives evaluation.