Improve ingestion performance

Pure code-motion refactor, no logic changes. Follows the repo convention
of one struct/enum + its impl per module file:

- engine/handle.rs: the left-right write/read plumbing (EngineOp,
  IndexWriter, IndexReadHandle, IndexReader, EngineHandle) and the actor
  loop (spawn/run) that owns EngineSnapshot.
- engine/message.rs: EngineQuery/EngineMessage/EngineResponse.
- engine/snapshot/: EngineSnapshot itself, split by concern into
  params.rs (declaration + auto-detect), metrics.rs (declare/apply/
  backfill), filter.rs (search-bar filter compilation/eval), sort.rs
  (sort-helper indexes), and query.rs (handle_query_inner + PromQL
  evaluation), with mod.rs holding the struct and the accessors shared
  across all of them.
- engine/tests.rs: the inline test module, kept as one file (per the
  task's own escape hatch) since most tests exercise ingest -> declare
  -> query across several of the new submodules at once rather than one
  in isolation.

Visibility was narrowed per call site: pub(super) where only sibling
snapshot submodules need an item, pub(crate) where engine::handle or
engine::tests (outside the snapshot subtree) also need it.

cargo build --all-targets and cargo test are green (291 passed, 1
ignored, same as before the split).

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

src/engine/handle.rs

@@ -0,0 +1,302 @@
+//! The left-right write/read plumbing around `EngineSnapshot`, the actor loop that owns it,
+//! and the `EngineHandle` other tasks use to talk to that actor.
+
+use crate::consts::{
+    INGEST_LAG_WATERMARK_MAX_ENTRIES, INGEST_LAG_WATERMARK_WINDOW_MS,
+};
+use crate::engine::alerts::{self, AlertRuntimeState};
+use crate::engine::data_value::DataValue;
+use crate::engine::index_key::IndexKey;
+use crate::engine::message::EngineMessage;
+use crate::engine::metric_primitive::{MetricPrimitive, Number};
+use crate::engine::snapshot::params::log_stream_header_transition;
+use crate::engine::snapshot::{approx_log_bytes, EngineSnapshot};
+use crate::engine::timestamp::Timestamp;
+use crate::engine::{ClearParamsResult, IngestedLine};
+use crate::config::SettingsHandle;
+use left_right::Absorb;
+use std::collections::{BTreeMap, HashMap};
+use tokio::sync::{mpsc, oneshot};
+
+struct EngineOp(Box<dyn FnMut(&mut EngineSnapshot) + Send + 'static>);
+
+impl Absorb<EngineOp> for EngineSnapshot {
+    fn absorb_first(&mut self, operation: &mut EngineOp, _other: &Self) {
+        (operation.0)(self);
+    }
+
+    fn drop_first(self: Box<Self>) {}
+
+    fn sync_with(&mut self, first: &Self) {
+        *self = first.clone();
+    }
+}
+
+struct IndexWriter {
+    handle: left_right::WriteHandle<EngineSnapshot, EngineOp>,
+}
+
+impl IndexWriter {
+    fn mutate(&mut self, op: impl FnMut(&mut EngineSnapshot) + Send + 'static) {
+        self.handle.append(EngineOp(Box::new(op)));
+        self.handle.publish();
+    }
+
+    fn peek(&mut self) -> Option<left_right::ReadGuard<'_, EngineSnapshot>> {
+        self.handle.enter()
+    }
+}
+
+pub struct IndexReadHandle(left_right::ReadHandle<EngineSnapshot>);
+
+impl IndexReadHandle {
+    pub(crate) fn enter(&self) -> Option<left_right::ReadGuard<'_, EngineSnapshot>> {
+        self.0.enter()
+    }
+}
+
+pub struct IndexReader {
+    factory: left_right::ReadHandleFactory<EngineSnapshot>,
+}
+
+impl Clone for IndexReader {
+    fn clone(&self) -> Self {
+        IndexReader { factory: self.factory.clone() }
+    }
+}
+
+impl IndexReader {
+    pub(crate) fn handle(&self) -> IndexReadHandle {
+        IndexReadHandle(self.factory.handle())
+    }
+}
+
+pub struct EngineHandle {
+    pub tx: mpsc::Sender<EngineMessage>,
+    pub reader: IndexReader,
+}
+
+impl Clone for EngineHandle {
+    fn clone(&self) -> Self {
+        EngineHandle { tx: self.tx.clone(), reader: self.reader.clone() }
+    }
+}
+
+impl EngineHandle {
+    pub async fn barrier(&self) {
+        let (respond_to, response) = oneshot::channel();
+        if self.tx.send(EngineMessage::Barrier(respond_to)).await.is_ok() {
+            let _ = response.await;
+        }
+    }
+}
+
+pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle {
+    let (tx, rx) = mpsc::channel(channel_capacity);
+    let (write, read) = left_right::new::<EngineSnapshot, EngineOp>();
+    let reader = IndexReader { factory: read.factory() };
+    tokio::spawn(run(rx, settings, IndexWriter { handle: write }));
+    EngineHandle { tx, reader }
+}
+
+#[cfg(test)]
+pub(crate) fn test_handle(channel_capacity: usize) -> (EngineHandle, mpsc::Receiver<EngineMessage>) {
+    let (tx, rx) = mpsc::channel(channel_capacity);
+    let (_write, read) = left_right::new::<EngineSnapshot, EngineOp>();
+    (EngineHandle { tx, reader: IndexReader { factory: read.factory() } }, rx)
+}
+
+async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle, mut write: IndexWriter) {
+    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();
+
+    while let Some(message) = rx.recv().await {
+        match message {
+            EngineMessage::Ingest(line) => {
+                let IngestedLine { sequence, timestamp, text } = line;
+                let processed_at = Timestamp::now();
+                record_ingest_lag(
+                    &mut ingest_lag_samples,
+                    processed_at,
+                    (processed_at.as_millis() - timestamp.as_millis()).max(0),
+                );
+                let ingest_lag_high_watermark_ms =
+                    ingest_lag_high_watermark_ms(&ingest_lag_samples, processed_at);
+
+                let is_new_header = crate::engine::logfmt::LogStreamHeader::detect(&text).is_some();
+                log_stream_header = log_stream_header_transition(&log_stream_header, &text);
+                let row_header = if is_new_header { None } else { 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 = text.clone();
+                write.mutate(move |snap: &mut EngineSnapshot| {
+                    snap.approx_bytes_total += approx_log_bytes(&text_for_op);
+                    snap.index.insert(
+                        IndexKey::IngestedLog { sequence, timestamp },
+                        DataValue::LogLiteral { text: text_for_op.clone() },
+                    );
+                    snap.log_count += 1;
+                    if let Some(header) = row_header.as_ref() {
+                        snap.derive_log_stream_params_from_row(header, &text_for_op);
+                    }
+                    snap.auto_declare_params(&text_for_op);
+                    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);
+                    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
+                    .labels
+                    .into_iter()
+                    .map(|(name, value)| (name, MetricPrimitive::String(value)))
+                    .collect();
+                let metric_name = sample.metric_name;
+                let sequence = metric_sample_sequence;
+                let timestamp = sample.timestamp;
+                let value = MetricPrimitive::Number(Number::Float(sample.value));
+                write.mutate(move |snap: &mut EngineSnapshot| {
+                    snap.push_bucket_value(&metric_name, labels.clone(), sequence, timestamp, value.clone());
+                });
+                metric_sample_sequence += 1;
+            }
+            EngineMessage::SetMaxLogs(new_max) => {
+                max_logs = new_max;
+                write.mutate(move |snap: &mut EngineSnapshot| {
+                    snap.evict_excess_logs(new_max);
+                });
+            }
+            EngineMessage::DeclareParam { request, respond_to } => {
+                let result = write
+                    .peek()
+                    .map(|guard| guard.validate_param_declaration(&request))
+                    .unwrap_or_else(|| Err("engine is shutting down".to_string()));
+                let succeeded = result.is_ok();
+                let name = request.name.clone();
+                let request_for_op = request.clone();
+                write.mutate(move |snap: &mut EngineSnapshot| {
+                    snap.sort_helper_indexes.remove(&name);
+                    if succeeded {
+                        snap.insert_param_declaration(request_for_op.clone());
+                    }
+                });
+                let _ = respond_to.send(result);
+            }
+            EngineMessage::RemoveParamDeclaration { name, respond_to } => {
+                let existed = write
+                    .peek()
+                    .is_some_and(|guard| guard.index.contains_key(&IndexKey::ParamDeclaration { name: name.clone() }));
+                let name_for_op = name.clone();
+                write.mutate(move |snap: &mut EngineSnapshot| {
+                    snap.index.remove(&IndexKey::ParamDeclaration { name: name_for_op.clone() });
+                    snap.sort_helper_indexes.remove(&name_for_op);
+                    snap.param_regex_cache.remove(&name_for_op);
+                });
+                let _ = respond_to.send(existed);
+            }
+            EngineMessage::ClearParams { respond_to } => {
+                let mut scratch = EngineSnapshot::from_index(
+                    write.peek().map(|guard| guard.index.clone()).unwrap_or_default(),
+                );
+                let removed = scratch.clear_all_param_declarations();
+                let (reextracted, logs_considered) = scratch.reextract_params_from_retained_logs();
+                write.mutate(move |snap: &mut EngineSnapshot| {
+                    snap.clear_all_param_declarations();
+                    snap.reextract_params_from_retained_logs();
+                    snap.sort_helper_indexes.clear();
+                });
+                let _ = respond_to.send(ClearParamsResult { removed, reextracted, logs_considered });
+            }
+            EngineMessage::DeclareMetric { request, respond_to } => {
+                let result = write
+                    .peek()
+                    .map(|guard| guard.validate_metric_request(&request).map(|_| ()))
+                    .unwrap_or_else(|| Err("engine is shutting down".to_string()));
+                let succeeded = result.is_ok();
+                let request_for_op = request.clone();
+                write.mutate(move |snap: &mut EngineSnapshot| {
+                    if succeeded {
+                        let _ = snap.declare_metric(request_for_op.clone());
+                    }
+                });
+                let _ = respond_to.send(result);
+            }
+            EngineMessage::UpdateMetric { name, request, respond_to } => {
+                let result = write
+                    .peek()
+                    .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 existing_name = name.clone();
+                let request_for_op = request.clone();
+                write.mutate(move |snap: &mut EngineSnapshot| {
+                    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| guard.index.contains_key(&IndexKey::MetricDeclaration { name: name.clone() }));
+                let name_for_op = name.clone();
+                write.mutate(move |snap: &mut EngineSnapshot| {
+                    snap.remove_metric_declaration(&name_for_op);
+                });
+                let _ = respond_to.send(existed);
+            }
+            EngineMessage::EnsureSortHelperIndex { param_name, respond_to } => {
+                write.mutate(move |snap: &mut EngineSnapshot| {
+                    snap.ensure_sort_helper_index_built(&param_name);
+                });
+                let _ = respond_to.send(());
+            }
+            EngineMessage::Barrier(respond_to) => {
+                let _ = respond_to.send(());
+            }
+        }
+    }
+}
+
+pub(crate) fn record_ingest_lag(samples: &mut Vec<(Timestamp, i64)>, now: Timestamp, lag_ms: i64) {
+    samples.retain(|(recorded_at, _)| now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS);
+    if samples.len() < INGEST_LAG_WATERMARK_MAX_ENTRIES {
+        samples.push((now, lag_ms));
+        return;
+    }
+    let Some((min_index, &(_, min_lag))) = samples.iter().enumerate().min_by_key(|(_, (_, lag))| *lag) else {
+        return;
+    };
+    if lag_ms > min_lag {
+        samples[min_index] = (now, lag_ms);
+    }
+}
+
+pub(crate) fn ingest_lag_high_watermark_ms(samples: &[(Timestamp, i64)], now: Timestamp) -> i64 {
+    samples
+        .iter()
+        .filter(|(recorded_at, _)| now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS)
+        .map(|(_, lag)| *lag)
+        .max()
+        .unwrap_or(0)
+}

src/engine/message.rs

@@ -0,0 +1,84 @@
+//! The engine actor's inbound message/query types and its outbound response type.
+
+use crate::api::dto::{
+    BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto,
+    PromqlSeriesDto, StatsDto, TimelineBucketDto, TimelineParams,
+};
+use crate::engine::{ClearParamsResult, DeclareMetricRequest, DeclareParamRequest, IngestedLine, MetricSample};
+use std::collections::BTreeMap;
+use tokio::sync::oneshot;
+
+#[allow(dead_code)]
+pub enum EngineQuery {
+    Logs(LogsParams),
+    Timeline(TimelineParams),
+    Stats,
+    ParamDeclarations,
+    MetricDeclarations,
+    BucketValues {
+        metric_name: String,
+        labels_filter: Option<BTreeMap<String, String>>,
+        start_ms: Option<i64>,
+        end_ms: Option<i64>,
+        since_sequence: Option<u64>,
+        limit: Option<usize>,
+    },
+    Promql { query: String, start_ms: Option<i64>, end_ms: Option<i64> },
+}
+
+#[allow(dead_code)]
+pub enum EngineMessage {
+    Ingest(IngestedLine),
+    IngestMetricSample(MetricSample),
+    DeclareParam {
+        request: DeclareParamRequest,
+        respond_to: oneshot::Sender<Result<(), String>>,
+    },
+    RemoveParamDeclaration {
+        name: String,
+        respond_to: oneshot::Sender<bool>,
+    },
+    ClearParams {
+        respond_to: oneshot::Sender<ClearParamsResult>,
+    },
+    DeclareMetric {
+        request: DeclareMetricRequest,
+        respond_to: oneshot::Sender<Result<(), String>>,
+    },
+    UpdateMetric {
+        name: String,
+        request: DeclareMetricRequest,
+        respond_to: oneshot::Sender<Result<(), String>>,
+    },
+    RemoveMetricDeclaration {
+        name: String,
+        respond_to: oneshot::Sender<bool>,
+    },
+    SetMaxLogs(Option<usize>),
+    EnsureSortHelperIndex {
+        param_name: String,
+        respond_to: oneshot::Sender<()>,
+    },
+    Barrier(oneshot::Sender<()>),
+}
+
+#[allow(dead_code)]
+pub enum EngineResponse {
+    Logs(Vec<LogEntryDto>),
+    Timeline(Vec<TimelineBucketDto>),
+    Stats(StatsDto),
+    ParamDeclarations(Vec<ParamDeclarationDto>),
+    MetricDeclarations(Vec<MetricDeclarationDto>),
+    BucketValues(Vec<BucketValueDto>),
+    Promql(Result<Vec<PromqlSeriesDto>, String>),
+    Error(String),
+}
+
+impl EngineQuery {
+    pub(crate) fn sort_by(&self) -> Option<&str> {
+        match self {
+            EngineQuery::Logs(params) => params.sort_by.as_deref(),
+            _ => None,
+        }
+    }
+}

src/engine/snapshot/filter.rs

@@ -0,0 +1,261 @@
+//! Compiling and evaluating the search-bar filter syntax (`filter_parser`) against log lines
+//! and buckets: regex matches, the `ottyl_seq`/`ottyl_ts`/`ottyl_fmt` pseudo-metrics, per-param
+//! metric-bucket comparisons, and ad-hoc JSON field comparisons.
+
+use crate::engine::filter_datetime;
+use crate::engine::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
+use crate::engine::index_key::IndexKey;
+use crate::engine::data_value::DataValue;
+use crate::engine::metric_primitive::MetricPrimitive;
+use crate::engine::snapshot::EngineSnapshot;
+use crate::engine::timestamp::Timestamp;
+use regex::Regex;
+use std::cmp::Ordering;
+use std::collections::HashSet;
+
+pub(super) enum CompiledFilter {
+    Regex(Regex),
+    Sequence { op: CompareOp, target: f64 },
+    Timestamp { op: CompareOp, target_ms: i64 },
+    Fmt { op: CompareOp, target: String },
+    MetricMatch(HashSet<u64>),
+    Json { field: String, op: CompareOp, value: FilterValue },
+    And(Box<CompiledFilter>, Box<CompiledFilter>),
+    Or(Box<CompiledFilter>, Box<CompiledFilter>),
+}
+
+impl EngineSnapshot {
+    pub(super) fn compile_filter_query(&self, filter: &str, now: Timestamp) -> Result<CompiledFilter, String> {
+        let expr = filter_parser::parse(filter).map_err(|err| err.to_string())?;
+        self.compile_filter(&expr, now)
+    }
+
+    fn compile_filter(&self, expr: &FilterExpr, now: Timestamp) -> Result<CompiledFilter, String> {
+        match expr {
+            FilterExpr::Regex(pattern) => Regex::new(pattern)
+                .map(CompiledFilter::Regex)
+                .map_err(|err| format!("invalid regex in filter: {err}")),
+            FilterExpr::Comparison { metric, param, op, value } => match metric.as_str() {
+                "ottyl_seq" => CompiledFilter::from_seq_comparison(*op, value),
+                "ottyl_ts" => CompiledFilter::from_ts_comparison(*op, value, now),
+                "ottyl_fmt" => Ok(CompiledFilter::from_fmt_comparison(*op, value)),
+                other => self.compile_metric_comparison(other, param.as_deref(), *op, value, now),
+            },
+            FilterExpr::JsonComparison { field, op, value } => Ok(CompiledFilter::Json {
+                field: field.clone(),
+                op: *op,
+                value: value.clone(),
+            }),
+            FilterExpr::And(lhs, rhs) => Ok(CompiledFilter::And(
+                Box::new(self.compile_filter(lhs, now)?),
+                Box::new(self.compile_filter(rhs, now)?),
+            )),
+            FilterExpr::Or(lhs, rhs) => Ok(CompiledFilter::Or(
+                Box::new(self.compile_filter(lhs, now)?),
+                Box::new(self.compile_filter(rhs, now)?),
+            )),
+        }
+    }
+
+    fn compile_metric_comparison(
+        &self,
+        metric: &str,
+        param: Option<&str>,
+        op: CompareOp,
+        value: &FilterValue,
+        now: Timestamp,
+    ) -> Result<CompiledFilter, String> {
+        let param = param.ok_or_else(|| {
+            format!(
+                "metric comparison on '{metric}' requires a [param] (e.g. '{metric}[some_param]{}...'); \
+                 v1 buckets only carry per-label occurrence data, not a single metric value",
+                match op {
+                    CompareOp::Lt => "<",
+                    CompareOp::Gt => ">",
+                    CompareOp::Le => "<=",
+                    CompareOp::Ge => ">=",
+                    CompareOp::Eq => "==",
+                    CompareOp::Ne => "!=",
+                }
+            )
+        })?;
+
+        if let Some(DataValue::MetricDeclaration { label_params, .. }) =
+            self.index.get(&IndexKey::MetricDeclaration { name: metric.to_string() })
+        {
+            if !label_params.iter().any(|p| p == param) {
+                return Err(format!(
+                    "param '{param}' is not a label param of metric '{metric}'; declared label params: [{}]",
+                    label_params.join(", ")
+                ));
+            }
+        }
+
+        let mut matching = HashSet::new();
+        for (labels, (sequence, _, _)) in self.bucket_values(metric, None) {
+            let Some(primitive) = labels.get(param) else {
+                continue;
+            };
+            if primitive.compare(op, value, now)? {
+                matching.insert(*sequence);
+            }
+        }
+        Ok(CompiledFilter::MetricMatch(matching))
+    }
+}
+
+impl CompiledFilter {
+    fn from_seq_comparison(op: CompareOp, value: &FilterValue) -> Result<Self, String> {
+        let target = match value {
+            FilterValue::Number(n) => *n,
+            FilterValue::Raw(raw) => raw.parse::<f64>().map_err(|_| {
+                format!("'ottyl_seq' comparisons require a numeric value, got {raw:?}")
+            })?,
+        };
+        Ok(CompiledFilter::Sequence { op, target })
+    }
+
+    fn from_ts_comparison(op: CompareOp, value: &FilterValue, now: Timestamp) -> Result<Self, String> {
+        let target_ms = match value {
+            FilterValue::Number(n) => *n as i64,
+            FilterValue::Raw(raw) => filter_datetime::parse_datetime(raw, now)
+                .map_err(|err| err.to_string())?
+                .as_millis(),
+        };
+        Ok(CompiledFilter::Timestamp { op, target_ms })
+    }
+
+    fn from_fmt_comparison(op: CompareOp, value: &FilterValue) -> Self {
+        let target = match value {
+            FilterValue::Number(n) => n.to_string(),
+            FilterValue::Raw(raw) => raw.clone(),
+        };
+        CompiledFilter::Fmt { op, target }
+    }
+}
+
+impl MetricPrimitive {
+    fn compare(&self, op: CompareOp, value: &FilterValue, now: Timestamp) -> Result<bool, String> {
+        let ordering = match (self, value) {
+            (MetricPrimitive::Number(n), FilterValue::Number(v)) => n.as_f64().partial_cmp(v),
+            (MetricPrimitive::Number(n), FilterValue::Raw(raw)) => {
+                let v: f64 = raw
+                    .parse()
+                    .map_err(|_| format!("expected a numeric value, got {raw:?}"))?;
+                n.as_f64().partial_cmp(&v)
+            }
+            (MetricPrimitive::Timestamp(ts), FilterValue::Number(v)) => {
+                ts.as_millis().partial_cmp(&(*v as i64))
+            }
+            (MetricPrimitive::Timestamp(ts), FilterValue::Raw(raw)) => {
+                let target = filter_datetime::parse_datetime(raw, now).map_err(|err| err.to_string())?;
+                ts.as_millis().partial_cmp(&target.as_millis())
+            }
+            (MetricPrimitive::String(s), FilterValue::Raw(raw)) => s.as_str().partial_cmp(raw.as_str()),
+            (MetricPrimitive::String(s), FilterValue::Number(n)) => {
+                s.as_str().partial_cmp(n.to_string().as_str())
+            }
+            (MetricPrimitive::Duration(ms), FilterValue::Number(v)) => (*ms as f64).partial_cmp(v),
+            (MetricPrimitive::Duration(ms), FilterValue::Raw(raw)) => {
+                let v: f64 = raw
+                    .parse()
+                    .map_err(|_| format!("expected a numeric value, got {raw:?}"))?;
+                (*ms as f64).partial_cmp(&v)
+            }
+            (MetricPrimitive::Boolean(b), FilterValue::Raw(raw)) => {
+                let v: bool = raw
+                    .parse()
+                    .map_err(|_| format!("expected true or false, got {raw:?}"))?;
+                b.partial_cmp(&v)
+            }
+            (MetricPrimitive::Boolean(_), FilterValue::Number(_)) => None,
+        };
+        Ok(ordering.is_some_and(|ordering| op.matches(ordering)))
+    }
+}
+
+pub(crate) fn eval_json_comparison(text: &str, field: &str, op: CompareOp, value: &FilterValue) -> bool {
+    let Ok(parsed) = serde_json::from_str::<serde_json::Value>(text) else {
+        return false;
+    };
+    let Some(field_value) = parsed.as_object().and_then(|obj| obj.get(field)) else {
+        return false;
+    };
+    compare_json_value(field_value, op, value)
+}
+
+fn compare_json_value(field_value: &serde_json::Value, op: CompareOp, value: &FilterValue) -> bool {
+    match field_value {
+        serde_json::Value::Number(n) => {
+            let target = match value {
+                FilterValue::Number(v) => Some(*v),
+                FilterValue::Raw(raw) => raw.parse::<f64>().ok(),
+            };
+            match (n.as_f64(), target) {
+                (Some(f), Some(v)) => f.partial_cmp(&v).is_some_and(|ordering| op.matches(ordering)),
+                _ => false,
+            }
+        }
+        serde_json::Value::String(s) => {
+            let target = match value {
+                FilterValue::Raw(raw) => raw.clone(),
+                FilterValue::Number(n) => n.to_string(),
+            };
+            s.as_str()
+                .partial_cmp(target.as_str())
+                .is_some_and(|ordering| op.matches(ordering))
+        }
+        serde_json::Value::Bool(b) => {
+            let FilterValue::Raw(raw) = value else {
+                return false;
+            };
+            match raw.parse::<bool>() {
+                Ok(v) => match op {
+                    CompareOp::Eq => *b == v,
+                    CompareOp::Ne => *b != v,
+                    _ => false,
+                },
+                Err(_) => false,
+            }
+        }
+        serde_json::Value::Null | serde_json::Value::Array(_) | serde_json::Value::Object(_) => false,
+    }
+}
+
+impl CompareOp {
+    pub(crate) fn matches(&self, ordering: Ordering) -> bool {
+        match self {
+            CompareOp::Lt => ordering == Ordering::Less,
+            CompareOp::Gt => ordering == Ordering::Greater,
+            CompareOp::Le => ordering != Ordering::Greater,
+            CompareOp::Ge => ordering != Ordering::Less,
+            CompareOp::Eq => ordering == Ordering::Equal,
+            CompareOp::Ne => ordering != Ordering::Equal,
+        }
+    }
+}
+
+impl CompiledFilter {
+    pub(super) fn eval(&self, sequence: u64, timestamp: Timestamp, text: &str) -> bool {
+        match self {
+            CompiledFilter::Regex(re) => re.is_match(text),
+            CompiledFilter::Sequence { op, target } => {
+                op.matches((sequence as f64).total_cmp(target))
+            }
+            CompiledFilter::Timestamp { op, target_ms } => {
+                op.matches(timestamp.as_millis().cmp(target_ms))
+            }
+            CompiledFilter::Fmt { op, target } => {
+                op.matches(crate::engine::logfmt::detect_format(text).cmp(target.as_str()))
+            }
+            CompiledFilter::MetricMatch(sequences) => sequences.contains(&sequence),
+            CompiledFilter::Json { field, op, value } => eval_json_comparison(text, field, *op, value),
+            CompiledFilter::And(lhs, rhs) => {
+                lhs.eval(sequence, timestamp, text) && rhs.eval(sequence, timestamp, text)
+            }
+            CompiledFilter::Or(lhs, rhs) => {
+                lhs.eval(sequence, timestamp, text) || rhs.eval(sequence, timestamp, text)
+            }
+        }
+    }
+}

src/engine/snapshot/metrics.rs

@@ -0,0 +1,333 @@
+//! Metric declaration and application: validating/storing declarations, applying them to
+//! ingested lines to populate metric buckets, and backfilling a newly (re)declared metric
+//! against already-ingested logs.
+
+use crate::engine::data_value::{DataValue, MetricPrimitiveKind};
+use crate::engine::index_key::IndexKey;
+use crate::engine::metric_primitive::{MetricPrimitive, Number};
+use crate::engine::snapshot::params::ResolvedParam;
+use crate::engine::snapshot::EngineSnapshot;
+use crate::engine::timestamp::Timestamp;
+use crate::engine::DeclareMetricRequest;
+use regex::Regex;
+use std::collections::{BTreeMap, HashMap};
+
+impl EngineSnapshot {
+    pub(crate) fn validate_metric_request(&self, request: &DeclareMetricRequest) -> Result<Regex, String> {
+        let re = Regex::new(&request.message_match_regex)
+            .map_err(|err| format!("invalid regex: {err}"))?;
+
+        let declared =
+            |name: &str| self.index.contains_key(&IndexKey::ParamDeclaration { name: name.to_string() });
+
+        let mut errors: Vec<String> = request
+            .label_params
+            .iter()
+            .filter(|name| !declared(name))
+            .map(|name| format!("label param '{name}' references an undeclared param"))
+            .collect();
+        if let Some(value_param) = &request.value_param {
+            if !declared(value_param) {
+                errors.push(format!("value param '{value_param}' references an undeclared param"));
+            }
+        }
+        if !errors.is_empty() {
+            return Err(errors.join("; "));
+        }
+
+        Ok(re)
+    }
+
+    pub(crate) fn declare_metric(&mut self, request: DeclareMetricRequest) -> Result<(), String> {
+        let re = self.validate_metric_request(&request)?;
+        self.metric_regex_cache.insert(request.name.clone(), re.clone());
+        let resolved_params = self.resolved_param_declarations();
+
+        self.backfill_metric(
+            &request.name,
+            &re,
+            &request.label_params,
+            request.value_param.as_deref(),
+            &resolved_params,
+            request.matched_labels.as_ref(),
+            request.unmatched_labels.as_ref(),
+        );
+        self.index.insert(
+            IndexKey::MetricDeclaration { name: request.name },
+            DataValue::MetricDeclaration {
+                message_match_regex: request.message_match_regex,
+                label_params: request.label_params,
+                value_param: request.value_param,
+                matched_labels: request.matched_labels,
+                unmatched_labels: request.unmatched_labels,
+            },
+        );
+        Ok(())
+    }
+
+    pub(crate) fn validate_update_metric(
+        &self,
+        existing_name: &str,
+        request: &DeclareMetricRequest,
+    ) -> Result<Regex, String> {
+        if !self.index.contains_key(&IndexKey::MetricDeclaration { name: existing_name.to_string() }) {
+            return Err(format!("no metric declaration named '{existing_name}'"));
+        }
+        if request.name != existing_name
+            && self.index.contains_key(&IndexKey::MetricDeclaration { name: request.name.clone() })
+        {
+            return Err(format!("a metric named '{}' already exists", request.name));
+        }
+        self.validate_metric_request(request)
+    }
+
+    pub(crate) fn update_metric(&mut self, existing_name: &str, request: DeclareMetricRequest) -> Result<(), String> {
+        let re = self.validate_update_metric(existing_name, &request)?;
+        let resolved_params = self.resolved_param_declarations();
+
+        self.clear_metric_data(existing_name);
+        if request.name != existing_name {
+            self.index.remove(&IndexKey::MetricDeclaration { name: existing_name.to_string() });
+            self.metric_regex_cache.remove(existing_name);
+        }
+        self.metric_regex_cache.insert(request.name.clone(), re.clone());
+
+        self.backfill_metric(
+            &request.name,
+            &re,
+            &request.label_params,
+            request.value_param.as_deref(),
+            &resolved_params,
+            request.matched_labels.as_ref(),
+            request.unmatched_labels.as_ref(),
+        );
+        self.index.insert(
+            IndexKey::MetricDeclaration { name: request.name.clone() },
+            DataValue::MetricDeclaration {
+                message_match_regex: request.message_match_regex,
+                label_params: request.label_params,
+                value_param: request.value_param,
+                matched_labels: request.matched_labels,
+                unmatched_labels: request.unmatched_labels,
+            },
+        );
+        Ok(())
+    }
+
+    fn clear_metric_data(&mut self, metric_name: &str) {
+        self.index.retain(|key, _| match key {
+            IndexKey::Bucket { metric_name: name, .. } => name != metric_name,
+            IndexKey::LastMetricEvent { metric_name: name, .. } => name != metric_name,
+            _ => true,
+        });
+    }
+
+    pub(crate) fn remove_metric_declaration(&mut self, name: &str) -> bool {
+        let existed = self.index.remove(&IndexKey::MetricDeclaration { name: name.to_string() }).is_some();
+        self.metric_regex_cache.remove(name);
+        self.clear_metric_data(name);
+        existed
+    }
+}
+
+pub(super) fn parse_metric_primitive(raw: &str, kind: MetricPrimitiveKind) -> Option<MetricPrimitive> {
+    match kind {
+        MetricPrimitiveKind::String => Some(MetricPrimitive::String(raw.to_string())),
+        MetricPrimitiveKind::Number => raw
+            .parse::<i64>()
+            .map(Number::Int)
+            .or_else(|_| raw.parse::<f64>().map(Number::Float))
+            .ok()
+            .map(MetricPrimitive::Number),
+        MetricPrimitiveKind::Timestamp => {
+            crate::engine::filter_datetime::parse_datetime(raw, Timestamp::now())
+                .ok()
+                .map(MetricPrimitive::Timestamp)
+        }
+        MetricPrimitiveKind::Duration => raw.parse::<i64>().ok().map(MetricPrimitive::Duration),
+        MetricPrimitiveKind::Boolean => raw.parse::<bool>().ok().map(MetricPrimitive::Boolean),
+    }
+}
+
+impl EngineSnapshot {
+    #[allow(clippy::too_many_arguments)]
+    fn apply_metric_to_log(
+        &mut self,
+        metric_name: &str,
+        regex: &Regex,
+        label_params: &[String],
+        value_param: Option<&str>,
+        resolved_params: &HashMap<String, ResolvedParam>,
+        matched_labels: Option<&BTreeMap<String, String>>,
+        unmatched_labels: Option<&BTreeMap<String, String>>,
+        sequence: u64,
+        timestamp: Timestamp,
+        text: &str,
+    ) {
+        if regex.captures(text).is_none() {
+            if let Some(unmatched_labels) = unmatched_labels {
+                let labels = unmatched_labels
+                    .iter()
+                    .map(|(name, value)| (name.clone(), MetricPrimitive::String(value.clone())))
+                    .collect();
+                self.push_bucket_value(metric_name, labels, sequence, timestamp, MetricPrimitive::Number(Number::Int(1)));
+            }
+            return;
+        }
+
+        let mut labels = BTreeMap::new();
+        for param_name in label_params {
+            let Some(resolved) = resolved_params.get(param_name) else { return };
+            let Some(parsed) = self.resolve_param_value(metric_name, resolved, &labels, text, timestamp) else {
+                return;
+            };
+            labels.insert(param_name.clone(), parsed);
+        }
+
+        if let Some(matched_labels) = matched_labels {
+            for (name, value) in matched_labels {
+                labels.insert(name.clone(), MetricPrimitive::String(value.clone()));
+            }
+        }
+
+        let value = match value_param {
+            Some(value_param) => {
+                let Some(resolved) = resolved_params.get(value_param) else { return };
+                match self.resolve_param_value(metric_name, resolved, &labels, text, timestamp) {
+                    Some(parsed) => parsed,
+                    None => return,
+                }
+            }
+            None => MetricPrimitive::Number(Number::Int(1)),
+        };
+
+        self.push_bucket_value(metric_name, labels, sequence, timestamp, value);
+    }
+
+    pub(crate) fn push_bucket_value(
+        &mut self,
+        metric_name: &str,
+        labels: BTreeMap<String, MetricPrimitive>,
+        sequence: u64,
+        timestamp: Timestamp,
+        value: MetricPrimitive,
+    ) {
+        let key = IndexKey::Bucket { metric_name: metric_name.to_string(), labels };
+        let entry = self
+            .index
+            .entry(key)
+            .or_insert_with(|| DataValue::BucketValue { values: Vec::new() });
+        if let DataValue::BucketValue { values } = entry {
+            values.push((sequence, timestamp, value));
+        }
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn backfill_metric(
+        &mut self,
+        metric_name: &str,
+        regex: &Regex,
+        label_params: &[String],
+        value_param: Option<&str>,
+        resolved_params: &HashMap<String, ResolvedParam>,
+        matched_labels: Option<&BTreeMap<String, String>>,
+        unmatched_labels: Option<&BTreeMap<String, 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,
+                regex,
+                label_params,
+                value_param,
+                resolved_params,
+                matched_labels,
+                unmatched_labels,
+                sequence,
+                timestamp,
+                &text,
+            );
+        }
+    }
+
+    pub(crate) fn apply_all_metrics(&mut self, sequence: u64, timestamp: Timestamp, text: &str) {
+        type SeededMetric = (
+            String,
+            Vec<String>,
+            Option<String>,
+            Option<BTreeMap<String, String>>,
+            Option<BTreeMap<String, String>>,
+        );
+        let metrics: Vec<SeededMetric> = self
+            .metric_declarations_range()
+            .map(|(name, _, label_params, value_param, matched_labels, unmatched_labels)| {
+                (
+                    name.to_string(),
+                    label_params.to_vec(),
+                    value_param.map(|s| s.to_string()),
+                    matched_labels.cloned(),
+                    unmatched_labels.cloned(),
+                )
+            })
+            .collect();
+        let resolved_params = self.resolved_param_declarations();
+        for (name, label_params, value_param, matched_labels, unmatched_labels) in metrics {
+            if let Some(re) = self.metric_regex_cache.get(&name).cloned() {
+                self.apply_metric_to_log(
+                    &name,
+                    &re,
+                    &label_params,
+                    value_param.as_deref(),
+                    &resolved_params,
+                    matched_labels.as_ref(),
+                    unmatched_labels.as_ref(),
+                    sequence,
+                    timestamp,
+                    text,
+                );
+            }
+        }
+    }
+
+    pub(crate) fn bucket_values<'a>(
+        &'a self,
+        metric_name: &'a str,
+        labels_filter: Option<&'a BTreeMap<String, MetricPrimitive>>,
+    ) -> impl Iterator<Item = (&'a BTreeMap<String, MetricPrimitive>, &'a (u64, Timestamp, MetricPrimitive))> + 'a
+    {
+        self.index
+            .range(IndexKey::MetricBucketSplit..IndexKey::End)
+            .filter_map(move |(key, value)| match (key, value) {
+                (IndexKey::Bucket { metric_name: name, labels }, DataValue::BucketValue { values })
+                    if name == metric_name && labels_filter.is_none_or(|filter| filter == labels) =>
+                {
+                    Some(values.iter().map(move |entry| (labels, entry)))
+                }
+                _ => None,
+            })
+            .flatten()
+    }
+}
+
+impl EngineSnapshot {
+    pub(crate) fn declared_value_kind(&self, metric_name: &str) -> Option<MetricPrimitiveKind> {
+        match self.index.get(&IndexKey::MetricDeclaration { name: metric_name.to_string() }) {
+            Some(DataValue::MetricDeclaration { value_param, .. }) => Some(
+                value_param
+                    .as_ref()
+                    .and_then(|name| {
+                        self.index.get(&IndexKey::ParamDeclaration { name: name.clone() })
+                    })
+                    .map(|value| match value {
+                        DataValue::ParamDeclaration(param) => param.kind,
+                        _ => MetricPrimitiveKind::Number,
+                    })
+                    .unwrap_or(MetricPrimitiveKind::Number),
+            ),
+            _ => None,
+        }
+    }
+}

src/engine/snapshot/mod.rs

@@ -0,0 +1,192 @@
+//! `EngineSnapshot` is the actor's live state: ingested log lines, declared params/metrics,
+//! computed metric buckets, and the sort-helper indexes built on demand — all stored in one
+//! flat `BTreeMap<IndexKey, DataValue>` and read/written through the left-right index in
+//! `super::handle`. Its impl is split across the submodules below by concern; this file holds
+//! the struct itself plus the accessors every submodule builds on.
+
+pub(crate) mod filter;
+pub(crate) mod metrics;
+pub(crate) mod params;
+pub(crate) mod query;
+pub(crate) mod sort;
+
+pub(crate) use filter::eval_json_comparison;
+
+use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration};
+use crate::engine::index_key::IndexKey;
+use crate::engine::message::{EngineQuery, EngineResponse};
+use crate::engine::timestamp::Timestamp;
+use query::handle_query_inner;
+use regex::Regex;
+use sort::{SortHelperAccess, SortHelperIndexEntry};
+use std::collections::{BTreeMap, HashMap};
+
+#[derive(Debug, Clone, Default)]
+pub(crate) struct EngineSnapshot {
+    pub(crate) index: BTreeMap<IndexKey, DataValue>,
+    pub(crate) sort_helper_indexes: HashMap<String, SortHelperIndexEntry>,
+    /// Compiled regexes for param/metric declarations, keyed by declaration name. Compiling a
+    /// regex is far more expensive than matching with one, and `apply_all_metrics`/
+    /// `update_sort_helper_indexes_on_ingest` run per ingested line, so every insertion or
+    /// removal of a declaration (see `params.rs`/`metrics.rs`) must keep these in sync rather
+    /// than recompiling from the stored pattern string on every line.
+    pub(crate) param_regex_cache: HashMap<String, Regex>,
+    pub(crate) metric_regex_cache: HashMap<String, Regex>,
+    pub(crate) log_count: usize,
+    pub(crate) approx_bytes_total: u64,
+    pub(crate) ingest_lag_high_watermark_ms: i64,
+}
+
+impl EngineSnapshot {
+    pub(crate) fn metric_declarations_range(
+        &self,
+    ) -> impl Iterator<
+        Item = (
+            &str,
+            &str,
+            &[String],
+            Option<&str>,
+            Option<&BTreeMap<String, String>>,
+            Option<&BTreeMap<String, String>>,
+        ),
+    > {
+        self.index
+            .range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit)
+            .filter_map(|(key, value)| match (key, value) {
+                (
+                    IndexKey::MetricDeclaration { name },
+                    DataValue::MetricDeclaration {
+                        message_match_regex,
+                        label_params,
+                        value_param,
+                        matched_labels,
+                        unmatched_labels,
+                    },
+                ) => Some((
+                    name.as_str(),
+                    message_match_regex.as_str(),
+                    label_params.as_slice(),
+                    value_param.as_deref(),
+                    matched_labels.as_ref(),
+                    unmatched_labels.as_ref(),
+                )),
+                _ => None,
+            })
+    }
+
+    pub(crate) fn param_declarations_range(
+        &self,
+    ) -> impl Iterator<Item = (&str, &str, &str, Option<&str>, MetricPrimitiveKind, bool, bool, bool)> {
+        self.index
+            .range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit)
+            .filter_map(|(key, value)| match (key, value) {
+                (
+                    IndexKey::ParamDeclaration { name },
+                    DataValue::ParamDeclaration(ParamDeclaration {
+                        name: _,
+                        message_match_regex,
+                        capture_group,
+                        json_field_path,
+                        kind,
+                        elapsed_since_last,
+                        auto_derived,
+                        log_stream_gated,
+                    }),
+                ) => Some((
+                    name.as_str(),
+                    message_match_regex.as_str(),
+                    capture_group.as_str(),
+                    json_field_path.as_deref(),
+                    *kind,
+                    *elapsed_since_last,
+                    *auto_derived,
+                    *log_stream_gated,
+                )),
+                _ => None,
+            })
+    }
+}
+
+pub(crate) fn approx_log_bytes(text: &str) -> u64 {
+    text.len() as u64
+}
+
+impl EngineSnapshot {
+    pub(crate) fn from_index(index: BTreeMap<IndexKey, DataValue>) -> Self {
+        let mut snapshot = EngineSnapshot { index, ..Default::default() };
+        snapshot.rebuild_regex_caches();
+        snapshot
+    }
+
+    /// Recompiles `param_regex_cache`/`metric_regex_cache` from the declarations currently in
+    /// `index`. Needed whenever a snapshot's declarations are populated by inserting straight
+    /// into `index` rather than through the normal declare/insert methods that keep the caches
+    /// in sync incrementally (e.g. `from_index` building a snapshot from a bare index).
+    pub(crate) fn rebuild_regex_caches(&mut self) {
+        self.param_regex_cache = self
+            .param_declarations_range()
+            .filter_map(|(name, message_match_regex, ..)| {
+                regex::Regex::new(message_match_regex).ok().map(|re| (name.to_string(), re))
+            })
+            .collect();
+        self.metric_regex_cache = self
+            .metric_declarations_range()
+            .filter_map(|(name, message_match_regex, ..)| {
+                regex::Regex::new(message_match_regex).ok().map(|re| (name.to_string(), re))
+            })
+            .collect();
+    }
+
+    pub(crate) fn evict_excess_logs(&mut self, max_logs: Option<usize>) {
+        let Some(max_logs) = max_logs else { return };
+        while self.log_count > max_logs {
+            let Some(oldest_key) = self
+                .index
+                .range(IndexKey::Start..IndexKey::LogParamSplit)
+                .next()
+                .map(|(key, _)| key.clone())
+            else {
+                break;
+            };
+            if let Some(DataValue::LogLiteral { text }) = self.index.get(&oldest_key) {
+                self.approx_bytes_total = self.approx_bytes_total.saturating_sub(approx_log_bytes(text));
+            }
+            self.index.remove(&oldest_key);
+            self.log_count -= 1;
+        }
+    }
+
+    pub(crate) fn logs_range(&self) -> impl Iterator<Item = (u64, Timestamp, &str)> {
+        self.index.range(IndexKey::Start..IndexKey::LogParamSplit).filter_map(|(key, value)| {
+            match (key, value) {
+                (IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text }) => {
+                    Some((*sequence, *timestamp, text.as_str()))
+                }
+                _ => None,
+            }
+        })
+    }
+
+    pub(crate) fn logs_since(
+        &self,
+        before_sequence: u64,
+        since_ms: i64,
+        now_ms: i64,
+    ) -> impl Iterator<Item = (u64, Timestamp, &str)> {
+        self.logs_range().filter(move |(sequence, timestamp, _)| {
+            *sequence < before_sequence
+                && timestamp.as_millis() > since_ms
+                && timestamp.as_millis() <= now_ms
+        })
+    }
+}
+
+impl EngineSnapshot {
+    pub(crate) fn handle_query_ro(&self, query: EngineQuery) -> EngineResponse {
+        handle_query_inner(self, query, SortHelperAccess::Immutable(&self.sort_helper_indexes))
+    }
+
+    pub(crate) fn has_sort_helper_index(&self, sort_by: &str) -> bool {
+        self.sort_helper_indexes.contains_key(sort_by)
+    }
+}

