Auto derived format

Cargo.lock

@@ -774,6 +774,12 @@ version = "0.3.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
 
+[[package]]
+name = "minimal-lexical"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
+
 [[package]]
 name = "mio"
 version = "1.2.3"
@@ -785,6 +791,16 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "nom"
+version = "7.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a"
+dependencies = [
+ "memchr",
+ "minimal-lexical",
+]
+
 [[package]]
 name = "num-traits"
 version = "0.2.19"
@@ -824,6 +840,7 @@ dependencies = [
  "http-body-util",
  "humantime",
  "include_dir",
+ "nom",
  "pest",
  "pest_derive",
  "regex",

Cargo.toml

@@ -10,6 +10,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 clap = { version = "4", features = ["derive"] }
 regex = "1"
+nom = "7"
 pest = "2"
 pest_derive = "2"
 chrono = "0.4"

src/api.rs

@@ -13,7 +13,7 @@ use crate::engine::{
     EngineResponse,
 };
 use crate::alerts::{deliver_ntfy, AlertError};
-use crate::profile::{self, Profile, ProfileError, ProfileStore};
+use crate::profile::{self, ParamDeclarationRecord, Profile, ProfileError, ProfileStore};
 use crate::settings::{CssError, SettingsHandle};
 use axum::extract::{Path as AxumPath, Query, State};
 use axum::http::{header, StatusCode};
@@ -55,13 +55,21 @@ impl ApiState {
         }
     }
 
