Document level_classify Other allocation (DSK-138)

crates/rustern-core/src/pipeline/level_classify.rs

@@ -2,6 +2,10 @@ use futures::stream::{Stream, StreamExt};
 
 use crate::source::{LogEvent, LogLevel, LogSourceError};
 
+/// Map a JSON level field to [`LogLevel`].
+///
+/// Non-standard tokens become [`LogLevel::Other`] with an owned copy of the raw
+/// string so `--exit-watch-level` can match arbitrary values later in the pipeline.
 fn classify_str(s: &str) -> LogLevel {
     if s.eq_ignore_ascii_case("error")
         || s.eq_ignore_ascii_case("err")
@@ -108,4 +112,33 @@ mod tests {
         let lv = out[0].as_ref().unwrap().level.as_ref().unwrap();
         assert!(matches!(lv, LogLevel::Warn));
     }
+
+    #[tokio::test]
+    async fn preserves_unknown_level_token() {
+        let raw = r#"{"level":"notice","msg":"hey"}"#;
+        let ev = LogEvent {
+            source: Arc::new(SourceMeta {
+                context: ContextName("ctx".into()),
+                namespace: "ns".into(),
+                pod: "p".into(),
+                container: "c".into(),
+                kind: SourceKind::PodLog,
+                node: None,
+                labels: Arc::new(Labels::default()),
+                uid: "u".into(),
+                palette_index: None,
+                container_palette_index: None,
+            }),
+            timestamp: Utc::now(),
+            message: Arc::from(raw),
+            structured: Some(ParsedJson::Serde(serde_json::from_str(raw).unwrap())),
+            level: None,
+            palette_index: None,
+            container_palette_index: None,
+        };
+        let s = futures::stream::iter(vec![Ok(ev)]);
+        let out: Vec<_> = level_classify(s, Some("level".into())).collect().await;
+        let lv = out[0].as_ref().unwrap().level.as_ref().unwrap();
+        assert!(matches!(lv, LogLevel::Other(s) if s == "notice"));
+    }
 }