Move metric declarations and bucket computation to the profile actor (ticket 104)

Metric declare/update/remove, backfill, and per-line apply_all_metrics
now run on the profile actor (engine::profile) instead of inline in the
primary's ingest closure, per idea 0001's resolved decision that the
primary handles params only. The profile actor reuses EngineSnapshot as
its own snapshot type -- declare_metric/update_metric/apply_all_metrics
work unchanged -- and mirrors the primary's param declarations into its
own snapshot on every wake (and synchronously before handling a
DeclareMetric/UpdateMetric message, since there's no ordering guarantee
between the watch-wake and message branches in tokio::select!) so
metric label/value resolution keeps working without storing a second
copy of raw log text; backfill reads lines through its cloned primary
IndexReader instead.

EngineHandle gained a profile_reader field so MetricDeclarations/
BucketValues/Promql queries route to the profile actor's snapshot
instead of the primary's -- without it, metrics could be declared but
never queried, since the primary no longer computes or stores them.
DeclareMetric/UpdateMetric/RemoveMetricDeclaration messages are still
sent to the primary's mpsc channel and proxied to the profile actor by
handle.rs's run() loop; EngineHandle's write-side API is otherwise
unchanged (ticket 106 gives API handlers a direct route). Alert
evaluation still reads the primary's (now metric-less) snapshot
directly -- an accepted gap until ticket 105 moves it, per idea 0001's
resolved decision that alerts move with metric computation.

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

src/api/mod.rs