src/engine/snapshot/params.rs

@@ -0,0 +1,360 @@
+//! Param declaration: validating and storing explicit declarations, and the auto-detection
+//! logic that derives declarations from logfmt/json/log-stream-shaped lines as they're ingested.
+
+use crate::consts::{MAX_AUTO_DECLARED_PARAMS, MAX_NEW_AUTO_PARAMS_PER_LINE, OTTYL_RESERVED_PARAM_NAMES};
+use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration};
+use crate::engine::index_key::IndexKey;
+use crate::engine::metric_primitive::MetricPrimitive;
+use crate::engine::snapshot::metrics::parse_metric_primitive;
+use crate::engine::snapshot::EngineSnapshot;
+use crate::engine::timestamp::Timestamp;
+use crate::engine::DeclareParamRequest;
+use regex::Regex;
+use std::collections::{BTreeMap, HashMap};
+
+impl EngineSnapshot {
+    pub(crate) fn validate_param_declaration(&self, request: &DeclareParamRequest) -> Result<(), String> {
+        if OTTYL_RESERVED_PARAM_NAMES.contains(&request.name.as_str()) {
+            return Err(format!(
+                "'{}' is a reserved name, present on every log line without declaring anything; \
+                 it can't be redeclared as a param",
+                request.name
+            ));
+        }
+        if let Some(DataValue::ParamDeclaration(existing)) =
+            self.index.get(&IndexKey::ParamDeclaration { name: request.name.clone() })
+        {
+            if existing.auto_derived {
+                return Err(format!(
+                    "param '{}' was auto-detected from the log stream and can't be edited",
+                    request.name
+                ));
+            }
+        }
+        let re = Regex::new(&request.message_match_regex)
+            .map_err(|err| format!("invalid regex: {err}"))?;
+        if request.elapsed_since_last && request.kind != MetricPrimitiveKind::Duration {
+            return Err("elapsed_since_last params must declare kind 'duration'".to_string());
+        }
+        let uses_json_field = request.json_field_path.as_deref().is_some_and(|path| !path.is_empty());
+        if request.kind != MetricPrimitiveKind::Boolean
+            && !request.elapsed_since_last
+            && !uses_json_field
+            && !re.capture_names().flatten().any(|name| name == request.capture_group)
+        {
+            return Err(format!(
+                "regex has no capture group named '{}'",
+                request.capture_group
+            ));
+        }
+        Ok(())
+    }
+
+    pub(crate) fn insert_param_declaration(&mut self, request: DeclareParamRequest) {
+        if let Ok(re) = Regex::new(&request.message_match_regex) {
+            self.param_regex_cache.insert(request.name.clone(), re);
+        }
+        self.index.insert(
+            IndexKey::ParamDeclaration { name: request.name.clone() },
+            DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }),
+        );
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn declare_param(snapshot: &mut EngineSnapshot, request: DeclareParamRequest) -> Result<(), String> {
+    snapshot.validate_param_declaration(&request)?;
+    snapshot.insert_param_declaration(request);
+    Ok(())
+}
+
+fn sanitize_group_name(key: &str) -> String {
+    let mut group: String =
+        key.chars().map(|c| if c.is_ascii_alphanumeric() || c == '_' { c } else { '_' }).collect();
+    if group.chars().next().is_none_or(|c| c.is_ascii_digit()) {
+        group.insert(0, '_');
+    }
+    group
+}
+
+fn infer_kind(sample: &str) -> MetricPrimitiveKind {
+    if sample.parse::<i64>().is_ok() || sample.parse::<f64>().is_ok() {
+        MetricPrimitiveKind::Number
+    } else {
+        MetricPrimitiveKind::String
+    }
+}
+
+fn build_auto_param_pattern(key: &str, group: &str, quoted: bool) -> String {
+    let escaped_key = regex::escape(key);
+    if quoted {
+        format!(r#"{escaped_key}="(?P<{group}>[^"]*)""#)
+    } else {
+        format!(r"{escaped_key}=(?P<{group}>\S+)")
+    }
+}
+
+
+struct AutoParamCandidate {
+    name: String,
+    value: String,
+    message_match_regex: String,
+    capture_group: String,
+    json_field_path: Option<String>,
+}
+
+impl EngineSnapshot {
+    fn auto_derived_param_count(&self) -> usize {
+        self.param_declarations_range().filter(|declaration| declaration.6).count()
+    }
+
+    pub(crate) fn auto_declare_params(&mut self, text: &str) {
+        let mut total = self.auto_derived_param_count();
+        if total >= MAX_AUTO_DECLARED_PARAMS {
+            return;
+        }
+        let candidates: Vec<AutoParamCandidate> = if crate::engine::logfmt::detect_format(text) == "json" {
+            crate::engine::data_value::flatten_json_object(text)
+                .into_iter()
+                .map(|(path, value)| AutoParamCandidate {
+                    name: path.clone(),
+                    value,
+                    message_match_regex: String::new(),
+                    capture_group: String::new(),
+                    json_field_path: Some(path),
+                })
+                .collect()
+        } else {
+            crate::engine::logfmt::extract_pairs(text)
+                .into_iter()
+                .map(|(key, value, quoted)| {
+                    let group = sanitize_group_name(key);
+                    AutoParamCandidate {
+                        name: key.to_string(),
+                        value: value.to_string(),
+                        message_match_regex: build_auto_param_pattern(key, &group, quoted),
+                        capture_group: group,
+                        json_field_path: None,
+                    }
+                })
+                .collect()
+        };
+
+        let mut new_this_line = 0;
+        for candidate in candidates {
+            if new_this_line >= MAX_NEW_AUTO_PARAMS_PER_LINE || total >= MAX_AUTO_DECLARED_PARAMS {
+                break;
+            }
+            let index_key = IndexKey::ParamDeclaration { name: candidate.name.clone() };
+            if self.index.contains_key(&index_key) {
+                continue;
+            }
+            if let Ok(re) = Regex::new(&candidate.message_match_regex) {
+                self.param_regex_cache.insert(candidate.name.clone(), re);
+            }
+            self.index.insert(
+                index_key,
+                DataValue::ParamDeclaration(ParamDeclaration {
+                    name: candidate.name,
+                    message_match_regex: candidate.message_match_regex,
+                    capture_group: candidate.capture_group,
+                    json_field_path: candidate.json_field_path,
+                    kind: infer_kind(&candidate.value),
+                    elapsed_since_last: false,
+                    auto_derived: true,
+                    log_stream_gated: false,
+                }),
+            );
+            new_this_line += 1;
+            total += 1;
+        }
+    }
+}
+
+pub(crate) fn log_stream_header_transition(
+    current: &Option<crate::engine::logfmt::LogStreamHeader>,
+    text: &str,
+) -> Option<crate::engine::logfmt::LogStreamHeader> {
+    crate::engine::logfmt::LogStreamHeader::detect(text).or_else(|| current.clone())
+}
+
+impl EngineSnapshot {
+    pub(crate) fn derive_log_stream_params_from_row(
+        &mut self,
+        header: &crate::engine::logfmt::LogStreamHeader,
+        text: &str,
+    ) {
+        if !crate::engine::logfmt::is_log_stream_line(text) {
+            return;
+        }
+        let row = header.split_row(text);
+
+        for (i, column) in header.columns.iter().enumerate() {
+            let Some(value) = row.get(i) else { continue };
+            let index_key = IndexKey::ParamDeclaration { name: column.name.clone() };
+            if self.index.contains_key(&index_key) {
+                continue;
+            }
+            let message_match_regex = header.column_pattern(i, &column.name);
+            if let Ok(re) = Regex::new(&message_match_regex) {
+                self.param_regex_cache.insert(column.name.clone(), re);
+            }
+            self.index.insert(
+                index_key,
+                DataValue::ParamDeclaration(ParamDeclaration {
+                    name: column.name.clone(),
+                    message_match_regex,
+                    capture_group: column.name.clone(),
+                    json_field_path: None,
+                    kind: infer_kind(value.trim()),
+                    elapsed_since_last: false,
+                    auto_derived: true,
+                    log_stream_gated: true,
+                }),
+            );
+        }
+
+        if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) {
+            let index_key = IndexKey::ParamDeclaration { name: "message".to_string() };
+            if !self.index.contains_key(&index_key) {
+                let message_match_regex = header.message_pattern("message");
+                if let Ok(re) = Regex::new(&message_match_regex) {
+                    self.param_regex_cache.insert("message".to_string(), re);
+                }
+                self.index.insert(
+                    index_key,
+                    DataValue::ParamDeclaration(ParamDeclaration {
+                        name: "message".to_string(),
+                        message_match_regex,
+                        capture_group: "message".to_string(),
+                        json_field_path: None,
+                        kind: infer_kind(message.trim()),
+                        elapsed_since_last: false,
+                        auto_derived: true,
+                        log_stream_gated: true,
+                    }),
+                );
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn auto_declare_log_stream_params(
+    snapshot: &mut EngineSnapshot,
+    header: &mut Option<crate::engine::logfmt::LogStreamHeader>,
+    text: &str,
+) {
+    let is_new_header = crate::engine::logfmt::LogStreamHeader::detect(text).is_some();
+    *header = log_stream_header_transition(header, text);
+    if is_new_header {
+        return;
+    }
+    if let Some(header) = header.clone() {
+        snapshot.derive_log_stream_params_from_row(&header, text);
+    }
+}
+
+impl EngineSnapshot {
+    pub(crate) fn clear_all_param_declarations(&mut self) -> usize {
+        let names: Vec<String> =
+            self.param_declarations_range().map(|(name, ..)| name.to_string()).collect();
+        let removed = names.len();
+        for name in names {
+            self.index.remove(&IndexKey::ParamDeclaration { name: name.clone() });
+            self.param_regex_cache.remove(&name);
+        }
+        removed
+    }
+
+    pub(crate) fn reextract_params_from_retained_logs(&mut self) -> (usize, usize) {
+        let texts: Vec<String> = self.logs_range().map(|(_, _, text)| text.to_string()).collect();
+        let logs_considered = texts.len();
+        for text in &texts {
+            self.auto_declare_params(text);
+        }
+        let reextracted = self.param_declarations_range().count();
+        (reextracted, logs_considered)
+    }
+}
+
+pub(super) struct ResolvedParam {
+    regex: Regex,
+    capture_group: String,
+    json_field_path: Option<String>,
+    kind: MetricPrimitiveKind,
+    elapsed_since_last: bool,
+    log_stream_gated: bool,
+}
+
+impl EngineSnapshot {
+    pub(super) fn resolved_param_declarations(&self) -> HashMap<String, ResolvedParam> {
+        self.param_declarations_range()
+            .filter_map(|(name, _, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| {
+                self.param_regex_cache.get(name).map(|regex| {
+                    (
+                        name.to_string(),
+                        ResolvedParam {
+                            regex: regex.clone(),
+                            capture_group: capture_group.to_string(),
+                            json_field_path: json_field_path.map(|path| path.to_string()),
+                            kind,
+                            elapsed_since_last,
+                            log_stream_gated,
+                        },
+                    )
+                })
+            })
+            .collect()
+    }
+}
+
+fn extract_param_raw(
+    text: &str,
+    regex: &Regex,
+    capture_group: &str,
+    json_field_path: Option<&str>,
+    log_stream_gated: bool,
+) -> Option<String> {
+    if log_stream_gated && !crate::engine::logfmt::is_log_stream_line(text) {
+        return None;
+    }
+    match json_field_path.filter(|path| !path.is_empty()) {
+        Some(field_path) => crate::engine::data_value::extract_json_field(text, field_path),
+        None => regex.captures(text)?.name(capture_group).map(|m| m.as_str().to_string()),
+    }
+}
+
+impl EngineSnapshot {
+    pub(super) fn resolve_param_value(
+        &mut self,
+        metric_name: &str,
+        resolved: &ResolvedParam,
+        labels_so_far: &BTreeMap<String, MetricPrimitive>,
+        text: &str,
+        timestamp: Timestamp,
+    ) -> Option<MetricPrimitive> {
+        if resolved.elapsed_since_last {
+            let last_event_key = IndexKey::LastMetricEvent {
+                metric_name: metric_name.to_string(),
+                labels: labels_so_far.clone(),
+            };
+            let previous_timestamp = match self.index.get(&last_event_key) {
+                Some(DataValue::LastEventTimestamp { timestamp: previous }) => Some(*previous),
+                _ => None,
+            };
+            self.index.insert(last_event_key, DataValue::LastEventTimestamp { timestamp });
+            return previous_timestamp
+                .map(|previous| MetricPrimitive::Duration(timestamp.as_millis() - previous.as_millis()));
+        }
+
+        let raw = extract_param_raw(
+            text,
+            &resolved.regex,
+            &resolved.capture_group,
+            resolved.json_field_path.as_deref(),
+            resolved.log_stream_gated,
+        )?;
+        parse_metric_primitive(&raw, resolved.kind)
+    }
+}

src/engine/snapshot/query.rs

@@ -0,0 +1,549 @@
+//! Answering `EngineQuery`s: log/timeline listing (including the sort-helper-index path),
+//! stats, declaration listing, bucket-value listing, and PromQL evaluation.
+
+use crate::api::dto::{
+    BucketValueDto, LogEntryDto, MetricDeclarationDto, ParamDeclarationDto, PromqlSeriesDto,
+    StatsDto, TimelineBucketDto,
+};
+use crate::consts::{DEFAULT_TIMELINE_BUCKET_MS, MERGE_WINDOW_MS};
+use crate::engine::data_value::{DataValue, MetricPrimitiveKind};
+use crate::engine::index_key::IndexKey;
+use crate::engine::message::{EngineQuery, EngineResponse};
+use crate::engine::metric_primitive::{MetricPrimitive, Number};
+use crate::engine::promql::{AggOp, LabelMatcher, PromqlQuery};
+use crate::engine::rate;
+#[cfg(test)]
+use crate::engine::snapshot::sort::SortHelperIndexEntry;
+use crate::engine::snapshot::sort::{log_entry_params, SortHelperAccess, SortHelperKey};
+use crate::engine::snapshot::EngineSnapshot;
+use crate::engine::timestamp::Timestamp;
+#[cfg(test)]
+use std::collections::HashMap;
+use std::collections::BTreeMap;
+
+#[cfg(test)]
+pub(crate) fn handle_query(
+    snapshot: &EngineSnapshot,
+    query: EngineQuery,
+    _approx_bytes_total: u64,
+    _ingest_lag_high_watermark_ms: i64,
+    sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>,
+) -> EngineResponse {
+    handle_query_inner(snapshot, query, SortHelperAccess::Mutable(sort_helper_indexes))
+}
+
+pub(super) fn handle_query_inner(
+    snapshot: &EngineSnapshot,
+    query: EngineQuery,
+    sort_helper_access: SortHelperAccess<'_>,
+) -> EngineResponse {
+    let index = &snapshot.index;
+    let approx_bytes_total = snapshot.approx_bytes_total;
+    let ingest_lag_high_watermark_ms = snapshot.ingest_lag_high_watermark_ms;
+    match query {
+        EngineQuery::Logs(params) => {
+            let compiled_filter = match params.filter.as_deref() {
+                Some(filter) => match snapshot.compile_filter_query(filter, Timestamp::now()) {
+                    Ok(compiled) => Some(compiled),
+                    Err(err) => return EngineResponse::Error(err),
+                },
+                None => None,
+            };
+            if let Some(sort_by) = params.sort_by.as_deref() {
+                if !snapshot.param_declarations_range().any(|(name, ..)| name == sort_by) {
+                    return EngineResponse::Error(format!("unknown param '{sort_by}' for sort_by"));
+                }
+            }
+            let limit = params.limit.unwrap_or(500).min(10_000);
+            let compiled_params = snapshot.compiled_param_declarations();
+            let filtered = snapshot
+                .logs_range()
+                .filter(|(sequence, timestamp, _)| {
+                    params.since_sequence.is_none_or(|since| *sequence > since)
+                        && params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
+                        && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
+                })
+                .filter(|(sequence, timestamp, text)| {
+                    compiled_filter
+                        .as_ref()
+                        .is_none_or(|filter| filter.eval(*sequence, *timestamp, text))
+                });
+
+            let entries: Vec<LogEntryDto> = match params.sort_by.as_deref() {
+                None => filtered
+                    .take(limit)
+                    .map(|(sequence, timestamp, text)| LogEntryDto {
+                        sequence,
+                        timestamp_ms: timestamp.as_millis(),
+                        text: text.to_string(),
+                        params: log_entry_params(text, &compiled_params),
+                    })
+                    .collect(),
+                Some(sort_by) => {
+                    static EMPTY_HELPER: std::sync::LazyLock<BTreeMap<SortHelperKey, Timestamp>> =
+                        std::sync::LazyLock::new(BTreeMap::new);
+                    let helper_index =
+                        sort_helper_access.resolve(index, sort_by).unwrap_or(&EMPTY_HELPER);
+
+                    let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true);
+                    let mut entries = Vec::new();
+                    for (helper_key, timestamp) in helper_index.iter() {
+                        let sequence = match helper_key {
+                            SortHelperKey::WithParam(_, sequence) => *sequence,
+                            SortHelperKey::WithoutParam(sequence) => {
+                                if !sort_missing_fallback {
+                                    continue;
+                                }
+                                *sequence
+                            }
+                        };
+                        if !(params.since_sequence.is_none_or(|since| sequence > since)
+                            && params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
+                            && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end))
+                        {
+                            continue;
+                        }
+                        let Some(DataValue::LogLiteral { text }) =
+                            index.get(&IndexKey::IngestedLog { sequence, timestamp: *timestamp })
+                        else {
+                            continue;
+                        };
+                        if !compiled_filter
+                            .as_ref()
+                            .is_none_or(|filter| filter.eval(sequence, *timestamp, text))
+                        {
+                            continue;
+                        }
+                        entries.push(LogEntryDto {
+                            sequence,
+                            timestamp_ms: timestamp.as_millis(),
+                            text: text.clone(),
+                            params: log_entry_params(text, &compiled_params),
+                        });
+                        if entries.len() >= limit {
+                            break;
+                        }
+                    }
+                    entries
+                }
+            };
+            let entries: Vec<LogEntryDto> = entries.into_iter().take(limit).collect();
+            EngineResponse::Logs(entries)
+        }
+        EngineQuery::Timeline(params) => {
+            let compiled_filter = match params.filter.as_deref() {
+                Some(filter) => match snapshot.compile_filter_query(filter, Timestamp::now()) {
+                    Ok(compiled) => Some(compiled),
+                    Err(err) => return EngineResponse::Error(err),
+                },
+                None => None,
+            };
+            let bucket_ms = params.bucket_ms.unwrap_or(DEFAULT_TIMELINE_BUCKET_MS).max(1);
+            let mut buckets: BTreeMap<i64, u64> = BTreeMap::new();
+            for (sequence, timestamp, text) in snapshot.logs_range().filter(|(_, timestamp, _)| {
+                params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
+                    && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
+            }) {
+                if compiled_filter
+                    .as_ref()
+                    .is_some_and(|filter| !filter.eval(sequence, timestamp, text))
+                {
+                    continue;
+                }
+                let bucket_start = (timestamp.as_millis() / bucket_ms) * bucket_ms;
+                *buckets.entry(bucket_start).or_insert(0) += 1;
+            }
+            let out = buckets
+                .into_iter()
+                .map(|(start_ms, count)| TimelineBucketDto { start_ms, count })
+                .collect();
+            EngineResponse::Timeline(out)
+        }
+        EngineQuery::Stats => {
+            let mut log_count = 0u64;
+            let mut first_sequence = None;
+            let mut last_sequence = None;
+            let mut first_timestamp_ms = None;
+            let mut last_timestamp_ms = None;
+            for (sequence, timestamp, _) in snapshot.logs_range() {
+                log_count += 1;
+                first_sequence.get_or_insert(sequence);
+                last_sequence = Some(sequence);
+                first_timestamp_ms.get_or_insert(timestamp.as_millis());
+                last_timestamp_ms = Some(timestamp.as_millis());
+            }
+            EngineResponse::Stats(StatsDto {
+                log_count,
+                first_sequence,
+                last_sequence,
+                first_timestamp_ms,
+                last_timestamp_ms,
+                approx_bytes_total,
+                ingest_lag_high_watermark_ms,
+            })
+        }
+        EngineQuery::ParamDeclarations => {
+            let declarations = snapshot.param_declarations_range()
+                .map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, auto_derived, _)| {
+                    ParamDeclarationDto {
+                        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(|path| path.to_string()),
+                        kind,
+                        elapsed_since_last,
+                        auto_derived,
+                        reserved: false,
+                    }
+                })
+                .collect();
+            EngineResponse::ParamDeclarations(declarations)
+        }
+        EngineQuery::MetricDeclarations => {
+            let declarations = snapshot.metric_declarations_range()
+                .map(|(name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels)| {
+                    MetricDeclarationDto {
+                        name: name.to_string(),
+                        message_match_regex: message_match_regex.to_string(),
+                        label_params: label_params.to_vec(),
+                        value_param: value_param.map(|s| s.to_string()),
+                        matched_labels: matched_labels.cloned(),
+                        unmatched_labels: unmatched_labels.cloned(),
+                    }
+                })
+                .collect();
+            EngineResponse::MetricDeclarations(declarations)
+        }
+        EngineQuery::BucketValues {
+            metric_name,
+            labels_filter,
+            start_ms,
+            end_ms,
+            since_sequence,
+            limit,
+        } => {
+            let limit = limit.unwrap_or(500).min(10_000);
+            let values: Vec<BucketValueDto> = snapshot.bucket_values(&metric_name, None)
+                .filter(|(labels, _)| {
+                    labels_filter.as_ref().is_none_or(|filter| {
+                        filter.iter().all(|(name, expected)| {
+                            labels
+                                .get(name)
+                                .is_some_and(|value| &value.to_string() == expected)
+                        })
+                    })
+                })
+                .filter(|(_, (sequence, timestamp, _))| {
+                    since_sequence.is_none_or(|since| *sequence > since)
+                        && start_ms.is_none_or(|start| timestamp.as_millis() >= start)
+                        && end_ms.is_none_or(|end| timestamp.as_millis() <= end)
+                })
+                .take(limit)
+                .map(|(labels, (sequence, timestamp, value))| BucketValueDto {
+                    labels: labels.clone(),
+                    sequence: *sequence,
+                    timestamp_ms: timestamp.as_millis(),
+                    value: value.clone(),
+                })
+                .collect();
+            EngineResponse::BucketValues(values)
+        }
+        EngineQuery::Promql { query, start_ms, end_ms } => {
+            let response = PromqlQuery::parse(&query)
+                .and_then(|parsed| snapshot.evaluate_promql(&parsed, start_ms, end_ms));
+            EngineResponse::Promql(response)
+        }
+    }
+}
+
+fn coerce_to_f64(value: &MetricPrimitive) -> f64 {
+    match value {
+        MetricPrimitive::Number(n) => match n {
+            Number::Int(v) => *v as f64,
+            Number::Float(v) => *v,
+        },
+        MetricPrimitive::Timestamp(ts) => ts.as_millis() as f64,
+        MetricPrimitive::Duration(ms) => *ms as f64,
+        MetricPrimitive::String(_) => 1.0,
+        MetricPrimitive::Boolean(b) => {
+            if *b {
+                1.0
+            } else {
+                0.0
+            }
+        }
+    }
+}
+
+fn matches_label_matchers(
+    labels: &BTreeMap<String, MetricPrimitive>,
+    matchers: &[LabelMatcher],
+) -> bool {
+    matchers.iter().all(|matcher| {
+        labels.get(&matcher.name).is_some_and(|value| value.to_string() == matcher.value)
+    })
+}
+
+fn project_labels(
+    labels: &BTreeMap<String, String>,
+    by_labels: &[String],
+) -> BTreeMap<String, String> {
+    by_labels
+        .iter()
+        .filter_map(|name| labels.get(name).map(|value| (name.clone(), value.clone())))
+        .collect()
+}
+
+fn windowed_rate_series(
+    points: &[(i64, f64)],
+    range_ms: i64,
+    start_ms: Option<i64>,
+    end_ms: Option<i64>,
+) -> Vec<(i64, f64)> {
+    let ts_points: Vec<(Timestamp, f64)> =
+        points.iter().map(|(ms, value)| (Timestamp::from_millis(*ms), *value)).collect();
+    let mut series: Vec<(i64, f64)> = rate::windowed_rate_series(&ts_points, range_ms)
+        .into_iter()
+        .map(|(ts, value)| (ts.as_millis(), value))
+        .collect();
+
+    if let Some(end) = end_ms {
+        if series.last().is_none_or(|(ts, _)| *ts < end) {
+            let value = rate::rate_at(&ts_points, Timestamp::from_millis(end), range_ms);
+            series.push((end, value));
+        }
+    }
+    if let Some(start) = start_ms {
+        if series.first().is_none_or(|(ts, _)| *ts > start) {
+            let value = rate::rate_at(&ts_points, Timestamp::from_millis(start), range_ms);
+            series.insert(0, (start, value));
+        }
+    }
+    series
+}
+
+fn merge_series_windowed(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> {
+    if sub_series.len() <= 1 {
+        return sub_series.first().cloned().unwrap_or_default();
+    }
+    let mut grouped: BTreeMap<i64, Vec<f64>> = BTreeMap::new();
+    for points in sub_series {
+        for (timestamp_ms, value) in points {
+            let window_start = timestamp_ms.div_euclid(MERGE_WINDOW_MS) * MERGE_WINDOW_MS;
+            grouped.entry(window_start).or_default().push(*value);
+        }
+    }
+    grouped
+        .into_iter()
+        .map(|(timestamp_ms, values)| {
+            let sum: f64 = values.iter().sum();
+            let value = if is_avg { sum / values.len() as f64 } else { sum };
+            (timestamp_ms, value)
+        })
+        .collect()
+}
+
+fn merge_series_locf(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> {
+    let mut all_timestamps: std::collections::BTreeSet<i64> = std::collections::BTreeSet::new();
+    for points in sub_series {
+        all_timestamps.extend(points.iter().map(|(ts, _)| *ts));
+    }
+    let mut cursors = vec![0usize; sub_series.len()];
+    let mut last_values = vec![0.0f64; sub_series.len()];
+    all_timestamps
+        .into_iter()
+        .map(|ts| {
+            for (i, points) in sub_series.iter().enumerate() {
+                while cursors[i] < points.len() && points[cursors[i]].0 <= ts {
+                    last_values[i] = points[cursors[i]].1;
+                    cursors[i] += 1;
+                }
+            }
+            let sum: f64 = last_values.iter().sum();
+            let value = if is_avg { sum / last_values.len() as f64 } else { sum };
+            (ts, value)
+        })
+        .collect()
+}
+
+fn ottyl_reserved_numeric_value_fn(metric_name: &str) -> Option<fn(u64, i64) -> f64> {
+    match metric_name {
+        "ottyl_seq" => Some(|sequence, _timestamp_ms| sequence as f64),
+        "ottyl_ts" => Some(|_sequence, timestamp_ms| timestamp_ms as f64),
+        _ => None,
+    }
+}
+
+impl EngineSnapshot {
+    pub(crate) fn evaluate_promql(
+        &self,
+        parsed: &PromqlQuery,
+        start_ms: Option<i64>,
+        end_ms: Option<i64>,
+    ) -> Result<Vec<PromqlSeriesDto>, String> {
+        if parsed.metric_name == "ottyl_fmt" {
+            return Err(
+                "'ottyl_fmt' is a categorical value (json/logfmt/none), not a number, so it can't be \
+                 used in a dashboard PromQL query; filter on it in the search bar instead, e.g. \
+                 ottyl_fmt==json"
+                    .to_string(),
+            );
+        }
+
+        if ottyl_reserved_numeric_value_fn(&parsed.metric_name).is_none()
+            && !self.index.contains_key(&IndexKey::MetricDeclaration { name: parsed.metric_name.clone() })
+            && self.bucket_values(&parsed.metric_name, None).next().is_none()
+        {
+            return Err(format!(
+                "unknown metric \"{}\": no declaration or recorded data for it (was it deleted?)",
+                parsed.metric_name
+            ));
+        }
+        if matches!(parsed.agg, Some(AggOp::Sum) | Some(AggOp::Avg))
+            && !parsed.inner_rate
+            && self.declared_value_kind(&parsed.metric_name) == Some(MetricPrimitiveKind::String)
+        {
+            return Err(format!(
+                "cannot sum/avg a string-typed metric \"{}\"; use count() or a numeric-typed metric",
+                parsed.metric_name
+            ));
+        }
+
+        let mut series_points: BTreeMap<BTreeMap<String, String>, Vec<(i64, f64)>> = BTreeMap::new();
+        if let Some(reserved_value) = ottyl_reserved_numeric_value_fn(&parsed.metric_name) {
+            let no_labels: BTreeMap<String, MetricPrimitive> = BTreeMap::new();
+            if matches_label_matchers(&no_labels, &parsed.matchers) {
+                for (sequence, timestamp, _text) in self.logs_range() {
+                    let timestamp_ms = timestamp.as_millis();
+                    if start_ms.is_none_or(|start| timestamp_ms >= start)
+                        && end_ms.is_none_or(|end| timestamp_ms <= end)
+                    {
+                        series_points
+                            .entry(BTreeMap::new())
+                            .or_default()
+                            .push((timestamp_ms, reserved_value(sequence, timestamp_ms)));
+                    }
+                }
+            }
+        } else {
+            for (labels, (_, timestamp, value)) in self.bucket_values(&parsed.metric_name, None)
+                .filter(|(labels, _)| matches_label_matchers(labels, &parsed.matchers))
+                .filter(|(_, (_, timestamp, _))| {
+                    start_ms.is_none_or(|start| timestamp.as_millis() >= start)
+                        && end_ms.is_none_or(|end| timestamp.as_millis() <= end)
+                })
+            {
+                let display_labels: BTreeMap<String, String> =
+                    labels.iter().map(|(name, value)| (name.clone(), value.to_string())).collect();
+                series_points
+                    .entry(display_labels)
+                    .or_default()
+                    .push((timestamp.as_millis(), coerce_to_f64(value)));
+            }
+        }
+    for points in series_points.values_mut() {
+        points.sort_by_key(|(timestamp_ms, _)| *timestamp_ms);
+    }
+
+    Ok(match parsed.agg {
+        None => series_points
+            .into_iter()
+            .map(|(labels, points)| PromqlSeriesDto { labels, points })
+            .collect(),
+        Some(AggOp::Count) => {
+            let mut groups: BTreeMap<BTreeMap<String, String>, Vec<Vec<(i64, f64)>>> = BTreeMap::new();
+            for (labels, points) in series_points {
+                let count_points: Vec<(i64, f64)> =
+                    points.iter().map(|(timestamp_ms, _)| (*timestamp_ms, 1.0)).collect();
+                let group_key = project_labels(&labels, &parsed.by_labels);
+                groups.entry(group_key).or_default().push(count_points);
+            }
+            groups
+                .into_iter()
+                .map(|(group_labels, sub_series)| {
+                    let points = merge_series_windowed(&sub_series, false);
+                    PromqlSeriesDto { labels: group_labels, points }
+                })
+                .collect()
+        }
+        Some(AggOp::Sum) | Some(AggOp::Avg) => {
+            let is_avg = parsed.agg == Some(AggOp::Avg);
+            let rate_range_ms = parsed
+                .inner_rate
+                .then(|| parsed.range.expect("inner rate() always carries a range").as_millis() as i64);
+
+            let mut groups: BTreeMap<BTreeMap<String, String>, Vec<Vec<(i64, f64)>>> = BTreeMap::new();
+            for (labels, points) in series_points {
+                let points = match rate_range_ms {
+                    Some(range_ms) => windowed_rate_series(&points, range_ms, start_ms, end_ms),
+                    None => points,
+                };
+                let group_key = project_labels(&labels, &parsed.by_labels);
+                groups.entry(group_key).or_default().push(points);
+            }
+
+            groups
+                .into_iter()
+                .map(|(group_labels, sub_series)| {
+                    let points = if rate_range_ms.is_some() {
+                        merge_series_locf(&sub_series, is_avg)
+                    } else {
+                        merge_series_windowed(&sub_series, is_avg)
+                    };
+                    PromqlSeriesDto { labels: group_labels, points }
+                })
+                .collect()
+        }
+        Some(AggOp::Rate) => {
+            let range_ms = parsed.range.expect("rate() always carries a range").as_millis() as i64;
+            series_points
+                .into_iter()
+                .map(|(labels, points)| PromqlSeriesDto {
+                    labels,
+                    points: windowed_rate_series(&points, range_ms, start_ms, end_ms),
+                })
+                .collect()
+        }
+        Some(AggOp::Gauge) => {
+            series_points
+                .into_iter()
+                .filter_map(|(labels, points)| {
+                    let last = *points.last()?;
+                    Some(PromqlSeriesDto { labels, points: vec![last] })
+                })
+                .collect()
+        }
+        Some(AggOp::Histogram) => {
+            let bucket_width =
+                parsed.bucket_width.expect("histogram() always carries a bucket width");
+            let last_ts = series_points
+                .values()
+                .flat_map(|points| points.iter().map(|(ts, _)| *ts))
+                .max()
+                .unwrap_or(0);
+            let mut groups: BTreeMap<BTreeMap<String, String>, Vec<&(i64, f64)>> = BTreeMap::new();
+            for (labels, points) in &series_points {
+                let group_key = project_labels(labels, &parsed.by_labels);
+                groups.entry(group_key).or_default().extend(points.iter());
+            }
+
+            groups
+                .into_iter()
+                .flat_map(|(group_labels, points)| {
+                    let mut counts: BTreeMap<i64, u64> = BTreeMap::new();
+                    for (_, value) in points {
+                        let bucket_index = (value / bucket_width).floor() as i64;
+                        *counts.entry(bucket_index).or_insert(0) += 1;
+                    }
+                    counts.into_iter().map(move |(bucket_index, count)| {
+                        let upper_bound = (bucket_index + 1) as f64 * bucket_width;
+                        let mut labels = group_labels.clone();
+                        labels.insert("le".to_string(), upper_bound.to_string());
+                        PromqlSeriesDto { labels, points: vec![(last_ts, count as f64)] }
+                    })
+                })
+                .collect()
+        }
+        })
+    }
+}

src/engine/snapshot/sort.rs

@@ -0,0 +1,183 @@
+//! Sort-helper indexes: a per-param `sort_by` lets the logs view return results ordered by a
+//! declared param's value instead of ingest order. Building that ordering from scratch is
+//! O(n log n) over every log, so it's memoized in `EngineSnapshot::sort_helper_indexes` and
+//! kept incrementally up to date on ingest, rather than rebuilt per query.
+
+use crate::consts::SORT_HELPER_INDEX_PURGE_AFTER;
+use crate::engine::data_value::DataValue;
+use crate::engine::index_key::IndexKey;
+use crate::engine::data_value::MetricPrimitiveKind;
+use crate::engine::metric_primitive::MetricPrimitive;
+use crate::engine::snapshot::metrics::parse_metric_primitive;
+use crate::engine::snapshot::EngineSnapshot;
+use crate::engine::timestamp::Timestamp;
+use regex::Regex;
+use std::collections::{BTreeMap, HashMap};
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
+pub(crate) enum SortHelperKey {
+    WithParam(MetricPrimitive, u64),
+    WithoutParam(u64),
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct SortHelperIndexEntry {
+    pub(crate) index: BTreeMap<SortHelperKey, Timestamp>,
+    pub(crate) last_accessed: std::time::Instant,
+}
+
+pub(super) type CompiledParamDeclaration = (String, Regex, String, Option<String>, MetricPrimitiveKind, bool);
+
+fn derive_single_param(
+    text: &str,
+    declarations: &[CompiledParamDeclaration],
+    param_name: &str,
+) -> Option<MetricPrimitive> {
+    let (_, regex, capture_group, json_field_path, kind, log_stream_gated) =
+        declarations.iter().find(|(name, ..)| name == param_name)?;
+    if *log_stream_gated && !crate::engine::logfmt::is_log_stream_line(text) {
+        return None;
+    }
+    if *kind == MetricPrimitiveKind::Boolean {
+        return Some(MetricPrimitive::Boolean(regex.is_match(text)));
+    }
+    let captures = regex.captures(text)?;
+    let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) {
+        Some(field_path) => crate::engine::data_value::extract_json_field(text, field_path)?,
+        None => captures.name(capture_group)?.as_str().to_string(),
+    };
+    parse_metric_primitive(&raw, *kind)
+}
+
+impl EngineSnapshot {
+    pub(super) fn build_sort_helper_index(&self, sort_by: &str) -> BTreeMap<SortHelperKey, Timestamp> {
+        let compiled_params = self.compiled_param_declarations();
+        self.logs_range()
+            .map(|(sequence, timestamp, text)| {
+                let key = match derive_single_param(text, &compiled_params, sort_by) {
+                    Some(value) => SortHelperKey::WithParam(value, sequence),
+                    None => SortHelperKey::WithoutParam(sequence),
+                };
+                (key, timestamp)
+            })
+            .collect()
+    }
+
+    pub(crate) fn update_sort_helper_indexes_on_ingest(&mut self, sequence: u64, timestamp: Timestamp, text: &str) {
+        if self.sort_helper_indexes.is_empty() {
+            return;
+        }
+        let compiled_params = self.compiled_param_declarations();
+        for (sort_by, entry) in self.sort_helper_indexes.iter_mut() {
+            let key = match derive_single_param(text, &compiled_params, sort_by) {
+                Some(value) => SortHelperKey::WithParam(value, sequence),
+                None => SortHelperKey::WithoutParam(sequence),
+            };
+            entry.index.insert(key, timestamp);
+        }
+    }
+
+    pub(crate) fn purge_stale_sort_helper_indexes(&mut self) {
+        self.sort_helper_indexes
+            .retain(|_, entry| entry.last_accessed.elapsed() < SORT_HELPER_INDEX_PURGE_AFTER);
+    }
+
+    pub(crate) fn ensure_sort_helper_index_built(&mut self, sort_by: &str) {
+        if !self.sort_helper_indexes.contains_key(sort_by) {
+            let built = self.build_sort_helper_index(sort_by);
+            self.sort_helper_indexes.insert(
+                sort_by.to_string(),
+                SortHelperIndexEntry { index: built, last_accessed: std::time::Instant::now() },
+            );
+        }
+        let entry = self.sort_helper_indexes.get_mut(sort_by).expect("just inserted above if missing");
+        entry.last_accessed = std::time::Instant::now();
+    }
+
+    pub(crate) fn compiled_param_declarations(&self) -> Vec<CompiledParamDeclaration> {
+        self.param_declarations_range()
+            .filter(|(_, _, _, _, _, elapsed_since_last, _, _)| !elapsed_since_last)
+            .filter_map(|(name, _, capture_group, json_field_path, kind, _, _, log_stream_gated)| {
+                self.param_regex_cache.get(name).map(|re| {
+                    (
+                        name.to_string(),
+                        re.clone(),
+                        capture_group.to_string(),
+                        json_field_path.map(|path| path.to_string()),
+                        kind,
+                        log_stream_gated,
+                    )
+                })
+            })
+            .collect()
+    }
+}
+
+pub(crate) fn derive_log_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap<String, MetricPrimitive> {
+    let mut params = BTreeMap::new();
+    for (name, regex, capture_group, json_field_path, kind, log_stream_gated) in declarations {
+        if *log_stream_gated && !crate::engine::logfmt::is_log_stream_line(text) {
+            continue;
+        }
+        if *kind == MetricPrimitiveKind::Boolean {
+            params.insert(name.clone(), MetricPrimitive::Boolean(regex.is_match(text)));
+            continue;
+        }
+        let Some(captures) = regex.captures(text) else {
+            continue;
+        };
+        let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) {
+            Some(field_path) => match crate::engine::data_value::extract_json_field(text, field_path) {
+                Some(value) => value,
+                None => continue,
+            },
+            None => match captures.name(capture_group) {
+                Some(m) => m.as_str().to_string(),
+                None => continue,
+            },
+        };
+        let Some(value) = parse_metric_primitive(&raw, *kind) else {
+            continue;
+        };
+        params.insert(name.clone(), value);
+    }
+    params
+}
+
+pub(super) fn log_entry_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap<String, MetricPrimitive> {
+    let mut params = derive_log_params(text, declarations);
+    params.insert("ottyl_fmt".to_string(), MetricPrimitive::String(crate::engine::logfmt::detect_format(text).to_string()));
+    params
+}
+
+pub(super) enum SortHelperAccess<'a> {
+    #[cfg(test)]
+    Mutable(&'a mut HashMap<String, SortHelperIndexEntry>),
+    Immutable(&'a HashMap<String, SortHelperIndexEntry>),
+}
+
+impl<'a> SortHelperAccess<'a> {
+    #[cfg_attr(not(test), allow(unused_variables))]
+    pub(super) fn resolve(
+        self,
+        index: &BTreeMap<IndexKey, DataValue>,
+        sort_by: &str,
+    ) -> Option<&'a BTreeMap<SortHelperKey, Timestamp>> {
+        match self {
+            #[cfg(test)]
+            SortHelperAccess::Mutable(map) => {
+                if !map.contains_key(sort_by) {
+                    let built = EngineSnapshot::from_index(index.clone()).build_sort_helper_index(sort_by);
+                    map.insert(
+                        sort_by.to_string(),
+                        SortHelperIndexEntry { index: built, last_accessed: std::time::Instant::now() },
+                    );
+                }
+                let entry = map.get_mut(sort_by).expect("just inserted above if missing");
+                entry.last_accessed = std::time::Instant::now();
+                map.get(sort_by).map(|entry| &entry.index)
+            }
+            SortHelperAccess::Immutable(map) => map.get(sort_by).map(|entry| &entry.index),
+        }
+    }
+}