-    /// Same as `persist_param_declaration`, for metric declarations.
-    fn persist_metric_declaration(&self, request: &DeclareMetricRequest) {
+    /// Same as `persist_param_declaration`, for metric declarations. Also promotes any of the
+    /// metric's referenced params that only exist as auto-derived engine state (see
+    /// `promoted_param_records`) into persisted profile params, so a metric built against an
+    /// auto-detected logfmt key survives a restart instead of being rejected as "undeclared" once
+    /// declarations replay before any log line re-derives it.
+    async fn persist_metric_declaration(&self, request: &DeclareMetricRequest) {
         let Some((store_path, profile_name)) = self.active_profile_location() else {
             return;
         };
         let mut store = ProfileStore::load(&store_path);
-        if let Err(err) = store.record_metric_declaration(&store_path, &profile_name, request) {
+        let existing_params = store.get(&profile_name).map(|p| p.params.clone()).unwrap_or_default();
+        let promoted = promoted_param_records(&self.engine, &existing_params, request).await;
+        if let Err(err) =
+            store.record_metric_declaration(&store_path, &profile_name, request, &promoted)
+        {
             eprintln!(
                 "warning: failed to persist metric declaration '{}' into profile '{profile_name}': {err}",
                 request.name
@@ -70,14 +78,20 @@ impl ApiState {
     }
 
     /// Same as `persist_metric_declaration`, for an in-place edit (`PUT /api/metrics/{name}`) — also drops the persisted record under `old_name` when the edit renames the metric, so a rename doesn't leave a stale duplicate declaration in the profile file.
-    fn persist_metric_declaration_update(&self, old_name: &str, request: &DeclareMetricRequest) {
+    async fn persist_metric_declaration_update(&self, old_name: &str, request: &DeclareMetricRequest) {
         let Some((store_path, profile_name)) = self.active_profile_location() else {
             return;
         };
         let mut store = ProfileStore::load(&store_path);
-        if let Err(err) =
-            store.record_metric_declaration_update(&store_path, &profile_name, old_name, request)
-        {
+        let existing_params = store.get(&profile_name).map(|p| p.params.clone()).unwrap_or_default();
+        let promoted = promoted_param_records(&self.engine, &existing_params, request).await;
+        if let Err(err) = store.record_metric_declaration_update(
+            &store_path,
+            &profile_name,
+            old_name,
+            request,
+            &promoted,
+        ) {
             eprintln!(
                 "warning: failed to persist metric declaration update '{}' into profile '{profile_name}': {err}",
                 request.name
@@ -177,6 +191,47 @@ async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result<EngineResponse
     response.await.map_err(|_| StatusCode::SERVICE_UNAVAILABLE)
 }
 
+/// Any of `request`'s referenced params (`label_params`/`value_param`) not already present in
+/// `existing_params` are looked up live in the engine and returned as records ready to persist
+/// alongside the metric (see `ProfileStore::record_metric_declaration`'s `promoted_params`). A
+/// metric can validly reference a param that's only auto-derived engine state (see
+/// `crate::engine::auto_declare_params`) — that param was never written into any profile, so without
+/// this promotion the metric would be rejected as "undeclared" the next time declarations replay on
+/// startup, before any log line has re-derived it.
+async fn promoted_param_records(
+    engine: &EngineHandle,
+    existing_params: &[ParamDeclarationRecord],
+    request: &DeclareMetricRequest,
+) -> Vec<ParamDeclarationRecord> {
+    let missing: Vec<&str> = request
+        .label_params
+        .iter()
+        .map(String::as_str)
+        .chain(request.value_param.as_deref())
+        .filter(|name| !existing_params.iter().any(|p| p.name == **name))
+        .collect();
+    if missing.is_empty() {
+        return Vec::new();
+    }
+    let Ok(EngineResponse::ParamDeclarations(declarations)) =
+        ask(engine, EngineQuery::ParamDeclarations).await
+    else {
+        return Vec::new();
+    };
+    declarations
+        .into_iter()
+        .filter(|param| missing.contains(&param.name.as_str()))
+        .map(|param| ParamDeclarationRecord {
+            name: param.name,
+            message_match_regex: param.message_match_regex,
+            capture_group: param.capture_group,
+            json_field_path: param.json_field_path,
+            kind: param.kind.into(),
+            elapsed_since_last: param.elapsed_since_last,
+        })
+        .collect()
+}
+
 async fn get_logs(
     State(state): State<ApiState>,
     Query(params): Query<LogsParams>,
@@ -245,6 +300,9 @@ async fn declare_param(
                 json_field_path: request.json_field_path,
                 kind: request.kind,
                 elapsed_since_last: request.elapsed_since_last,
+                // A request that reached here was accepted by `declare_param`, which always stores
+                // it as non-auto (see `engine::declare_param`).
+                auto_derived: false,
             };
             (StatusCode::CREATED, Json(dto)).into_response()
         }
@@ -277,7 +335,7 @@ async fn declare_metric(
     }
     match response.await {
         Ok(Ok(())) => {
-            state.persist_metric_declaration(&request);
+            state.persist_metric_declaration(&request).await;
             let dto = MetricDeclarationDto {
                 name: request.name,
                 message_match_regex: request.message_match_regex,
@@ -311,7 +369,7 @@ async fn update_metric(
     }
     match response.await {
         Ok(Ok(())) => {
-            state.persist_metric_declaration_update(&name, &request);
+            state.persist_metric_declaration_update(&name, &request).await;
             let dto = MetricDeclarationDto {
                 name: request.name,
                 message_match_regex: request.message_match_regex,
@@ -602,11 +660,14 @@ async fn delete_profile(
 }
 
 /// Removes every currently-declared param/metric from the engine, then replays `profile`'s declarations into it.
+/// Auto-derived params (`ParamDeclarationDto::auto_derived`) are left alone — they aren't part of
+/// any profile, just facts discovered from the log stream, so a profile switch shouldn't discard
+/// them (they'd otherwise only reappear once a new line re-introduces that key).
 async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) {
     if let Ok(EngineResponse::ParamDeclarations(params)) =
         ask(engine, EngineQuery::ParamDeclarations).await
     {
-        for param in params {
+        for param in params.into_iter().filter(|param| !param.auto_derived) {
             let (respond_to, response) = oneshot::channel();
             if engine
                 .send(EngineMessage::RemoveParamDeclaration { name: param.name, respond_to })
@@ -1206,6 +1267,75 @@ mod tests {
         assert_eq!(profile.params[0].name, "level");
     }
 
+    /// Regression test: declaring a metric against a param that only exists as auto-derived engine
+    /// state (discovered from an ingested logfmt key, never manually declared via `POST
+    /// /api/params`) must persist that param into the active profile too, not just the metric —
+    /// otherwise a restart replays the metric before any log line re-derives the auto param, and the
+    /// engine rejects it as referencing an undeclared param (see `promoted_param_records`).
+    #[tokio::test]
+    async fn declaring_metric_against_auto_derived_param_promotes_it_into_the_profile() {
+        let (app, engine, settings_handle, dir) = test_app();
+
+        let settings_path = dir.path().join("settings.json");
+        let profiles_dir = {
+            let settings = settings_handle.lock().unwrap();
+            profile::resolve_profiles_dir(&settings, &settings_path)
+        };
+        let store_path = profile::profiles_store_path(&profiles_dir);
+        let mut store = ProfileStore::load(&store_path);
+        {
+            let mut settings = settings_handle.lock().unwrap();
+            store
+                .create(&store_path, &mut settings, &settings_path, "web".to_string())
+                .unwrap();
+            store
+                .set_active(&mut settings, &settings_path, Some("web".to_string()))
+                .unwrap();
+        }
+
+        // Ingest a line carrying a logfmt key before any param is manually declared, so "dx" only
+        // exists as engine-internal auto-derived state (never persisted anywhere).
+        ingest_lines(&engine, 0, &["dx=1 moved"]).await;
+
+        let (status, _body) = send_json(
+            &app,
+            "POST",
+            "/api/metrics",
+            json!({
+                "name": "direction_x",
+                "message_match_regex": r"dx=(?P<dx>\S+) moved",
+                "label_params": ["dx"]
+            }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::CREATED);
+
+        let reloaded_store = ProfileStore::load(&store_path);
+        let profile = reloaded_store.get("web").unwrap();
+        assert_eq!(profile.metrics.len(), 1);
+        assert_eq!(profile.metrics[0].name, "direction_x");
+        // The auto-derived "dx" param got promoted into the profile alongside the metric.
+        assert_eq!(profile.params.len(), 1);
+        assert_eq!(profile.params[0].name, "dx");
+
+        // Simulate a restart: replay the persisted profile into a fresh engine with no prior ingestion.
+        let (fresh_engine_tx, fresh_engine_rx) = mpsc::channel(16);
+        let fresh_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default()));
+        tokio::spawn(crate::engine::run(fresh_engine_rx, fresh_settings_handle));
+        profile::load_into_engine(&fresh_engine_tx, profile).await;
+
+        let (respond_to, response) = oneshot::channel();
+        fresh_engine_tx
+            .send(EngineMessage::Query { query: EngineQuery::MetricDeclarations, respond_to })
+            .await
+            .unwrap();
+        let EngineResponse::MetricDeclarations(metrics) = response.await.unwrap() else {
+            panic!("unexpected response");
+        };
+        assert_eq!(metrics.len(), 1);
+        assert_eq!(metrics[0].name, "direction_x");
+    }
+
     async fn delete_request(app: &Router, uri: &str) -> StatusCode {
         let request = Request::builder().method("DELETE").uri(uri).body(Body::empty()).unwrap();
         app.clone().oneshot(request).await.unwrap().status()

src/data_value.rs

@@ -33,6 +33,14 @@ pub struct ParamDeclaration {
     /// regex/JSON-field extraction.
     #[serde(default)]
     pub elapsed_since_last: bool,
+    /// Set when this param was discovered by `crate::engine::auto_declare_params` (a nom-based
+    /// `key=value` scan run on every ingested line) rather than declared through `POST /api/params`.
+    /// `declare_param` refuses to overwrite an auto-derived param by name, and it's never persisted
+    /// to a profile — see `crate::engine::declare_param` and `crate::api::reload_engine_declarations`.
+    /// `#[serde(default)]` so profiles/requests saved before this field existed still load as
+    /// ordinary (non-auto) params.
+    #[serde(default)]
+    pub auto_derived: bool,
 }
 
 /// Reads a value out of a JSON object by dot-separated field path (e.g. `"foo.bar"`), for

src/dto.rs

@@ -25,6 +25,9 @@ pub struct ParamDeclarationDto {
     pub kind: MetricPrimitiveKind,
     /// See `crate::data_value::ParamDeclaration::elapsed_since_last`.
     pub elapsed_since_last: bool,
+    /// See `crate::data_value::ParamDeclaration::auto_derived`. The frontend uses this to disable
+    /// editing — `POST /api/params` also rejects a request naming an auto-derived param.
+    pub auto_derived: bool,
 }
 
 #[derive(Debug, Clone, Serialize)]

src/engine.rs

@@ -152,6 +152,7 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
                     DataValue::LogLiteral { text: text.clone() },
                 );
                 log_count += 1;
+                auto_declare_params(&mut index, &text);
                 apply_all_metrics(&mut index, sequence, timestamp, &text);
                 evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total);
 
@@ -219,6 +220,16 @@ fn declare_param(
     index: &mut BTreeMap<IndexKey, DataValue>,
     request: DeclareParamRequest,
 ) -> Result<(), String> {
+    if let Some(DataValue::ParamDeclaration(existing)) =
+        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
+            ));
+        }
+    }
     // An `elapsed_since_last` param is never text-extracted at all, so `message_match_regex` is
     // irrelevant to it — but the field always compiles as a regex regardless of extractor (an empty
     // pattern is valid and matches trivially), so it's harmless to still require it to parse.
@@ -243,11 +254,78 @@ fn declare_param(
 
     index.insert(
         IndexKey::ParamDeclaration { name: request.name.clone() },
-        DataValue::ParamDeclaration(request),
+        // `auto_derived` is never trusted from the request — only `auto_declare_params` may set it.
+        DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }),
     );
     Ok(())
 }
 
+/// Turns an arbitrary log-line key into a valid regex named-capture-group identifier (ASCII
+/// alphanumeric/underscore, not starting with a digit) — the param's own `name` (shown to the user,
+/// used as the index key) keeps the raw key text; only the capture group embedded in its generated
+/// regex needs sanitizing. Two different keys sanitizing to the same group name is harmless since
+/// each param's regex/capture-group pair is private to that one param.
+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
+}
+
+/// Conservative kind inference for an auto-derived param's first sample: only `Number` is guessed
+/// (int, falling back to float), everything else stays `String` — `Boolean`/`Timestamp`/`Duration`
+/// carry extraction semantics beyond "parses as", so guessing them wrong would silently drop values
+/// that don't fit (see `parse_metric_primitive`).
+fn infer_kind(sample: &str) -> MetricPrimitiveKind {
+    if sample.parse::<i64>().is_ok() || sample.parse::<f64>().is_ok() {
+        MetricPrimitiveKind::Number
+    } else {
+        MetricPrimitiveKind::String
+    }
+}
+
+/// Builds the regex an auto-derived param uses to re-extract its value from any line on read
+/// (`derive_log_params`) — quoted keys always look for a `"..."` value, unquoted keys a
+/// whitespace-delimited bareword, matching how `crate::logfmt::extract_pairs` found the sample.
+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+)")
+    }
+}
+
+/// Scans `text` for `key=value`/`key="value"` pairs (`crate::logfmt::extract_pairs`) and declares a
+/// standalone param for any key not already declared — auto or manual. A manual declaration for a
+/// name always wins and is never touched here; this only fills in names nobody has declared yet.
+/// Run on every ingested line (`EngineMessage::Ingest`), same as `apply_all_metrics`, but note param
+/// *values* are still recomputed from raw text at read time (`derive_log_params`), not stored here —
+/// so a key discovered on line N also back-fills the column for lines before N on the next read.
+fn auto_declare_params(index: &mut BTreeMap<IndexKey, DataValue>, text: &str) {
+    for (key, value, quoted) in crate::logfmt::extract_pairs(text) {
+        let index_key = IndexKey::ParamDeclaration { name: key.to_string() };
+        if index.contains_key(&index_key) {
+            continue;
+        }
+        let group = sanitize_group_name(key);
+        index.insert(
+            index_key,
+            DataValue::ParamDeclaration(ParamDeclaration {
+                name: key.to_string(),
+                message_match_regex: build_auto_param_pattern(key, &group, quoted),
+                capture_group: group,
+                json_field_path: None,
+                kind: infer_kind(value),
+                elapsed_since_last: false,
+                auto_derived: true,
+            }),
+        );
+    }
+}
+
 /// Validates the metric's own regex plus that every name in `label_params`/`value_param` refers to
 /// an already-declared standalone param (not just the first bad one, so a rejection reports every
 /// problem at once). Shared by `declare_metric` and `update_metric` so both reject a bad
@@ -407,7 +485,7 @@ struct ResolvedParam {
 
 fn resolved_param_declarations(index: &BTreeMap<IndexKey, DataValue>) -> HashMap<String, ResolvedParam> {
     param_declarations_range(index)
-        .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last)| {
+        .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, _)| {
             Regex::new(message_match_regex).ok().map(|regex| {
                 (
                     name.to_string(),
@@ -956,19 +1034,20 @@ fn metric_declarations_range(
 
 fn param_declarations_range(
     index: &BTreeMap<IndexKey, DataValue>,
-) -> impl Iterator<Item = (&str, &str, &str, Option<&str>, MetricPrimitiveKind, bool)> {
+) -> impl Iterator<Item = (&str, &str, &str, Option<&str>, MetricPrimitiveKind, bool, bool)> {
     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,
                 }),
             ) => Some((
                 name.as_str(),
@@ -977,6 +1056,7 @@ fn param_declarations_range(
                 json_field_path.as_deref(),
                 *kind,
                 *elapsed_since_last,
+                *auto_derived,
             )),
             _ => None,
         })
@@ -991,8 +1071,8 @@ fn compiled_param_declarations(
     index: &BTreeMap<IndexKey, DataValue>,
 ) -> Vec<(String, Regex, String, Option<String>, MetricPrimitiveKind)> {
     param_declarations_range(index)
-        .filter(|(.., elapsed_since_last)| !elapsed_since_last)
-        .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, _)| {
+        .filter(|(_, _, _, _, _, elapsed_since_last, _)| !elapsed_since_last)
+        .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, _, _)| {
             Regex::new(message_match_regex).ok().map(|re| {
                 (
                     name.to_string(),
@@ -1198,7 +1278,7 @@ fn handle_query(
         }
         EngineQuery::ParamDeclarations => {
             let declarations = param_declarations_range(index)
-                .map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last)| {
+                .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(),
@@ -1206,6 +1286,7 @@ fn handle_query(
                         json_field_path: json_field_path.map(|path| path.to_string()),
                         kind,
                         elapsed_since_last,
+                        auto_derived,
                     }
                 })
                 .collect();
@@ -1578,6 +1659,7 @@ mod tests {
             json_field_path: None,
             kind: MetricPrimitiveKind::String,
             elapsed_since_last: false,
+            auto_derived: false,
         }
     }
 
@@ -1594,6 +1676,51 @@ mod tests {
         assert_eq!(declarations[0].2, "level");
     }
 
+    #[test]
+    fn auto_declare_params_discovers_new_keys_with_inferred_kinds_and_flags_them_auto() {
+        let mut index = BTreeMap::new();
+        auto_declare_params(&mut index, r#"ts=1788648105773 msg="entity_moved" x1=-4 y1=34"#);
+
+        let declarations: HashMap<&str, _> =
+            param_declarations_range(&index).map(|d| (d.0, d)).collect();
+        assert_eq!(declarations.len(), 4);
+        assert_eq!(declarations["x1"].4, MetricPrimitiveKind::Number);
+        assert!(declarations["x1"].6, "expected x1 to be flagged auto_derived");
+        assert_eq!(declarations["msg"].4, MetricPrimitiveKind::String);
+
+        let compiled = compiled_param_declarations(&index);
+        let params = derive_log_params(r#"ts=1788648105773 msg="entity_moved" x1=-4 y1=34"#, &compiled);
+        assert_eq!(params.get("x1"), Some(&MetricPrimitive::Number(Number::Int(-4))));
+        assert_eq!(params.get("msg"), Some(&MetricPrimitive::String("entity_moved".to_string())));
+    }
+
+    #[test]
+    fn auto_declare_params_never_overwrites_an_existing_declaration() {
+        let mut index = BTreeMap::new();
+        declare_param(&mut index, request("x1", r#"x1="(?P<x1>[^"]+)""#, "x1")).unwrap();
+
+        auto_declare_params(&mut index, "x1=-4");
+
+        let declarations: Vec<_> = param_declarations_range(&index).collect();
+        assert_eq!(declarations.len(), 1);
+        assert_eq!(declarations[0].1, r#"x1="(?P<x1>[^"]+)""#, "manual declaration must survive untouched");
+        assert!(!declarations[0].6, "a manual declaration must never be flagged auto_derived");
+    }
+
+    #[test]
+    fn declare_param_rejects_editing_an_auto_derived_param() {
+        let mut index = BTreeMap::new();
+        auto_declare_params(&mut index, "x1=-4");
+
+        let result = declare_param(&mut index, request("x1", r#"x1="(?P<x1>[^"]+)""#, "x1"));
+
+        let err = result.unwrap_err();
+        assert!(err.contains("auto-detected"), "unexpected error message: {err}");
+        let declarations: Vec<_> = param_declarations_range(&index).collect();
+        assert_eq!(declarations.len(), 1);
+        assert!(declarations[0].6, "the original auto-derived declaration must be untouched");
+    }
+
     #[test]
     fn rejects_invalid_regex() {
         let mut index = BTreeMap::new();
@@ -1635,6 +1762,7 @@ mod tests {
                 json_field_path: None,
                 kind: MetricPrimitiveKind::String,
                 elapsed_since_last: true,
+                auto_derived: false,
             },
         );
         assert!(result.is_err());
@@ -1655,6 +1783,7 @@ mod tests {
                 json_field_path: Some("foo.bar".to_string()),
                 kind: MetricPrimitiveKind::String,
                 elapsed_since_last: false,
+                auto_derived: false,
             },
         );
         assert!(result.is_ok());
@@ -1676,6 +1805,7 @@ mod tests {
                 json_field_path: Some("foo.bar".to_string()),
                 kind: MetricPrimitiveKind::Number,
                 elapsed_since_last: false,
+                auto_derived: false,
             },
         )
         .unwrap();
@@ -1792,6 +1922,7 @@ mod tests {
                 json_field_path: None,
                 kind,
                 elapsed_since_last: false,
+                auto_derived: false,
             },
         )
         .unwrap();
@@ -1811,6 +1942,7 @@ mod tests {
                 json_field_path: None,
                 kind: MetricPrimitiveKind::Duration,
                 elapsed_since_last: true,
+                auto_derived: false,
             },
         )
         .unwrap();
@@ -2032,6 +2164,7 @@ mod tests {
                 json_field_path: Some("request.path".to_string()),
                 kind: MetricPrimitiveKind::String,
                 elapsed_since_last: false,
+                auto_derived: false,
             },
         )
         .unwrap();

src/logfmt.rs

@@ -0,0 +1,111 @@
+//! Fast, tolerant `key=value`/`key="value"` tokenizer (logfmt-ish), used by
+//! `crate::engine::auto_declare_params` to discover param names as lines are ingested. Scans
+//! anywhere in the line rather than requiring the whole line to be logfmt, so it works on lines
+//! with an arbitrary non-logfmt prefix/suffix (e.g. `ts=... msg="..." x1=-4 y1=34`).
+
+use nom::branch::alt;
+use nom::bytes::complete::{escaped, take_while1};
+use nom::character::complete::{char, none_of, one_of};
+use nom::sequence::delimited;
+use nom::IResult;
+
+fn key(input: &str) -> IResult<&str, &str> {
+    take_while1(|c: char| c.is_ascii_alphanumeric() || c == '_' || c == '.' || c == '-')(input)
+}
+
+fn quoted_value(input: &str) -> IResult<&str, &str> {
+    delimited(
+        char('"'),
+        alt((escaped(none_of("\\\""), '\\', one_of("\"\\nt")), nom::combinator::success(""))),
+        char('"'),
+    )(input)
+}
+
+fn bareword_value(input: &str) -> IResult<&str, &str> {
+    take_while1(|c: char| !c.is_whitespace())(input)
+}
+
+/// One `key=value` pair found on a line: the key, the raw captured value (unescaped as-is, quotes
+/// stripped but any inner `\"`/`\\` left literal since callers only need it for numeric-kind
+/// inference), and whether the value was quoted.
+fn pair(input: &str) -> IResult<&str, (&str, &str, bool)> {
+    let (input, k) = key(input)?;
+    let (input, _) = char('=')(input)?;
+    if let Ok((rest, v)) = quoted_value(input) {
+        Ok((rest, (k, v, true)))
+    } else {
+        let (rest, v) = bareword_value(input)?;
+        Ok((rest, (k, v, false)))
+    }
+}
+
+/// Finds every `key=value`/`key="value"` pair anywhere in `line`. A position that doesn't start a
+/// pair is skipped one char at a time, so this never fails outright — a line with no pairs at all
+/// (plain prose, a JSON blob) just returns an empty vec.
+pub fn extract_pairs(line: &str) -> Vec<(&str, &str, bool)> {
+    let mut results = Vec::new();
+    let mut rest = line;
+    while !rest.is_empty() {
+        match pair(rest) {
+            Ok((remaining, found)) => {
+                results.push(found);
+                rest = remaining;
+            }
+            Err(_) => {
+                let mut chars = rest.chars();
+                chars.next();
+                rest = chars.as_str();
+            }
+        }
+    }
+    results
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn extracts_quoted_value() {
+        let pairs = extract_pairs(r#"msg="entity_moved""#);
+        assert_eq!(pairs, vec![("msg", "entity_moved", true)]);
+    }
+
+    #[test]
+    fn extracts_unquoted_bareword_value() {
+        let pairs = extract_pairs("x1=-4");
+        assert_eq!(pairs, vec![("x1", "-4", false)]);
+    }
+
+    #[test]
+    fn extracts_mixed_quoted_and_unquoted_pairs_in_one_line() {
+        let pairs = extract_pairs(r#"ts=1788648105773 msg="entity_moved" x1=-4 y1=34"#);
+        assert_eq!(
+            pairs,
+            vec![
+                ("ts", "1788648105773", false),
+                ("msg", "entity_moved", true),
+                ("x1", "-4", false),
+                ("y1", "34", false),
+            ]
+        );
+    }
+
+    #[test]
+    fn returns_empty_for_lines_with_no_pairs() {
+        assert!(extract_pairs("just some plain prose here").is_empty());
+        assert!(extract_pairs(r#"{"foo": {"bar": 42}}"#).is_empty());
+    }
+
+    #[test]
+    fn does_not_panic_on_empty_value_at_end_of_line() {
+        assert_eq!(extract_pairs("key="), Vec::<(&str, &str, bool)>::new());
+        assert_eq!(extract_pairs(r#"key="""#), vec![("key", "", true)]);
+    }
+
+    #[test]
+    fn keeps_escaped_quote_inside_quoted_value() {
+        let pairs = extract_pairs(r#"msg="he said \"hi\"""#);
+        assert_eq!(pairs, vec![("msg", r#"he said \"hi\""#, true)]);
+    }
+}

src/main.rs

@@ -7,6 +7,7 @@ mod filter_datetime;
 mod filter_parser;
 mod index_key;
 mod ingest;
+mod logfmt;
 mod metric_primitive;
 mod profile;
 mod promql;

src/profile.rs

@@ -98,6 +98,8 @@ impl From<&ParamDeclarationRecord> for DeclareParamRequest {
             json_field_path: record.json_field_path.clone(),
             kind: record.kind.into(),
             elapsed_since_last: record.elapsed_since_last,
+            // A profile only ever stores manually-declared params (see `record_param_declaration`).
+            auto_derived: false,
         }
     }
 }
@@ -682,17 +684,32 @@ impl ProfileStore {
     }
 
     /// Records/updates a declared metric into `profile_name`'s stored declarations (matched by metric name), persisting the store; same integration intent as `record_param_declaration`.
+    ///
+    /// `promoted_params` is any of the metric's referenced `label_params`/`value_param` that live
+    /// only as auto-derived engine state (see `crate::engine::auto_declare_params`) rather than a
+    /// persisted profile param — the caller (`ApiState::persist_metric_declaration`) resolves these
+    /// against the live engine before calling in. Without persisting them alongside the metric, a
+    /// restart would replay the metric before any log line re-derives its auto params, and the
+    /// engine would reject it as referencing an undeclared param. Each is added only if `profile
+    /// .params` doesn't already have an entry of that name, so an existing (e.g. manually-declared)
+    /// param is never clobbered.
     pub fn record_metric_declaration(
         &mut self,
         store_path: &Path,
         profile_name: &str,
         request: &DeclareMetricRequest,
+        promoted_params: &[ParamDeclarationRecord],
     ) -> Result<(), ProfileError> {
         let profile = self
             .profiles
             .iter_mut()
             .find(|p| p.name == profile_name)
             .ok_or(ProfileError::NotFound)?;
+        for param in promoted_params {
+            if !profile.params.iter().any(|p| p.name == param.name) {
+                profile.params.push(param.clone());
+            }
+        }
         let record: MetricDeclarationRecord = request.into();
         match profile.metrics.iter_mut().find(|r| r.name == record.name) {
             Some(existing) => *existing = record,
@@ -702,13 +719,14 @@ impl ProfileStore {
         Ok(())
     }
 
-    /// Same as `record_metric_declaration`, for an in-place edit that may rename the metric: drops any persisted record under `old_name` first (when it differs from the new name) so a rename doesn't leave a stale duplicate declaration behind, then upserts by the new name as usual.
+    /// Same as `record_metric_declaration`, for an in-place edit that may rename the metric: drops any persisted record under `old_name` first (when it differs from the new name) so a rename doesn't leave a stale duplicate declaration behind, then upserts by the new name as usual. `promoted_params` has the same meaning as in `record_metric_declaration`.
     pub fn record_metric_declaration_update(
         &mut self,
         store_path: &Path,
         profile_name: &str,
         old_name: &str,
         request: &DeclareMetricRequest,
+        promoted_params: &[ParamDeclarationRecord],
     ) -> Result<(), ProfileError> {
         let profile = self
             .profiles
@@ -718,6 +736,11 @@ impl ProfileStore {
         if old_name != request.name {
             profile.metrics.retain(|r| r.name != old_name);
         }
+        for param in promoted_params {
+            if !profile.params.iter().any(|p| p.name == param.name) {
+                profile.params.push(param.clone());
+            }
+        }
         let record: MetricDeclarationRecord = request.into();
         match profile.metrics.iter_mut().find(|r| r.name == record.name) {
             Some(existing) => *existing = record,
@@ -815,6 +838,7 @@ mod tests {
             json_field_path: None,
             kind: MetricPrimitiveKind::String,
             elapsed_since_last: false,
+            auto_derived: false,
         }
     }
 
@@ -1080,7 +1104,7 @@ mod tests {
         store.record_param_declaration(&store_path, "web", &param_req).unwrap();
 
         let metric_req = metric_request("requests", r"duration=(?P<duration>\d+)", "duration");
-        store.record_metric_declaration(&store_path, "web", &metric_req).unwrap();
+        store.record_metric_declaration(&store_path, "web", &metric_req, &[]).unwrap();
 
         let profile = store.get("web").unwrap();
         assert_eq!(profile.params.len(), 1);
@@ -1134,6 +1158,7 @@ mod tests {
                 &store_path,
                 "web",
                 &metric_request("requests", r"request duration=(?P<duration>\d+)", "duration"),
+                &[],
             )
             .unwrap();
         store.set_active(&mut settings, &settings_path, Some("web".to_string())).unwrap();
@@ -1465,7 +1490,7 @@ mod tests {
             )
             .unwrap();
         store
-            .record_metric_declaration(&store_path, "web", &metric_request("requests", r"path=(?P<path>\S+)", "path"))
+            .record_metric_declaration(&store_path, "web", &metric_request("requests", r"path=(?P<path>\S+)", "path"), &[])
             .unwrap();
 
         let before = store.get("web").unwrap().clone();
@@ -1473,4 +1498,83 @@ mod tests {
         let after = reloaded.get("web").unwrap().clone();
         assert_eq!(before, after);
     }
+
+    /// Acceptance criterion: a metric declared against a param that only exists as auto-derived
+    /// engine state (not yet a persisted profile param) still ingests correctly after a simulated
+    /// restart, because `record_metric_declaration`'s `promoted_params` persists that param into the
+    /// profile alongside the metric. Without this, `load_into_engine` would reject the metric on
+    /// restart since declarations replay before any log line re-derives the auto param.
+    #[tokio::test]
+    async fn metric_referencing_auto_derived_param_survives_restart_via_promotion() {
+        let dir = tempfile::tempdir().unwrap();
+        let store_path = dir.path().join("profiles.json");
+        let settings_path = dir.path().join("settings.json");
+        let mut settings = Settings::default();
+
+        let mut store = ProfileStore::load(&store_path);
+        store.create(&store_path, &mut settings, &settings_path, "web".to_string()).unwrap();
+        store.set_active(&mut settings, &settings_path, Some("web".to_string())).unwrap();
+
+        // Simulate the auto-derive path: ingest a line carrying a logfmt key before any param is
+        // manually declared, so `dx` only exists as engine-internal auto-derived state.
+        let (engine_tx, engine_rx) = mpsc::channel(16);
+        let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone()));
+        tokio::spawn(crate::engine::run(engine_rx, settings_handle));
+        engine_tx
+            .send(EngineMessage::Ingest(IngestedLine {
+                sequence: 0,
+                timestamp: Timestamp::from_millis(0),
+                text: "dx=1 moved".to_string(),
+            }))
+            .await
+            .unwrap();
+
+        // Declare a metric referencing the auto-derived "dx" param — this succeeds live (the param
+        // exists in the engine), mirroring the interactive-declare flow that hit this bug.
+        let metric_req = metric_request("direction_x", r"dx=(?P<dx>\S+) moved", "dx");
+        let (respond_to, response) = oneshot::channel();
+        engine_tx
+            .send(EngineMessage::DeclareMetric { request: metric_req.clone(), respond_to })
+            .await
+            .unwrap();
+        response.await.unwrap().unwrap();
+
+        // Promote "dx" from its live (auto-derived) engine state, as `ApiState::persist_metric_declaration` does.
+        let EngineResponse::ParamDeclarations(live_params) =
+            ask(&engine_tx, EngineQuery::ParamDeclarations).await
+        else {
+            panic!("unexpected response");
+        };
+        let dx = live_params.iter().find(|p| p.name == "dx").unwrap();
+        assert!(dx.auto_derived);
+        let promoted = ParamDeclarationRecord {
+            name: dx.name.clone(),
+            message_match_regex: dx.message_match_regex.clone(),
+            capture_group: dx.capture_group.clone(),
+            json_field_path: dx.json_field_path.clone(),
+            kind: dx.kind.into(),
+            elapsed_since_last: dx.elapsed_since_last,
+        };
+        store.record_metric_declaration(&store_path, "web", &metric_req, &[promoted]).unwrap();
+
+        // Simulate a restart: drop everything, reload from disk into a *fresh* engine with no prior ingestion.
+        drop(store);
+        let reloaded_store = ProfileStore::load(&store_path);
+        let profile = reloaded_store.get("web").unwrap().clone();
+        assert_eq!(profile.params.len(), 1);
+        assert_eq!(profile.params[0].name, "dx");
+
+        let (fresh_engine_tx, fresh_engine_rx) = mpsc::channel(16);
+        let fresh_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default()));
+        tokio::spawn(crate::engine::run(fresh_engine_rx, fresh_settings_handle));
+        load_into_engine(&fresh_engine_tx, &profile).await;
+
+        let EngineResponse::MetricDeclarations(metrics) =
+            ask(&fresh_engine_tx, EngineQuery::MetricDeclarations).await
+        else {
+            panic!("unexpected response");
+        };
+        assert_eq!(metrics.len(), 1);
+        assert_eq!(metrics[0].name, "direction_x");
+    }
 }

src/promql.rs

@@ -289,6 +289,10 @@ impl<'a> Parser<'a> {
             _ => None,
         };
         self.skip_whitespace();
+        // `sum by (label) (metric)`: a `by (...)` clause may also precede the aggregation's
+        // parenthesized argument, not just follow it.
+        let prefix_by_labels = if agg.is_some() { self.parse_by_clause()? } else { Vec::new() };
+        self.skip_whitespace();
         if agg.is_some() && self.peek() == Some('(') {
             self.expect_char('(')?;
             self.skip_whitespace();
@@ -323,7 +327,15 @@ impl<'a> Parser<'a> {
             };
             self.expect_char(')')?;
             inner.agg = agg;
-            inner.by_labels = self.parse_by_clause()?;
+            let trailing_by_labels = self.parse_by_clause()?;
+            if !prefix_by_labels.is_empty() && !trailing_by_labels.is_empty() {
+                return Err(
+                    "by (...) clause may only appear once, either before or after the aggregation's argument"
+                        .to_string(),
+                );
+            }
+            inner.by_labels =
+                if prefix_by_labels.is_empty() { trailing_by_labels } else { prefix_by_labels };
             Ok(inner)
         } else {
             // Rewind and parse as a plain vector selector, covering metric names that collide with an agg_op keyword.
@@ -547,4 +559,23 @@ mod tests {
     fn rejects_unterminated_by_clause() {
         assert!(parse_query("sum(requests) by (path").is_err());
     }
+
+    #[test]
+    fn parses_sum_with_prefix_by_clause() {
+        let query = parse_query("sum by (path) (requests)").unwrap();
+        assert_eq!(query.agg, Some(AggOp::Sum));
+        assert_eq!(query.by_labels, vec!["path".to_string()]);
+    }
+
+    #[test]
+    fn parses_sum_with_prefix_by_clause_multiple_labels() {
+        let query = parse_query("sum by (path, method) (requests)").unwrap();
+        assert_eq!(query.by_labels, vec!["path".to_string(), "method".to_string()]);
+    }
+
+    #[test]
+    fn rejects_by_clause_both_before_and_after_aggregation() {
+        let err = parse_query("sum by (path) (requests) by (method)").unwrap_err();
+        assert!(err.contains("only appear once"));
+    }
 }