@@ -249,6 +249,21 @@ fn mime_for_extension(extension: &str) -> &'static str {
 
 impl EngineHandle {
     async fn ask(&self, query: EngineQuery) -> Result<EngineResponse, StatusCode> {
+        // Metric declarations and bucket data live on the profile actor (ticket 104), not the
+        // primary -- route those query variants through its reader instead.
+        if matches!(
+            query,
+            EngineQuery::MetricDeclarations
+                | EngineQuery::BucketValues { .. }
+                | EngineQuery::Promql { .. }
+        ) {
+            let profile_handle = self.profile_reader.handle();
+            let guard = profile_handle
+                .enter()
+                .ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
+            return Ok(guard.handle_query_ro(query));
+        }
+
         let read_handle = self.reader.handle();
         if let Some(sort_by) = query.sort_by() {
             let needs_build = match read_handle.enter() {

src/config/profile.rs

@@ -751,7 +751,19 @@ mod tests {
     }
 
     async fn ask(engine: &EngineHandle, query: EngineQuery) -> EngineResponse {
-        let read_handle = engine.reader.handle();
+        // Metric declarations and bucket data live on the profile actor (ticket 104), not the
+        // primary -- mirrors the routing in `api::EngineHandle::ask`.
+        let profile_query = matches!(
+            query,
+            EngineQuery::MetricDeclarations
+                | EngineQuery::BucketValues { .. }
+                | EngineQuery::Promql { .. }
+        );
+        let read_handle = if profile_query {
+            engine.profile_reader.handle()
+        } else {
+            engine.reader.handle()
+        };
         let guard = read_handle.enter().expect("engine unavailable");
         guard.handle_query_ro(query)
     }

src/engine/handle.rs

@@ -61,7 +61,7 @@ impl IndexReadHandle {
 }
 
 pub struct IndexReader {
-    factory: left_right::ReadHandleFactory<EngineSnapshot>,
+    pub(crate) factory: left_right::ReadHandleFactory<EngineSnapshot>,
 }
 
 impl Clone for IndexReader {
@@ -81,6 +81,11 @@ impl IndexReader {
 pub struct EngineHandle {
     pub tx: mpsc::Sender<EngineMessage>,
     pub reader: IndexReader,
+    /// Read-only view of the profile actor's snapshot (`super::profile`), which owns metric
+    /// declarations and bucket data (ticket 104) — `MetricDeclarations`/`BucketValues`/`Promql`
+    /// queries read through this instead of `reader`. Not part of the write path; the profile
+    /// actor's own `mpsc` channel (proxied through `EngineMessage`'s metric variants) is that.
+    pub(crate) profile_reader: IndexReader,
 }
 
 impl Clone for EngineHandle {
@@ -88,6 +93,7 @@ impl Clone for EngineHandle {
         EngineHandle {
             tx: self.tx.clone(),
             reader: self.reader.clone(),
+            profile_reader: self.profile_reader.clone(),
         }
     }
 }
@@ -106,13 +112,16 @@ impl EngineHandle {
     }
 }
 
-/// Builds the handle/receiver/writer/version-receiver quadruple shared by `spawn` and, in
+/// Builds the primary actor's sender/receiver/reader/writer/version-receiver, shared by `spawn`
-/// tests, by callers that want the raw receiver without an actor loop draining it.
+/// and, in tests, by callers that want the raw receiver without an actor loop draining it.
+/// Stops short of building `EngineHandle` itself, since that also needs a profile-actor reader
+/// that doesn't exist until the caller spawns (or fakes, see `test_handle`) the profile actor.
 fn new_channel_and_handle(
     channel_capacity: usize,
 ) -> (
-    EngineHandle,
+    mpsc::Sender<EngineMessage>,
     mpsc::Receiver<EngineMessage>,
+    IndexReader,
     IndexWriter,
     watch::Receiver<u64>,
 ) {
@@ -123,8 +132,9 @@ fn new_channel_and_handle(
     };
     let (version_tx, version_rx) = watch::channel(0u64);
     (
-        EngineHandle { tx, reader },
+        tx,
         rx,
+        reader,
         IndexWriter {
             handle: write,
             version: 0,
@@ -135,9 +145,14 @@ fn new_channel_and_handle(
 }
 
 pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle {
-    let (handle, rx, write, version_rx) = new_channel_and_handle(channel_capacity);
+    let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity);
-    crate::engine::profile::spawn(handle.reader.clone(), version_rx);
+    let profile = crate::engine::profile::spawn(reader.clone(), version_rx);
-    tokio::spawn(run(rx, settings, write));
+    let handle = EngineHandle {
+        tx,
+        reader,
+        profile_reader: profile.reader.clone(),
+    };
+    tokio::spawn(run(rx, settings, write, profile));
     handle
 }
 
@@ -145,7 +160,14 @@ pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle
 pub(crate) fn test_handle(
     channel_capacity: usize,
 ) -> (EngineHandle, mpsc::Receiver<EngineMessage>) {
-    let (handle, rx, _write, _version_rx) = new_channel_and_handle(channel_capacity);
+    let (tx, rx, reader, _write, _version_rx) = new_channel_and_handle(channel_capacity);
+    let handle = EngineHandle {
+        tx,
+        reader,
+        // No actor loop is spawned here at all, so nothing ever publishes -- an inert reader
+        // over an always-empty snapshot is all a query through this handle could see anyway.
+        profile_reader: crate::engine::profile::inert_reader(),
+    };
     (handle, rx)
 }
 
@@ -156,9 +178,14 @@ pub(crate) fn spawn_with_profile_observer(
     settings: SettingsHandle,
     channel_capacity: usize,
 ) -> (EngineHandle, tokio::sync::mpsc::UnboundedReceiver<u64>) {
-    let (handle, rx, write, version_rx) = new_channel_and_handle(channel_capacity);
+    let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity);
-    let observed = crate::engine::profile::spawn_observed(handle.reader.clone(), version_rx);
+    let (profile, observed) = crate::engine::profile::spawn_observed(reader.clone(), version_rx);
-    tokio::spawn(run(rx, settings, write));
+    let handle = EngineHandle {
+        tx,
+        reader,
+        profile_reader: profile.reader.clone(),
+    };
+    tokio::spawn(run(rx, settings, write, profile));
     (handle, observed)
 }
 
@@ -166,6 +193,7 @@ 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;
@@ -209,10 +237,10 @@ async fn run(
 
                 let text_for_op = Arc::clone(&text);
                 write.mutate(move |snap: &mut EngineSnapshot| {
-                    // Two independently-callable stages (ticket 101): primary/param
+                    // Primary/param extraction only (ticket 101) -- metric computation
-                    // extraction, then profile-scoped metric computation.
+                    // (ticket 104) now happens on the 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.apply_all_metrics(sequence, timestamp, &text_for_op);
                     snap.update_sort_helper_indexes_on_ingest(sequence, timestamp, &text_for_op);
                     snap.purge_stale_sort_helper_indexes();
                     snap.evict_excess_logs(max_logs);
@@ -327,49 +355,40 @@ async fn run(
                 request,
                 respond_to,
             } => {
-                let result = write
+                // Metric declarations live on the profile actor (ticket 104), not here --
-                    .peek()
+                // proxy the message rather than mutating the primary's snapshot.
-                    .map(|guard| guard.validate_metric_request(&request).map(|_| ()))
+                let _ = profile
-                    .unwrap_or_else(|| Err("engine is shutting down".to_string()));
+                    .tx
-                let succeeded = result.is_ok();
+                    .send(crate::engine::profile::ProfileMessage::DeclareMetric {
-                let request_for_op = request.clone();
+                        request,
-                write.mutate(move |snap: &mut EngineSnapshot| {
+                        respond_to,
-                    if succeeded {
+                    })
-                        let _ = snap.declare_metric(request_for_op.clone());
+                    .await;
-                    }
-                });
-                let _ = respond_to.send(result);
             }
             EngineMessage::UpdateMetric {
                 name,
                 request,
                 respond_to,
             } => {
-                let result = write
+                let _ = profile
-                    .peek()
+                    .tx
-                    .map(|guard| guard.validate_update_metric(&name, &request).map(|_| ()))
+                    .send(crate::engine::profile::ProfileMessage::UpdateMetric {
-                    .unwrap_or_else(|| Err("engine is shutting down".to_string()));
+                        name,
-                let succeeded = result.is_ok();
+                        request,
-                let existing_name = name.clone();
+                        respond_to,
-                let request_for_op = request.clone();
+                    })
-                write.mutate(move |snap: &mut EngineSnapshot| {
+                    .await;
-                    if succeeded {
-                        let _ = snap.update_metric(&existing_name, request_for_op.clone());
-                    }
-                });
-                let _ = respond_to.send(result);
             }
             EngineMessage::RemoveMetricDeclaration { name, respond_to } => {
-                let existed = write.peek().is_some_and(|guard| {
+                let _ = profile
-                    guard
+                    .tx
-                        .index
+                    .send(
-                        .contains_key(&IndexKey::MetricDeclaration { name: name.clone() })
+                        crate::engine::profile::ProfileMessage::RemoveMetricDeclaration {
-                });
+                            name,
-                let name_for_op = name.clone();
+                            respond_to,
-                write.mutate(move |snap: &mut EngineSnapshot| {
+                        },
-                    snap.remove_metric_declaration(&name_for_op);
+                    )
-                });
+                    .await;
-                let _ = respond_to.send(existed);
             }
             EngineMessage::EnsureSortHelperIndex {
                 param_name,

src/engine/profile.rs

@@ -1,18 +1,30 @@
-//! The profile actor (ticket 103, first slice of the split described in
+//! The profile actor (ticket 104, building on ticket 103's skeleton). Owns metric declarations
-//! `ideas/0001-actor-split.md`). It wakes whenever the primary actor's `IndexWriter`
+//! and bucket computation, moved off the primary per `ideas/0001-actor-split.md`'s resolved
-//! (`super::handle`) publishes a `left_right` change, reads whatever primary log lines have
+//! decision that the primary "extracts declared + auto params only... no metrics, no profile
-//! become newly visible since it last looked, and — for now — does nothing with them beyond
+//! logic." Reuses `EngineSnapshot` itself as this actor's own snapshot type — `declare_metric`/
-//! tracking how far it's read. Metric computation (ticket 104) and alert evaluation
+//! `update_metric`/`remove_metric_declaration`/`apply_all_metrics` all work unchanged — rather
-//! (ticket 105) land in this wake loop next; nothing outside `engine::handle::spawn` talks to
+//! than inventing a parallel type, since metric evaluation needs param declarations present in
-//! this actor yet, so it has no public handle of its own.
+//! the same snapshot it reads (`resolved_param_declarations`); this actor mirrors the primary's
+//! param declarations into its own snapshot on every wake so that stays true, but stores no raw
+//! log text of its own, reading it through `primary_reader` (a clone of the primary's
+//! `IndexReader`) instead. `EngineHandle`'s public API is unchanged by this ticket —
+//! `DeclareMetric`/`UpdateMetric`/`RemoveMetricDeclaration` are still sent to the primary's
+//! `mpsc` channel and proxied here by `handle.rs`'s `run()` loop; ticket 106 gives API handlers
+//! a direct route.
 
+use crate::engine::DeclareMetricRequest;
+use crate::engine::data_value::{DataValue, ParamDeclaration};
 use crate::engine::handle::IndexReader;
+use crate::engine::index_key::IndexKey;
+use crate::engine::snapshot::EngineSnapshot;
+use crate::engine::timestamp::Timestamp;
 use left_right::Absorb;
-use tokio::sync::watch;
+use regex::Regex;
+use tokio::sync::{mpsc, oneshot, watch};
 
-struct ProfileOp(Box<dyn FnMut(&mut ProfileSnapshot) + Send + 'static>);
+struct ProfileOp(Box<dyn FnMut(&mut EngineSnapshot) + Send + 'static>);
 
-impl Absorb<ProfileOp> for ProfileSnapshot {
+impl Absorb<ProfileOp> for EngineSnapshot {
     fn absorb_first(&mut self, operation: &mut ProfileOp, _other: &Self) {
         (operation.0)(self);
     }
@@ -24,70 +36,281 @@ impl Absorb<ProfileOp> for ProfileSnapshot {
     }
 }
 
-/// Profile-actor-owned derived state. Empty besides a read watermark until ticket 104 moves
+/// Metric-declaration messages proxied to the profile actor by the primary's `run()` loop.
-/// metric declarations/buckets here and ticket 105 moves alert runtime state here.
+pub(crate) enum ProfileMessage {
-#[derive(Debug, Clone, Default)]
+    DeclareMetric {
-struct ProfileSnapshot {
+        request: DeclareMetricRequest,
-    /// Highest primary log sequence this actor has read so far, so the wake loop resumes from
+        respond_to: oneshot::Sender<Result<(), String>>,
-    /// there instead of rescanning the primary's whole retained range on every wake.
+    },
-    last_seen_sequence: Option<u64>,
+    UpdateMetric {
+        name: String,
+        request: DeclareMetricRequest,
+        respond_to: oneshot::Sender<Result<(), String>>,
+    },
+    RemoveMetricDeclaration {
+        name: String,
+        respond_to: oneshot::Sender<bool>,
+    },
+}
+
+pub(crate) struct ProfileHandle {
+    pub(crate) tx: mpsc::Sender<ProfileMessage>,
+    /// Read-only view of this actor's own snapshot (metric declarations, bucket data) --
+    /// `MetricDeclarations`/`BucketValues`/`Promql` queries read through this (see
+    /// `EngineHandle::profile_reader` in `handle.rs`) instead of the primary's reader.
+    pub(crate) reader: IndexReader,
+}
+
+/// A read-only view over a fresh, never-written `EngineSnapshot` -- used by `handle::test_handle`,
+/// which spawns no actor loop at all (so nothing would ever publish to a real profile reader
+/// anyway), just to give `EngineHandle::profile_reader` a valid value.
+#[cfg(test)]
+pub(crate) fn inert_reader() -> IndexReader {
+    let (_write, read) = left_right::new::<EngineSnapshot, ProfileOp>();
+    IndexReader {
+        factory: read.factory(),
+    }
 }
 
 /// Spawns the profile actor. `primary_reader` is a clone of the primary's `IndexReader`, giving
-/// this actor its own read-only view of the primary's index; `primary_version` is bumped by the
+/// this actor its own read-only view of the primary's index (raw log text, param declarations);
-/// primary's `IndexWriter::mutate` after every `left_right` publish, so this actor blocks on
+/// `primary_version` is bumped by the primary's `IndexWriter::mutate` after every `left_right`
-/// `changed()` between primary writes rather than polling.
+/// publish, so this actor blocks on `changed()` between primary writes rather than polling.
-pub(crate) fn spawn(primary_reader: IndexReader, primary_version: watch::Receiver<u64>) {
+pub(crate) fn spawn(
-    spawn_inner(primary_reader, primary_version, None);
+    primary_reader: IndexReader,
+    primary_version: watch::Receiver<u64>,
+) -> 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));
+    ProfileHandle { tx, reader }
 }
 
-/// Test-only variant of `spawn` that reports every watermark advance on an unbounded channel,
+/// Test-only variant of `spawn` that reports every ingest-driven watermark advance on an
-/// so a test can assert the wake/read loop actually ran without reaching into actor-private
+/// unbounded channel, so a test can assert the wake/read loop actually ran without reaching
-/// state (`ProfileSnapshot` has no public read API yet — this ticket is plumbing only).
+/// into actor-private state.
 #[cfg(test)]
 pub(crate) fn spawn_observed(
     primary_reader: IndexReader,
     primary_version: watch::Receiver<u64>,
-) -> tokio::sync::mpsc::UnboundedReceiver<u64> {
+) -> (ProfileHandle, mpsc::UnboundedReceiver<u64>) {
-    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+    let (tx, rx) = mpsc::channel(crate::consts::ENGINE_CHANNEL_CAPACITY);
-    spawn_inner(primary_reader, primary_version, Some(tx));
+    let (write, read) = left_right::new::<EngineSnapshot, ProfileOp>();
-    rx
+    let reader = IndexReader {
+        factory: read.factory(),
+    };
+    let (advance_tx, advance_rx) = mpsc::unbounded_channel();
+    tokio::spawn(run(
+        primary_reader,
+        primary_version,
+        rx,
+        write,
+        Some(advance_tx),
+    ));
+    (ProfileHandle { tx, reader }, advance_rx)
 }
 
-fn spawn_inner(
+async fn run(
     primary_reader: IndexReader,
     mut primary_version: watch::Receiver<u64>,
-    #[cfg_attr(not(test), allow(unused_variables))] on_advance: Option<
+    mut rx: mpsc::Receiver<ProfileMessage>,
-        tokio::sync::mpsc::UnboundedSender<u64>,
+    mut write: left_right::WriteHandle<EngineSnapshot, ProfileOp>,
-    >,
+    on_advance: Option<mpsc::UnboundedSender<u64>>,
 ) {
-    tokio::spawn(async move {
+    let mut last_seen_sequence: Option<u64> = None;
-        let (mut write, _read) = left_right::new::<ProfileSnapshot, ProfileOp>();
 
-        while primary_version.changed().await.is_ok() {
+    loop {
+        tokio::select! {
+            changed = primary_version.changed() => {
+                if changed.is_err() {
+                    break; // primary actor gone, shut down
+                }
                 let primary_handle = primary_reader.handle();
                 let Some(primary) = primary_handle.enter() else {
                     continue;
                 };
-            let last_seen = write.enter().and_then(|guard| guard.last_seen_sequence);
-            let range_start = last_seen.map_or(0, |sequence| sequence.saturating_add(1));
 
-            let mut new_last_seen = last_seen;
+                mirror_param_declarations(&primary, &mut write);
-            for (sequence, _timestamp, _text) in primary.logs_in_sequence_range(range_start..) {
+
-                new_last_seen =
+                let range_start = last_seen_sequence.map_or(0, |sequence| sequence.saturating_add(1));
-                    Some(new_last_seen.map_or(sequence, |current| current.max(sequence)));
+                let new_lines: Vec<(u64, Timestamp, String)> = primary
-            }
+                    .logs_in_sequence_range(range_start..)
+                    .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string()))
+                    .collect();
                 drop(primary);
 
-            if new_last_seen != last_seen {
+                for (sequence, timestamp, text) in new_lines {
-                write.append(ProfileOp(Box::new(move |snap: &mut ProfileSnapshot| {
+                    write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| {
-                    snap.last_seen_sequence = new_last_seen;
+                        snap.apply_all_metrics(sequence, timestamp, &text);
                     })));
+                    last_seen_sequence = Some(sequence);
+                }
                 write.publish();
-                if let (Some(sequence), Some(tx)) = (new_last_seen, on_advance.as_ref()) {
+                if let (Some(sequence), Some(tx)) = (last_seen_sequence, on_advance.as_ref()) {
                     let _ = tx.send(sequence);
                 }
             }
+            message = rx.recv() => {
+                let Some(message) = message else {
+                    break; // primary actor dropped its ProfileHandle, shut down
+                };
+                handle_message(message, &primary_reader, &mut write);
             }
+        }
+    }
+}
+
+fn handle_message(
+    message: ProfileMessage,
+    primary_reader: &IndexReader,
+    write: &mut left_right::WriteHandle<EngineSnapshot, ProfileOp>,
+) {
+    match message {
+        ProfileMessage::DeclareMetric {
+            request,
+            respond_to,
+        } => {
+            // Freshen the param mirror synchronously rather than relying on the watch-wake
+            // loop having already caught up -- a param declared just before this metric (e.g.
+            // `Profile::load_into_engine` declaring params then metrics in sequence) may not
+            // have been mirrored yet when this message is handled, since there's no ordering
+            // guarantee between the wake branch and this one in `tokio::select!`.
+            if let Some(primary) = primary_reader.handle().enter() {
+                mirror_param_declarations(&primary, write);
+            }
+            let result = write
+                .enter()
+                .map(|guard| guard.validate_metric_request(&request).map(|_| ()))
+                .unwrap_or_else(|| Err("engine is shutting down".to_string()));
+            let succeeded = result.is_ok();
+            let lines = primary_owned_lines(primary_reader);
+            let request_for_op = request.clone();
+            write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| {
+                if succeeded {
+                    let _ = snap.declare_metric_with_lines(request_for_op.clone(), &lines);
+                }
+            })));
+            write.publish();
+            let _ = respond_to.send(result);
+        }
+        ProfileMessage::UpdateMetric {
+            name,
+            request,
+            respond_to,
+        } => {
+            if let Some(primary) = primary_reader.handle().enter() {
+                mirror_param_declarations(&primary, write);
+            }
+            let result = write
+                .enter()
+                .map(|guard| guard.validate_update_metric(&name, &request).map(|_| ()))
+                .unwrap_or_else(|| Err("engine is shutting down".to_string()));
+            let succeeded = result.is_ok();
+            let lines = primary_owned_lines(primary_reader);
+            let existing_name = name.clone();
+            let request_for_op = request.clone();
+            write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| {
+                if succeeded {
+                    let _ = snap.update_metric_with_lines(
+                        &existing_name,
+                        request_for_op.clone(),
+                        &lines,
+                    );
+                }
+            })));
+            write.publish();
+            let _ = respond_to.send(result);
+        }
+        ProfileMessage::RemoveMetricDeclaration { name, respond_to } => {
+            let existed = write.enter().is_some_and(|guard| {
+                guard
+                    .index
+                    .contains_key(&IndexKey::MetricDeclaration { name: name.clone() })
             });
+            let name_for_op = name.clone();
+            write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| {
+                snap.remove_metric_declaration(&name_for_op);
+            })));
+            write.publish();
+            let _ = respond_to.send(existed);
+        }
+    }
+}
+
+/// Reads all currently-retained primary lines through `primary_reader`, for backfilling a
+/// newly declared/updated metric — the profile actor's own snapshot holds no raw log text, so
+/// this is its only source of backfill history (see `EngineSnapshot::declare_metric_with_lines`).
+fn primary_owned_lines(primary_reader: &IndexReader) -> Vec<(u64, Timestamp, String)> {
+    let primary_handle = primary_reader.handle();
+    primary_handle
+        .enter()
+        .map(|guard| {
+            guard
+                .logs_range()
+                .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string()))
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
+/// Replaces the profile snapshot's mirrored `ParamDeclaration` entries and `param_regex_cache`
+/// with the primary's current ones, so metric label/value resolution
+/// (`EngineSnapshot::resolved_param_declarations`) sees the same params the primary has —
+/// cheap enough to do unconditionally on every wake since declared params are capped
+/// (`MAX_AUTO_DECLARED_PARAMS`).
+fn mirror_param_declarations(
+    primary: &EngineSnapshot,
+    write: &mut left_right::WriteHandle<EngineSnapshot, ProfileOp>,
+) {
+    let declarations: Vec<ParamDeclaration> = primary
+        .param_declarations_range()
+        .map(
+            |(
+                name,
+                message_match_regex,
+                capture_group,
+                json_field_path,
+                kind,
+                elapsed_since_last,
+                auto_derived,
+                log_stream_gated,
+            )| ParamDeclaration {
+                name: name.to_string(),
+                message_match_regex: message_match_regex.to_string(),
+                capture_group: capture_group.to_string(),
+                json_field_path: json_field_path.map(str::to_string),
+                kind,
+                elapsed_since_last,
+                auto_derived,
+                log_stream_gated,
+            },
+        )
+        .collect();
+
+    write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| {
+        let stale: Vec<IndexKey> = snap
+            .index
+            .range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit)
+            .map(|(key, _)| key.clone())
+            .collect();
+        for key in stale {
+            snap.index.remove(&key);
+        }
+        snap.param_regex_cache.clear();
+        for declaration in &declarations {
+            if let Ok(re) = Regex::new(&declaration.message_match_regex) {
+                snap.param_regex_cache.insert(declaration.name.clone(), re);
+            }
+            snap.index.insert(
+                IndexKey::ParamDeclaration {
+                    name: declaration.name.clone(),
+                },
+                DataValue::ParamDeclaration(declaration.clone()),
+            );
+        }
+    })));
+    write.publish();
 }

src/engine/snapshot/mod.rs

@@ -394,18 +394,52 @@ impl EngineSnapshot {
         Ok(re)
     }
 
+    /// Self-contained declare, backfilling from `self.logs_range()` — only used by tests that
+    /// build a standalone `EngineSnapshot` without the actor split; production code always goes
+    /// through the profile actor, which calls `declare_metric_with_lines` directly (its own
+    /// snapshot holds no raw log text of its own to backfill from).
+    #[cfg_attr(not(test), allow(dead_code))]
     pub(crate) fn declare_metric(&mut self, request: DeclareMetricRequest) -> Result<(), String> {
+        let lines = self.owned_logs_range();
+        self.declare_metric_with_lines(request, &lines)
+    }
+
+    /// Like `declare_metric`, but backfills from `lines` instead of `self.logs_range()` — the
+    /// profile actor (`engine::profile`) calls this, since its own snapshot holds no raw log
+    /// text (only mirrored param declarations plus its own metric declarations/buckets); it
+    /// sources backfill lines by reading through its cloned primary `IndexReader` instead.
+    pub(crate) fn declare_metric_with_lines(
+        &mut self,
+        request: DeclareMetricRequest,
+        lines: &[(u64, Timestamp, String)],
+    ) -> Result<(), String> {
         let re = self.validate_metric_request(&request)?;
         self.metric_regex_cache
             .insert(request.name.clone(), re.clone());
-        self.resolve_backfill_and_store_metric(request, &re);
+        self.resolve_backfill_and_store_metric(request, &re, lines);
         Ok(())
     }
 
-    /// Shared by `declare_metric` and `update_metric`: resolves param declarations, backfills
+    /// `logs_range` collected into owned `(sequence, timestamp, text)` triples, for callers
-    /// the metric's buckets from existing log history, then writes the `MetricDeclaration` entry.
+    /// (like `declare_metric`) that need to hand backfill lines to a method taking them by
-    /// Callers differ only in what happens before this (rename bookkeeping for updates).
+    /// value rather than borrowing `self` for the duration.
-    fn resolve_backfill_and_store_metric(&mut self, request: DeclareMetricRequest, re: &Regex) {
+    #[cfg_attr(not(test), allow(dead_code))]
+    fn owned_logs_range(&self) -> Vec<(u64, Timestamp, String)> {
+        self.logs_range()
+            .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string()))
+            .collect()
+    }
+
+    /// Shared by `declare_metric`/`update_metric` (and their `_with_lines` variants): resolves
+    /// param declarations, backfills the metric's buckets from `lines`, then writes the
+    /// `MetricDeclaration` entry. Callers differ only in what happens before this (rename
+    /// bookkeeping for updates) and where `lines` comes from.
+    fn resolve_backfill_and_store_metric(
+        &mut self,
+        request: DeclareMetricRequest,
+        re: &Regex,
+        lines: &[(u64, Timestamp, String)],
+    ) {
         let resolved_params = self.resolved_param_declarations();
         self.backfill_metric(
             &request.name,
@@ -415,6 +449,7 @@ impl EngineSnapshot {
             &resolved_params,
             request.matched_labels.as_ref(),
             request.unmatched_labels.as_ref(),
+            lines,
         );
         self.index.insert(
             IndexKey::MetricDeclaration {
@@ -451,10 +486,25 @@ impl EngineSnapshot {
         self.validate_metric_request(request)
     }
 
+    /// Self-contained update, backfilling from `self.logs_range()` — see `declare_metric`'s doc
+    /// comment; only used by tests that build a standalone `EngineSnapshot`.
+    #[cfg_attr(not(test), allow(dead_code))]
     pub(crate) fn update_metric(
         &mut self,
         existing_name: &str,
         request: DeclareMetricRequest,
+    ) -> Result<(), String> {
+        let lines = self.owned_logs_range();
+        self.update_metric_with_lines(existing_name, request, &lines)
+    }
+
+    /// Like `update_metric`, but backfills from `lines` instead of `self.logs_range()` — see
+    /// `declare_metric_with_lines`.
+    pub(crate) fn update_metric_with_lines(
+        &mut self,
+        existing_name: &str,
+        request: DeclareMetricRequest,
+        lines: &[(u64, Timestamp, String)],
     ) -> Result<(), String> {
         let re = self.validate_update_metric(existing_name, &request)?;
 
@@ -468,7 +518,7 @@ impl EngineSnapshot {
         self.metric_regex_cache
             .insert(request.name.clone(), re.clone());
 
-        self.resolve_backfill_and_store_metric(request, &re);
+        self.resolve_backfill_and_store_metric(request, &re, lines);
         Ok(())
     }
 
@@ -593,11 +643,8 @@ impl EngineSnapshot {
         resolved_params: &HashMap<String, ResolvedParam>,
         matched_labels: Option<&BTreeMap<String, String>>,
         unmatched_labels: Option<&BTreeMap<String, String>>,
+        lines: &[(u64, Timestamp, String)],
     ) {
-        let lines: Vec<(u64, Timestamp, String)> = self
-            .logs_range()
-            .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string()))
-            .collect();
         for (sequence, timestamp, text) in lines {
             self.apply_metric_to_log(
                 metric_name,
@@ -607,9 +654,9 @@ impl EngineSnapshot {
                 resolved_params,
                 matched_labels,
                 unmatched_labels,
-                sequence,
+                *sequence,
-                timestamp,
+                *timestamp,
-                &text,
+                text,
             );
         }
     }

tickets/104-move-metric-computation-to-profile-actor.md

@@ -1,73 +0,0 @@
-# 104 — Move metric declarations and bucket computation into the profile actor
-
-## Summary
-
-Second slice of the split: metric declarations, `apply_all_metrics`,
-and the declare/update/remove/backfill machinery move off the primary
-`EngineSnapshot` and onto the profile actor's `ProfileSnapshot`
-(stood up in [[103-profile-actor-skeleton-and-primary-signaling]]).
-The primary keeps raw log storage and param declarations only, per
-[[0001-actor-split]]'s resolved decision that the primary "extracts
-declared + auto params only... no metrics, no profile logic." The
-profile actor consumes newly-visible primary lines (via the
-watch-wake loop from ticket 103) and runs metric regexes against them
-itself, building its own bucket data independent of the primary.
-
-## Where
-
-- `src/engine/snapshot/mod.rs:364-434` (`validate_metric_request`,
-  `declare_metric`, `resolve_backfill_and_store_metric`) and
-  `:434-487` (`validate_update_metric`, `update_metric`,
-  `remove_metric_declaration`) — the declare/update/remove machinery
-  to move onto `ProfileSnapshot`.
-- `src/engine/snapshot/mod.rs:587-617` (`backfill_metric`,
-  `apply_all_metrics`) — backfill replays `self.logs_range()` (a
-  primary-side accessor) through `apply_metric_to_log`; once metrics
-  live on the profile actor, backfill needs to read logs through the
-  profile actor's cloned primary `IndexReader` instead of `self`.
-- `src/engine/handle.rs` — `EngineMessage::DeclareMetric`/
-  `UpdateMetric`/`RemoveMetricDeclaration` arms (see the message list
-  at `src/engine/message.rs:30-63`) currently route through the
-  single actor's `write.mutate` against `EngineSnapshot`; these need
-  to route to the profile actor's `ProfileSnapshot` instead.
-- `src/engine/handle.rs`'s `Ingest` arm — `snap.apply_all_metrics(...)`
-  is called directly after `ingest_params` today
-  (see ticket 101's change); this call moves out of the primary's
-  ingest closure entirely once the profile actor computes metrics
-  itself from the lines it observes via the watch/read loop, rather
-  than the primary computing them inline.
-
-## Scope
-
-- Move `MetricDeclaration`/bucket-related `IndexKey`/`DataValue`
-  variants' *usage* (not necessarily their type definitions, which are
-  shared) onto `ProfileSnapshot`'s own index/map, separate from the
-  primary's.
-- Move `declare_metric`/`update_metric`/`remove_metric_declaration`/
-  `backfill_metric`/`apply_all_metrics` (and their validation
-  counterparts) to operate on `ProfileSnapshot`, reading raw log text
-  via the profile actor's cloned primary `IndexReader` wherever they
-  need `logs_range`-style access, instead of `self.index` on the
-  primary.
-- Remove the `apply_all_metrics` call from the primary's `Ingest` arm
-  in `handle.rs` — the primary no longer computes metrics inline; the
-  profile actor's watch-wake loop (from ticket 103) now calls
-  `apply_all_metrics`-equivalent logic itself per newly-visible line.
-- `DeclareMetric`/`UpdateMetric`/`RemoveMetricDeclaration` messages
-  need a new route to the profile actor (a second `mpsc` channel, or
-  folding into the profile actor's own message enum) — decide the
-  concrete mechanism during implementation; `EngineHandle`'s public
-  API doesn't need to change yet if this ticket keeps routing internal
-  to `handle.rs` (full API surface changes are
-  [[106-split-enginehandle-api-for-primary-and-profile]]'s scope).
-- Out of scope: alert evaluation (still reads the primary's snapshot
-  directly today — [[105-move-alert-evaluation-to-profile-actor]]
-  moves it), and the profile-switch (`activate_profile`) reload path
-  (still targets whatever this ticket lands as the metric-owning
-  actor, but retargeting `src/api/mod.rs:812-852` explicitly is
-  [[106-split-enginehandle-api-for-primary-and-profile]]'s scope).
-- Backfill cost and blocking behavior: per [[0001-actor-split]]'s
-  performance rationale, this move is what stops a slow backfill from
-  blocking the primary's ingest/query-serving — confirm with a test
-  that ingesting new lines on the primary keeps working while the
-  profile actor is mid-backfill on a newly declared metric.