Resolve tickets 059-061: generic log-stream columns + sort helper indexes

- 059: log stream header/column detection is now generic
  whitespace-columnar splitting (2+ spaces, single tab) instead of one
  fixed six-column regex. Declared params come from whatever columns
  the header actually names, plus a trailing unlabelled message field
  that's never swallowed by the last named column.
- 060: timeline sort-by-property now walks an incrementally-maintained
  per-sort-key helper index (BTreeMap keyed so "has the param, ordered
  by value" sorts before "missing it, ordered by ottyl_seq") instead of
  materializing and re-sorting the full filtered range on every query.
- 061: helper indexes are swept for staleness on ingest and dropped
  after 5 minutes unused; a purged (or never-built) index transparently
  rebuilds via the same lazy one-time scan on the next sorted query.
- Also invalidate a param's helper index on manual re-declaration,
  removal, or ClearParams, so a redefined regex can't leave a sort
  helper index serving values derived under the old declaration.

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

src/engine.rs

@@ -164,6 +164,13 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
     // every currently-ingested log, kept in lockstep with `log_count` — incremented
     // here on ingest, decremented in `evict_excess_logs` when a log is dropped.
     let mut approx_bytes_total: u64 = 0;
+    // Ticket 060/061: one entry per param name that's ever been used as a `sort_by`, populated
+    // lazily on first sorted query and kept incrementally up to date on every ingest thereafter;
+    // swept for staleness on ingest too (see `purge_stale_sort_helper_indexes`).
+    let mut sort_helper_indexes: HashMap<String, SortHelperIndexEntry> = HashMap::new();
+    // Ticket 059: the most recently seen `log stream` header line's column names, cached across
+    // ingested lines since a `log stream` pipe only prints its header once, before any data line.
+    let mut log_stream_header: Option<crate::logfmt::LogStreamHeader> = None;
 
     while let Some(message) = rx.recv().await {
         match message {
@@ -175,9 +182,11 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
                     DataValue::LogLiteral { text: text.clone() },
                 );
                 log_count += 1;
-                auto_declare_log_stream_params(&mut index, &text);
+                auto_declare_log_stream_params(&mut index, &mut log_stream_header, &text);
                 auto_declare_params(&mut index, &text);
                 apply_all_metrics(&mut index, sequence, timestamp, &text);
+                update_sort_helper_indexes_on_ingest(&mut sort_helper_indexes, &index, sequence, timestamp, &text);
+                purge_stale_sort_helper_indexes(&mut sort_helper_indexes);
                 evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total);
 
                 // Evaluate alert rules against just this line, then hand any fired rule off to delivery via tokio::spawn so it stays off the hot path.
@@ -212,16 +221,26 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
                 evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total);
             }
             EngineMessage::DeclareParam { request, respond_to } => {
+                // A manual re-declaration can change the regex/kind under the same name — drop any
+                // helper index built under the old declaration so a sorted query rebuilds it fresh
+                // rather than serving values derived from the now-stale regex.
+                sort_helper_indexes.remove(&request.name);
                 let result = declare_param(&mut index, request);
                 let _ = respond_to.send(result);
             }
             EngineMessage::RemoveParamDeclaration { name, respond_to } => {
-                let existed = index.remove(&IndexKey::ParamDeclaration { name }).is_some();
+                let existed = index.remove(&IndexKey::ParamDeclaration { name: name.clone() }).is_some();
+                // A helper index built under the old declaration (or a same-named one re-declared
+                // with a different regex/kind later) would otherwise keep serving stale derived
+                // values until ticket 061's sweep happens to purge it — dropping it here instead
+                // means the next sorted query for this param rebuilds it fresh.
+                sort_helper_indexes.remove(&name);
                 let _ = respond_to.send(existed);
             }
             EngineMessage::ClearParams { respond_to } => {
                 let removed = clear_all_param_declarations(&mut index);
                 let (reextracted, logs_considered) = reextract_params_from_retained_logs(&mut index);
+                sort_helper_indexes.clear();
                 let _ = respond_to.send(ClearParamsResult { removed, reextracted, logs_considered });
             }
             EngineMessage::DeclareMetric { request, respond_to } => {
@@ -237,7 +256,7 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
                 let _ = respond_to.send(existed);
             }
             EngineMessage::Query { query, respond_to } => {
-                let response = handle_query(&index, query, approx_bytes_total);
+                let response = handle_query(&index, query, approx_bytes_total, &mut sort_helper_indexes);
                 let _ = respond_to.send(response);
             }
         }
@@ -438,43 +457,68 @@ fn auto_declare_params(index: &mut BTreeMap<IndexKey, DataValue>, text: &str) {
     }
 }
 
-/// Declares `log_type`/`process`/`pid` as standalone params the first time a `log stream`-shaped line
+/// Declares one param per column of the most-recently-seen `log stream` header line, plus a trailing
-/// is seen (ticket 050/054) — `Type`, the process name, and `PID` are low-cardinality, genuinely
+/// unlabelled `message` param for whatever text follows the last named column (ticket 059) — replacing
-/// useful columns derivable directly from `log stream`'s fixed-column shape
+/// the old fixed three (`log_type`/`process`/`pid`) with whatever columns the header actually names.
-/// (`crate::logfmt::is_log_stream_line`), with none of the `=`-scanning ambiguity `auto_declare_params`
+/// `header` is threaded through by the caller (`run`'s ingest loop) rather than owned here: a
-/// has to guard against. Gated on that shape so ordinary logfmt/JSON/prose input never gets these three
+/// `log stream` pipe prints its header once, before any data line, and this function only ever sees
-/// params. Unlike `auto_declare_params`, adding these is never *skipped* on account of
+/// one ingested line at a time, so the header has to be cached across calls somewhere. A later header
-/// `MAX_AUTO_DECLARED_PARAMS` — it can only ever add exactly these three names, once each — though
+/// line (e.g. a fresh `log stream` invocation piped into the same run) simply replaces the cached one.
-/// since they're stored as `auto_derived` too, once present they do count against that shared ceiling,
+/// Unlike `auto_declare_params`, adding these is never *skipped* on account of
-/// tightening (slightly) the budget left for generic `key=value` pairs. Like `auto_declare_params`,
+/// `MAX_AUTO_DECLARED_PARAMS` — though since they're stored as `auto_derived` too, once present they do
-/// never overwrites an existing declaration (manual or auto) for any of the three names.
+/// count against that shared ceiling, tightening (slightly) the budget left for generic `key=value`
-fn auto_declare_log_stream_params(index: &mut BTreeMap<IndexKey, DataValue>, text: &str) {
+/// pairs. Like `auto_declare_params`, never overwrites an existing declaration (manual or auto) for
+/// any of these names.
+fn auto_declare_log_stream_params(
+    index: &mut BTreeMap<IndexKey, DataValue>,
+    header: &mut Option<crate::logfmt::LogStreamHeader>,
+    text: &str,
+) {
+    if let Some(detected) = crate::logfmt::detect_log_stream_header(text) {
+        *header = Some(detected);
+        return;
+    }
+    let Some(header) = header.as_ref() else {
+        return;
+    };
     if !crate::logfmt::is_log_stream_line(text) {
         return;
     }
-    let candidates: [(&str, &str, &str, MetricPrimitiveKind); 3] = [
+    let row = crate::logfmt::split_log_stream_row(text, header);
-        ("log_type", crate::logfmt::LOG_STREAM_TYPE_PATTERN, "log_type", MetricPrimitiveKind::String),
+
-        ("process", crate::logfmt::LOG_STREAM_PROCESS_PATTERN, "process", MetricPrimitiveKind::String),
+    for (i, column) in header.columns.iter().enumerate() {
-        ("pid", crate::logfmt::LOG_STREAM_PID_PATTERN, "pid", MetricPrimitiveKind::Number),
+        let Some(value) = row.get(i) else { continue };
-    ];
+        let index_key = IndexKey::ParamDeclaration { name: column.name.clone() };
-    for (name, pattern, group, kind) in candidates {
-        let index_key = IndexKey::ParamDeclaration { name: name.to_string() };
         if index.contains_key(&index_key) {
             continue;
         }
         index.insert(
             index_key,
             DataValue::ParamDeclaration(ParamDeclaration {
-                name: name.to_string(),
+                name: column.name.clone(),
-                message_match_regex: pattern.to_string(),
+                message_match_regex: crate::logfmt::log_stream_column_pattern(header, i, &column.name),
-                capture_group: group.to_string(),
+                capture_group: column.name.clone(),
                 json_field_path: None,
-                kind,
+                kind: infer_kind(value.trim()),
                 elapsed_since_last: false,
                 auto_derived: true,
             }),
         );
     }
+
+    if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) {
+        index.entry(IndexKey::ParamDeclaration { name: "message".to_string() }).or_insert_with(|| {
+            DataValue::ParamDeclaration(ParamDeclaration {
+                name: "message".to_string(),
+                message_match_regex: crate::logfmt::log_stream_message_pattern(header, "message"),
+                capture_group: "message".to_string(),
+                json_field_path: None,
+                kind: infer_kind(message.trim()),
+                elapsed_since_last: false,
+                auto_derived: true,
+            })
+        });
+    }
 }
 
 /// Ticket 052: removes every `IndexKey::ParamDeclaration` entry — manual and auto-derived alike.
@@ -1285,6 +1329,103 @@ fn param_declarations_range(
 /// meaningful reading as a standalone logs-table column (its value depends on which metric's label
 /// set it's grouped under, which only exists once it's referenced by a metric's `label_params`/
 /// `value_param`).
+/// Ticket 060: key type for a sort helper index — one such index exists per param name that's ever
+/// been used as a `sort_by`. Declaration order matters for the derived `Ord`: every `WithParam` key
+/// sorts before every `WithoutParam` key, and within each variant ordering falls out of the tuple
+/// fields (value then sequence; sequence alone), which is exactly `sort_missing_fallback`'s "sorted
+/// lines first, then lines missing the param ordered by `ottyl_seq`" semantics — expressed as part of
+/// the index's own ordering instead of a separate partition-and-append step per query.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
+enum SortHelperKey {
+    WithParam(MetricPrimitive, u64),
+    WithoutParam(u64),
+}
+
+/// One param's helper index plus when a query last actually read from it — ticket 061 sweeps entries
+/// whose `last_accessed` is stale. The map's value is that line's `Timestamp`, not `()`, purely so a
+/// query can reconstruct the exact `IndexKey::IngestedLog { sequence, timestamp }` primary-index key
+/// and look its text up in O(log n) instead of re-scanning `logs_range` to find it by sequence alone.
+struct SortHelperIndexEntry {
+    index: BTreeMap<SortHelperKey, Timestamp>,
+    last_accessed: std::time::Instant,
+}
+
+/// How long a sort helper index may sit unread before ticket 061's sweep drops it. Not exposed via
+/// `Settings` (ticket 061 explicitly scopes that out) — a dropped index is always transparently
+/// rebuildable on the next query for that sort key, so this is purely a memory-bound knob, not a
+/// user-visible behavior change.
+const SORT_HELPER_INDEX_PURGE_AFTER: std::time::Duration = std::time::Duration::from_secs(300);
+
+/// Derives a single named param's value from `text`, for helper-index maintenance where deriving
+/// every declared param (`derive_log_params`) per ingested line would be wasted work when only one
+/// sort key's helper index actually needs updating.
+fn derive_single_param(
+    text: &str,
+    declarations: &[(String, Regex, String, Option<String>, MetricPrimitiveKind)],
+    param_name: &str,
+) -> Option<MetricPrimitive> {
+    let (_, regex, capture_group, json_field_path, kind) =
+        declarations.iter().find(|(name, ..)| name == param_name)?;
+    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::data_value::extract_json_field(text, field_path)?,
+        None => captures.name(capture_group)?.as_str().to_string(),
+    };
+    parse_metric_primitive(&raw, *kind)
+}
+
+/// One-time full scan building a sort helper index for `sort_by` from every currently-ingested log —
+/// used both the first time a param is sorted by (no helper index exists yet) and to transparently
+/// rebuild one ticket 061's sweep purged.
+fn build_sort_helper_index(index: &BTreeMap<IndexKey, DataValue>, sort_by: &str) -> BTreeMap<SortHelperKey, Timestamp> {
+    let compiled_params = compiled_param_declarations(index);
+    logs_range(index)
+        .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()
+}
+
+/// Incrementally extends every currently-live helper index with one freshly-ingested line, called
+/// from `EngineMessage::Ingest` right alongside the primary index insert (ticket 060) — this is what
+/// lets a sorted query avoid re-deriving params for the whole timeline on every call once a helper
+/// index for that sort key exists. A no-op when no helper index has been built yet (the common case
+/// for a param nobody has sorted by), so ingest pays nothing for sort keys that are never queried.
+fn update_sort_helper_indexes_on_ingest(
+    sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>,
+    index: &BTreeMap<IndexKey, DataValue>,
+    sequence: u64,
+    timestamp: Timestamp,
+    text: &str,
+) {
+    if sort_helper_indexes.is_empty() {
+        return;
+    }
+    let compiled_params = compiled_param_declarations(index);
+    for (sort_by, entry) in 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);
+    }
+}
+
+/// Ticket 061: drops any helper index whose `last_accessed` exceeds `SORT_HELPER_INDEX_PURGE_AFTER`.
+/// Cheap to call opportunistically on every ingest (`HashMap::retain` is a no-op scan when nothing's
+/// stale) rather than needing a dedicated timer/task — a purged index rebuilds transparently via
+/// `build_sort_helper_index` the next time something sorts by it.
+fn purge_stale_sort_helper_indexes(sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>) {
+    sort_helper_indexes.retain(|_, entry| entry.last_accessed.elapsed() < SORT_HELPER_INDEX_PURGE_AFTER);
+}
+
 fn compiled_param_declarations(
     index: &BTreeMap<IndexKey, DataValue>,
 ) -> Vec<(String, Regex, String, Option<String>, MetricPrimitiveKind)> {
@@ -1409,6 +1550,7 @@ fn handle_query(
     index: &BTreeMap<IndexKey, DataValue>,
     query: EngineQuery,
     approx_bytes_total: u64,
+    sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>,
 ) -> EngineResponse {
     match query {
         EngineQuery::Logs(params) => {
@@ -1453,34 +1595,64 @@ fn handle_query(
                         params: derive_log_params(text, &compiled_params),
                     })
                     .collect(),
-                // An alternate sort key needs every matching line's params derived up front (there's
+                // Ticket 060: walk the (lazily built-or-reused) helper index for this sort key in
-                // no way to know which lines will end up in the first `limit` before sorting), then
+                // key order instead of materializing and sorting the full filtered set — the index's
-                // reordered before the limit is finally applied.
+                // own key ordering already expresses "sorted-by-value lines, then lines missing the
+                // param ordered by ottyl_seq" (`sort_missing_fallback`'s semantics), so this just
+                // filters/limits as it walks rather than partition-sort-concat afterward.
                 Some(sort_by) => {
-                    let mut all: Vec<LogEntryDto> = filtered
+                    let helper = sort_helper_indexes.entry(sort_by.to_string()).or_insert_with(|| {
-                        .map(|(sequence, timestamp, text)| LogEntryDto {
+                        SortHelperIndexEntry {
+                            index: build_sort_helper_index(index, sort_by),
+                            last_accessed: std::time::Instant::now(),
+                        }
+                    });
+                    helper.last_accessed = std::time::Instant::now();
+
+                    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;
+                        }
+                        // A line evicted (by `max_logs`) since this key was recorded is simply
+                        // skipped, same as buckets/other derived data aren't retroactively pruned
+                        // on eviction elsewhere in this file.
+                        let Some(DataValue::LogLiteral { text }) =
+                            index.get(&IndexKey::IngestedLog { sequence, timestamp: *timestamp })
+                        else {
+                            continue;
+                        };
+                        if !compiled_filter
+                            .as_ref()
+                            .is_none_or(|filter| eval_compiled(filter, sequence, *timestamp, text))
+                        {
+                            continue;
+                        }
+                        entries.push(LogEntryDto {
                             sequence,
                             timestamp_ms: timestamp.as_millis(),
-                            text: text.to_string(),
+                            text: text.clone(),
                             params: derive_log_params(text, &compiled_params),
-                        })
+                        });
-                        .collect();
+                        if entries.len() >= limit {
-                    let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true);
+                            break;
-                    if sort_missing_fallback {
-                        let (mut with_param, mut without_param): (Vec<_>, Vec<_>) =
-                            all.drain(..).partition(|entry| entry.params.contains_key(sort_by));
-                        with_param.sort_by(|a, b| a.params[sort_by].cmp(&b.params[sort_by]));
-                        // Lines missing the sort param are only ordered relative to each other (by
-                        // `ottyl_seq`) and appended after the sorted lines, rather than interleaved
-                        // into the overall ordering — see ticket 045.
-                        without_param.sort_by_key(|entry| entry.sequence);
-                        with_param.into_iter().chain(without_param).collect()
-                    } else {
-                        all.retain(|entry| entry.params.contains_key(sort_by));
-                        all.sort_by(|a, b| a.params[sort_by].cmp(&b.params[sort_by]));
-                        all
                         }
                     }
+                    entries
+                }
             };
             let entries: Vec<LogEntryDto> = entries.into_iter().take(limit).collect();
             EngineResponse::Logs(entries)
@@ -2106,8 +2278,9 @@ mod tests {
 
     /// Runs both auto-declaration passes over every fixture line, same order as `EngineMessage::Ingest`.
     fn ingest_log_stream_fixture(index: &mut BTreeMap<IndexKey, DataValue>) {
+        let mut header = None;
         for line in LOG_STREAM_FIXTURE.lines() {
-            auto_declare_log_stream_params(index, line);
+            auto_declare_log_stream_params(index, &mut header, line);
             auto_declare_params(index, line);
         }
     }
@@ -2115,13 +2288,13 @@ mod tests {
     #[test]
     fn ingesting_log_stream_fixture_does_not_blow_up_declared_param_count() {
         // Ticket 050: unbounded auto-declaration on this real `log stream` dump produced 220 distinct
-        // params. The bound (MAX_AUTO_DECLARED_PARAMS + the 3 fixed log_stream params) must keep the
+        // params. The bound (MAX_AUTO_DECLARED_PARAMS + the header-derived log_stream columns) must
-        // total in the tens, nowhere near "hundreds".
+        // keep the total in the tens, nowhere near "hundreds".
         let mut index = BTreeMap::new();
         ingest_log_stream_fixture(&mut index);
 
         let declared = param_declarations_range(&index).count();
-        // `log_type`/`process`/`pid` are stored as `auto_derived` too, so the shared
+        // The header-derived columns (ticket 059) are stored as `auto_derived` too, so the shared
         // `MAX_AUTO_DECLARED_PARAMS` ceiling covers all of them together (see
         // `auto_declare_log_stream_params`'s doc comment).
         assert!(
@@ -2144,29 +2317,38 @@ mod tests {
     }
 
     #[test]
-    fn log_stream_shaped_lines_expose_type_process_and_pid_as_queryable_params() {
+    fn log_stream_shaped_lines_expose_header_named_columns_as_queryable_params() {
         let mut index = BTreeMap::new();
         ingest_log_stream_fixture(&mut index);
 
         let declared: HashSet<&str> = param_declarations_range(&index).map(|d| d.0).collect();
-        assert!(declared.contains("log_type"));
+        // Derived generically from the fixture's own header row (ticket 059), not a fixed set of three.
-        assert!(declared.contains("process"));
+        for name in ["Timestamp", "Thread", "Type", "Activity", "PID", "TTL", "message"] {
-        assert!(declared.contains("pid"));
+            assert!(declared.contains(name), "expected {name} to be declared, got {declared:?}");
+        }
 
         let compiled = compiled_param_declarations(&index);
         let data_line = LOG_STREAM_FIXTURE.lines().nth(1).unwrap();
         let params = derive_log_params(data_line, &compiled);
-        assert_eq!(params.get("log_type"), Some(&MetricPrimitive::String("Default".to_string())));
+        assert_eq!(params.get("Type"), Some(&MetricPrimitive::String("Default".to_string())));
-        assert_eq!(params.get("process"), Some(&MetricPrimitive::String("kitty".to_string())));
+        assert_eq!(params.get("PID"), Some(&MetricPrimitive::Number(Number::Int(1343))));
-        assert_eq!(params.get("pid"), Some(&MetricPrimitive::Number(Number::Int(1343))));
+        assert_eq!(params.get("TTL"), Some(&MetricPrimitive::Number(Number::Int(7))));
+        assert_eq!(
+            params.get("message"),
+            Some(&MetricPrimitive::String(
+                "kitty: (SkyLight) [com.apple.SkyLight:default] SLSGetDisplaysWithOpenGLDisplayMask, 3614: Display 0x00000001: Mode 54, (0 x 0) [1728 x 1117], Depth 32 BPP, active".to_string()
+            ))
+        );
     }
 
     #[test]
-    fn non_log_stream_lines_do_not_get_log_type_process_or_pid_declared() {
+    fn non_log_stream_lines_do_not_get_any_columns_declared_without_a_cached_header() {
         let mut index = BTreeMap::new();
-        auto_declare_log_stream_params(&mut index, r#"ts=1788648105773 msg="entity_moved" x1=-4"#);
+        let mut header = None;
+        auto_declare_log_stream_params(&mut index, &mut header, r#"ts=1788648105773 msg="entity_moved" x1=-4"#);
 
         assert!(param_declarations_range(&index).next().is_none());
+        assert!(header.is_none());
     }
 
     #[test]
@@ -2176,8 +2358,9 @@ mod tests {
         let mut index = BTreeMap::new();
         let start = std::time::Instant::now();
         for _ in 0..repeats {
+            let mut header = None;
             for line in LOG_STREAM_FIXTURE.lines() {
-                auto_declare_log_stream_params(&mut index, line);
+                auto_declare_log_stream_params(&mut index, &mut header, line);
                 auto_declare_params(&mut index, line);
             }
         }
@@ -3578,6 +3761,7 @@ mod tests {
                 end_ms: None,
             },
             0,
+            &mut HashMap::new(),
         );
         match response {
             EngineResponse::Promql(Err(message)) => assert!(!message.is_empty()),
@@ -3599,6 +3783,7 @@ mod tests {
                 end_ms: None,
             },
             0,
+            &mut HashMap::new(),
         );
         match response {
             EngineResponse::Promql(Ok(series)) => {
@@ -3691,6 +3876,7 @@ mod tests {
             &index,
             EngineQuery::Promql { query: "count(ottyl_seq)".to_string(), start_ms: None, end_ms: None },
             0,
+            &mut HashMap::new(),
         );
         match response {
             EngineResponse::Promql(Ok(series)) => {
@@ -3753,7 +3939,7 @@ mod tests {
         ingest_log(&mut index, 2, "still fine");
         ingest_log(&mut index, 3, "another error here");
 
-        match handle_query(&index, logs_query(Some(r#"regex("error")"#)), 0) {
+        match handle_query(&index, logs_query(Some(r#"regex("error")"#)), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![1, 3]);
@@ -3769,7 +3955,7 @@ mod tests {
             ingest_log(&mut index, i, &format!("line {i}"));
         }
 
-        match handle_query(&index, logs_query(Some("ottyl_seq<2")), 0) {
+        match handle_query(&index, logs_query(Some("ottyl_seq<2")), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![0, 1]);
@@ -3789,7 +3975,7 @@ mod tests {
             );
         }
 
-        match handle_query(&index, logs_query(Some("ottyl_ts>2000")), 0) {
+        match handle_query(&index, logs_query(Some("ottyl_ts>2000")), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![3, 4]);
@@ -3816,7 +4002,7 @@ mod tests {
         ingest_log(&mut index, 1, "priority=1 boom");
         ingest_log(&mut index, 2, "priority=2 boom");
 
-        match handle_query(&index, logs_query_sorted("priority", None), 0) {
+        match handle_query(&index, logs_query_sorted("priority", None), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![1, 2, 0]);
@@ -3844,7 +4030,7 @@ mod tests {
         ingest_log(&mut index, 2, "priority=1 boom"); // has param
         ingest_log(&mut index, 3, "also missing"); // missing
 
-        match handle_query(&index, logs_query_sorted("priority", Some(true)), 0) {
+        match handle_query(&index, logs_query_sorted("priority", Some(true)), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 // Lines with the param sorted ascending by value (2 then 0), followed by the
@@ -3870,7 +4056,7 @@ mod tests {
         ingest_log(&mut index, 2, "priority=1 boom");
         ingest_log(&mut index, 3, "also missing");
 
-        match handle_query(&index, logs_query_sorted("priority", Some(false)), 0) {
+        match handle_query(&index, logs_query_sorted("priority", Some(false)), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![2, 0]);
@@ -3884,7 +4070,7 @@ mod tests {
         let mut index = BTreeMap::new();
         ingest_log(&mut index, 0, "hello");
 
-        match handle_query(&index, logs_query_sorted("does_not_exist", None), 0) {
+        match handle_query(&index, logs_query_sorted("does_not_exist", None), 0, &mut HashMap::new()) {
             EngineResponse::Error(err) => assert!(err.contains("does_not_exist")),
             _ => panic!("expected EngineResponse::Error"),
         }
@@ -3908,7 +4094,7 @@ mod tests {
         ingest_log(&mut index, 2, "priority=1 boom");
         ingest_log(&mut index, 3, "also missing");
 
-        match handle_query(&index, logs_query(None), 0) {
+        match handle_query(&index, logs_query(None), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![0, 1, 2, 3]);
@@ -3917,6 +4103,136 @@ mod tests {
         }
     }
 
+    // -- Ticket 060: sort helper index --
+
+    #[test]
+    fn sort_helper_index_is_built_lazily_and_reused_across_queries() {
+        let mut index = BTreeMap::new();
+        declare_test_param(
+            &mut index,
+            "priority",
+            r"priority=(?P<priority>\d+)",
+            "priority",
+            MetricPrimitiveKind::Number,
+        );
+        ingest_log(&mut index, 0, "priority=3 boom");
+        ingest_log(&mut index, 1, "priority=1 boom");
+        ingest_log(&mut index, 2, "priority=2 boom");
+
+        let mut sort_helper_indexes = HashMap::new();
+        assert!(!sort_helper_indexes.contains_key("priority"));
+
+        // First query has no helper index yet — must lazily build one rather than erroring or
+        // silently returning unsorted results.
+        match handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes) {
+            EngineResponse::Logs(entries) => {
+                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
+                assert_eq!(sequences, vec![1, 2, 0]);
+            }
+            _ => panic!("expected Logs response"),
+        }
+        assert!(sort_helper_indexes.contains_key("priority"));
+        assert_eq!(sort_helper_indexes["priority"].index.len(), 3);
+
+        // Ingest more lines directly into the already-built helper index (bypassing the full
+        // `EngineMessage::Ingest` actor loop, same as every other test in this file exercises
+        // `handle_query` directly) via the same incremental-update path `run()` calls on ingest,
+        // then requery — the new lines must appear correctly ordered without the helper index
+        // having been thrown away and rebuilt from scratch.
+        ingest_log(&mut index, 3, "priority=0 boom");
+        update_sort_helper_indexes_on_ingest(
+            &mut sort_helper_indexes,
+            &index,
+            3,
+            Timestamp::from_millis(3),
+            "priority=0 boom",
+        );
+        ingest_log(&mut index, 4, "no priority here");
+        update_sort_helper_indexes_on_ingest(
+            &mut sort_helper_indexes,
+            &index,
+            4,
+            Timestamp::from_millis(4),
+            "no priority here",
+        );
+        assert_eq!(sort_helper_indexes["priority"].index.len(), 5);
+
+        match handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes) {
+            EngineResponse::Logs(entries) => {
+                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
+                // 3 (priority=0) sorts before the original three; 4 (missing the param) falls back
+                // to ottyl_seq order, appended after every line that has the param.
+                assert_eq!(sequences, vec![3, 1, 2, 0, 4]);
+            }
+            _ => panic!("expected Logs response"),
+        }
+    }
+
+    #[test]
+    fn sort_helper_index_incremental_update_is_a_noop_when_no_index_is_live() {
+        // Guards `update_sort_helper_indexes_on_ingest`'s early return: ingest must not pay to
+        // derive a param nobody has ever sorted by.
+        let mut sort_helper_indexes: HashMap<String, SortHelperIndexEntry> = HashMap::new();
+        let index = BTreeMap::new();
+        update_sort_helper_indexes_on_ingest(&mut sort_helper_indexes, &index, 0, Timestamp::from_millis(0), "hello");
+        assert!(sort_helper_indexes.is_empty());
+    }
+
+    // -- Ticket 061: purging stale sort helper indexes --
+
+    #[test]
+    fn purge_stale_sort_helper_indexes_drops_only_entries_past_the_purge_duration() {
+        let mut sort_helper_indexes: HashMap<String, SortHelperIndexEntry> = HashMap::new();
+        sort_helper_indexes.insert(
+            "stale".to_string(),
+            SortHelperIndexEntry {
+                index: BTreeMap::new(),
+                last_accessed: std::time::Instant::now() - SORT_HELPER_INDEX_PURGE_AFTER
+                    - std::time::Duration::from_secs(1),
+            },
+        );
+        sort_helper_indexes.insert(
+            "fresh".to_string(),
+            SortHelperIndexEntry { index: BTreeMap::new(), last_accessed: std::time::Instant::now() },
+        );
+
+        purge_stale_sort_helper_indexes(&mut sort_helper_indexes);
+
+        assert!(!sort_helper_indexes.contains_key("stale"));
+        assert!(sort_helper_indexes.contains_key("fresh"));
+    }
+
+    #[test]
+    fn sorted_query_transparently_rebuilds_a_purged_helper_index() {
+        let mut index = BTreeMap::new();
+        declare_test_param(
+            &mut index,
+            "priority",
+            r"priority=(?P<priority>\d+)",
+            "priority",
+            MetricPrimitiveKind::Number,
+        );
+        ingest_log(&mut index, 0, "priority=3 boom");
+        ingest_log(&mut index, 1, "priority=1 boom");
+        ingest_log(&mut index, 2, "priority=2 boom");
+
+        let mut sort_helper_indexes = HashMap::new();
+        let _ = handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes);
+        assert!(sort_helper_indexes.contains_key("priority"));
+
+        // Simulate ticket 061's sweep having purged this entry for staleness.
+        sort_helper_indexes.remove("priority");
+
+        match handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes) {
+            EngineResponse::Logs(entries) => {
+                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
+                assert_eq!(sequences, vec![1, 2, 0]);
+            }
+            _ => panic!("expected Logs response"),
+        }
+        assert!(sort_helper_indexes.contains_key("priority"));
+    }
+
     #[test]
     fn logs_query_with_ottyl_fmt_comparison_filters_by_detected_format_for_mixed_format_lines() {
         let mut index = BTreeMap::new();
@@ -3926,7 +4242,7 @@ mod tests {
         ingest_log(&mut index, 2, "just some plain prose here");
         ingest_log(&mut index, 3, r#"{"level":"info","code":200}"#);
 
-        match handle_query(&index, logs_query(Some("ottyl_fmt==json")), 0) {
+        match handle_query(&index, logs_query(Some("ottyl_fmt==json")), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![0, 3]);
@@ -3934,7 +4250,7 @@ mod tests {
             _ => panic!("expected Logs response"),
         }
 
-        match handle_query(&index, logs_query(Some("ottyl_fmt==logfmt")), 0) {
+        match handle_query(&index, logs_query(Some("ottyl_fmt==logfmt")), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![1]);
@@ -3942,7 +4258,7 @@ mod tests {
             _ => panic!("expected Logs response"),
         }
 
-        match handle_query(&index, logs_query(Some("ottyl_fmt==none")), 0) {
+        match handle_query(&index, logs_query(Some("ottyl_fmt==none")), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![2]);
@@ -3958,7 +4274,7 @@ mod tests {
         ingest_log(&mut index, 1, r#"{"level":"info","code":200}"#);
         ingest_log(&mut index, 2, "not json at all");
 
-        match handle_query(&index, logs_query(Some(r#"json[level]==error"#)), 0) {
+        match handle_query(&index, logs_query(Some(r#"json[level]==error"#)), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![0]);
@@ -3966,7 +4282,7 @@ mod tests {
             _ => panic!("expected Logs response"),
         }
 
-        match handle_query(&index, logs_query(Some("json[code]>=300")), 0) {
+        match handle_query(&index, logs_query(Some("json[code]>=300")), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![0]);
@@ -3991,6 +4307,7 @@ mod tests {
             &index,
             logs_query(Some(r#"regex("boom") and requests[level]==error"#)),
             0,
+            &mut HashMap::new(),
         ) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
@@ -4004,6 +4321,7 @@ mod tests {
             &index,
             logs_query(Some(r#"regex("boom") or requests[level]==error"#)),
             0,
+            &mut HashMap::new(),
         ) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
@@ -4020,7 +4338,7 @@ mod tests {
         ingest_log(&mut index, 0, "level=info hello");
         ingest_log(&mut index, 1, "no match here");
 
-        let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0) else {
+        let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0, &mut HashMap::new()) else {
             panic!("expected Logs response");
         };
 
@@ -4042,7 +4360,7 @@ mod tests {
         ingest_log(&mut index, 0, "an ERROR occurred");
         ingest_log(&mut index, 1, "everything is fine");
 
-        let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0) else {
+        let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0, &mut HashMap::new()) else {
             panic!("expected Logs response");
         };
 
@@ -4058,7 +4376,7 @@ mod tests {
         ingest_log(&mut index, 1, "an error occurred");
         ingest_log(&mut index, 2, "another error here");
 
-        match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000), 0) {
+        match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000), 0, &mut HashMap::new()) {
             EngineResponse::Timeline(buckets) => {
                 let total: u64 = buckets.iter().map(|b| b.count).sum();
                 assert_eq!(total, 2);
@@ -4072,12 +4390,12 @@ mod tests {
         let mut index = BTreeMap::new();
         ingest_log(&mut index, 0, "hello");
 
-        match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@")), 0) {
+        match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@")), 0, &mut HashMap::new()) {
             EngineResponse::Error(message) => assert!(!message.is_empty()),
             _ => panic!("expected an Error response for a malformed filter string"),
         }
 
-        match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000), 0) {
+        match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000), 0, &mut HashMap::new()) {
             EngineResponse::Error(message) => assert!(!message.is_empty()),
             _ => panic!("expected an Error response for a malformed filter string"),
         }
@@ -4091,7 +4409,7 @@ mod tests {
         declare_metric(&mut index, metric_request("requests", pattern, vec!["level"])).unwrap();
         ingest_log(&mut index, 0, "level=error");
 
-        match handle_query(&index, logs_query(Some("requests==5")), 0) {
+        match handle_query(&index, logs_query(Some("requests==5")), 0, &mut HashMap::new()) {
             EngineResponse::Error(message) => assert!(message.contains("param")),
             _ => panic!("expected an Error response for a metric comparison without [param]"),
         }
@@ -4106,7 +4424,7 @@ mod tests {
         declare_metric(&mut index, metric_request("requests", pattern, vec!["level", "duration"])).unwrap();
         ingest_log(&mut index, 0, "level=error duration=5");
 
-        match handle_query(&index, logs_query(Some("requests[nonexistent_param]==\"x\"")), 0) {
+        match handle_query(&index, logs_query(Some("requests[nonexistent_param]==\"x\"")), 0, &mut HashMap::new()) {
             EngineResponse::Error(message) => {
                 assert!(
                     message.contains("nonexistent_param"),
@@ -4136,7 +4454,7 @@ mod tests {
         ingest_log(&mut index, 0, "level=error duration=5");
         ingest_log(&mut index, 1, "level=info duration=2");
 
-        match handle_query(&index, logs_query(Some("requests[level]==error")), 0) {
+        match handle_query(&index, logs_query(Some("requests[level]==error")), 0, &mut HashMap::new()) {
             EngineResponse::Logs(entries) => {
                 let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                 assert_eq!(sequences, vec![0]);

src/logfmt.rs

@@ -101,16 +101,223 @@ pub fn is_log_stream_line(line: &str) -> bool {
     LOG_STREAM_RE.is_match(line)
 }
 
-/// `message_match_regex` patterns `crate::engine::auto_declare_log_stream_params` uses to declare
+/// Regex fragment for one column separator under the generic whitespace-columnar rule ticket 059
-/// `log_type`/`process`/`pid` as standalone params the first time a `log stream`-shaped line is seen.
+/// asks for: a run of 2+ whitespace chars, or a lone tab, separates columns; a single plain space does
-/// Each isolates exactly one named capture group (matching `derive_log_params`'s one-group-per-param
+/// not (so a value that legitimately contains one -- e.g. `log stream`'s `date time+tz` timestamp --
-/// convention) over the same column layout `LOG_STREAM_RE` detects.
+/// stays intact within its own column). `is_separator_run` is the same rule applied at split time;
-pub const LOG_STREAM_TYPE_PATTERN: &str =
+/// this is that rule re-expressed as a regex fragment for the trailing message field's
-    r"^\d{4}-\d{2}-\d{2}\s+\S+\s+0x[0-9a-fA-F]+\s+(?P<log_type>\S+)\s+0x[0-9a-fA-F]+\s+\d+\s+\d+\s+[^\s:]+:\s";
+/// `message_match_regex`, which re-derives its value the same way whenever a caller queries by it
-pub const LOG_STREAM_PROCESS_PATTERN: &str =
+/// later (declarations don't cache extracted values -- see `crate::engine::extract_param_raw`).
-    r"^\d{4}-\d{2}-\d{2}\s+\S+\s+0x[0-9a-fA-F]+\s+\S+\s+0x[0-9a-fA-F]+\s+\d+\s+\d+\s+(?P<process>[^\s:]+):\s";
+const COLUMN_SEPARATOR_PATTERN: &str = r"(?:\t|[ \t]{2,})";
-pub const LOG_STREAM_PID_PATTERN: &str =
+
-    r"^\d{4}-\d{2}-\d{2}\s+\S+\s+0x[0-9a-fA-F]+\s+\S+\s+0x[0-9a-fA-F]+\s+(?P<pid>\d+)\s+\d+\s+[^\s:]+:\s";
+/// A column's value: one or more non-whitespace tokens joined by single spaces, with no upper bound
+/// on length. Used for the trailing (unbounded-width) message field and the last named column, where
+/// there's no next column's start position to bound against.
+const COLUMN_VALUE_PATTERN: &str = r"\S+(?: \S+)*";
+
+/// Same idea as `COLUMN_VALUE_PATTERN`, but bounded to a known column `width` (in chars) -- needed for
+/// every column except the last, since two adjacent `log stream` values can legitimately be separated
+/// by only a single space when the first one nearly fills its column's on-screen width (e.g. the
+/// `Timestamp` column: its own `date time+tz` value already contains one internal space, so a
+/// content-only pattern can't tell that boundary apart from the one-space gap before the next column
+/// starts -- seed this from `LogStreamColumn::start` positions instead). Greedily matches up to
+/// `width` chars while requiring the match to end on a non-whitespace char, which is what excludes the
+/// column's own trailing padding spaces from the capture.
+fn bounded_column_value_pattern(width: usize) -> String {
+    if width <= 1 {
+        r"\S".to_string()
+    } else {
+        format!(r"\S(?:.{{0,{}}}\S)?", width - 2)
+    }
+}
+
+fn is_separator_run(run: &str) -> bool {
+    run.len() >= 2 || run.contains('\t')
+}
+
+static WHITESPACE_RUN: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"[ \t]+").expect("static regex must compile"));
+
+/// Splits `line` at whitespace runs that qualify as a column separator (`is_separator_run`), returning
+/// each piece alongside its byte offset in `line`; a lone plain space never splits. Consumes at most
+/// `max_columns.map(|n| n - 1)` separators -- the final piece returned is the untouched remainder of
+/// the line, including any further whitespace runs inside it. That's what lets a trailing unlabelled
+/// `log stream` message tail (whose own internal spacing can be arbitrary) come back as one field
+/// instead of being chopped up by it.
+fn split_columns_with_offsets(line: &str, max_columns: Option<usize>) -> Vec<(usize, &str)> {
+    let mut columns = Vec::new();
+    let mut start = 0usize;
+    let mut consumed = 0usize;
+    for m in WHITESPACE_RUN.find_iter(line) {
+        if max_columns.is_some_and(|max| consumed + 1 >= max) {
+            break;
+        }
+        let run = &line[m.start()..m.end()];
+        if is_separator_run(run) {
+            columns.push((start, &line[start..m.start()]));
+            start = m.end();
+            consumed += 1;
+        }
+    }
+    columns.push((start, &line[start..]));
+    columns
+}
+
+fn split_columns(line: &str, max_columns: Option<usize>) -> Vec<&str> {
+    split_columns_with_offsets(line, max_columns).into_iter().map(|(_, s)| s).collect()
+}
+
+/// A column-name identifier a generated `message_match_regex`'s named capture group can use as-is
+/// (the regex crate requires `[A-Za-z_][0-9A-Za-z_]*`).
+static COLUMN_NAME: LazyLock<Regex> =
+    LazyLock::new(|| Regex::new(r"^[A-Za-z][A-Za-z0-9_]*$").expect("static regex must compile"));
+
+/// One column named by a detected `log stream` header (ticket 059). `start` is that column's position
+/// in `char`s (not bytes -- see `char_slice`) within the header line -- meaningful only when
+/// `LogStreamHeader::tab_delimited` is `false` (see that field's doc comment).
+pub struct LogStreamColumn {
+    pub name: String,
+    start: usize,
+}
+
+/// A detected `log stream` header: its named columns, plus whether they were tab- or space-separated.
+/// `log stream`'s real output right-pads each header name to a fixed on-screen width using runs of
+/// 2+ spaces, and every data line's values land in that same fixed alignment regardless of how much
+/// of that width they individually fill -- so for that shape, a data line's column boundaries are
+/// best found at the header's own fixed character positions, not by re-splitting the data line's own
+/// whitespace (see `split_log_stream_row`'s doc comment for why that alone isn't reliable there).
+/// A tab-separated header carries no such alignment guarantee (values aren't padded to a shared
+/// width), but a lone tab is itself an unambiguous separator, so that case instead re-splits each data
+/// line directly.
+pub struct LogStreamHeader {
+    pub columns: Vec<LogStreamColumn>,
+    tab_delimited: bool,
+}
+
+/// Byte-indexed `str` slicing doesn't work directly with `LogStreamColumn::start` (a *char* offset),
+/// since `log stream` output is expected to stay ASCII in practice but callers shouldn't panic if a
+/// message contains multi-byte chars before the slice point. `end` of `None` slices to end of string.
+fn char_slice(s: &str, start: usize, end: Option<usize>) -> &str {
+    if end.is_some_and(|end| end <= start) {
+        return "";
+    }
+    let mut indices = s.char_indices();
+    let byte_start = indices.nth(start).map(|(i, _)| i).unwrap_or(s.len());
+    let byte_end = match end {
+        Some(end) => s.char_indices().nth(end).map(|(i, _)| i).unwrap_or(s.len()),
+        None => s.len(),
+    };
+    &s[byte_start..byte_end.max(byte_start)]
+}
+
+/// Recognizes a `log stream`-style header row generically (ticket 059), instead of assuming the fixed
+/// `Timestamp Thread Type Activity PID TTL` names: a short run of bareword column names separated by
+/// qualifying whitespace runs (`split_columns_with_offsets`) that is, critically, *not itself* a
+/// `log stream` data line (`is_log_stream_line`) -- so a header is never mistaken for the data line
+/// immediately following it. `crate::engine::auto_declare_log_stream_params` caches the result across
+/// ingested lines, since `log stream` prints its header once, before any data line, and this only ever
+/// sees one line at a time.
+pub fn detect_log_stream_header(line: &str) -> Option<LogStreamHeader> {
+    if is_log_stream_line(line) {
+        return None;
+    }
+    let mut columns = Vec::new();
+    for (byte_start, slice) in split_columns_with_offsets(line, None) {
+        let trimmed = slice.trim();
+        if trimmed.is_empty() {
+            continue;
+        }
+        let leading_ws = slice.len() - slice.trim_start().len();
+        let start = line[..byte_start + leading_ws].chars().count();
+        columns.push(LogStreamColumn { name: trimmed.to_string(), start });
+    }
+    if columns.len() < 2 || !columns.iter().all(|c| COLUMN_NAME.is_match(&c.name)) {
+        return None;
+    }
+    let tab_delimited = WHITESPACE_RUN.find_iter(line).any(|m| m.as_str().contains('\t'));
+    Some(LogStreamHeader { columns, tab_delimited })
+}
+
+/// Splits a `log stream` data line into `header`'s named columns plus, when present, one trailing
+/// unlabelled tail -- `crate::engine::auto_declare_log_stream_params` maps the first
+/// `header.columns.len()` pieces onto the header's names and the optional last piece onto the
+/// `message` param.
+///
+/// For a space-padded fixed-width header, every column but the last is sliced at the header's own
+/// fixed character positions (`LogStreamColumn::start`) rather than by re-splitting the data line's
+/// own whitespace, since two adjacent values can legitimately be joined by only a single space when
+/// the first nearly fills its column's width (e.g. `log stream`'s `Timestamp` column, whose value
+/// already contains one internal space). Only the last named column's end is whitespace-run-detected
+/// (`split_columns`), since that's what distinguishes its own value from the free-text message tail
+/// `log stream` appends after it.
+///
+/// For a tab-delimited header, there's no shared on-screen width to align against, but a lone tab is
+/// itself unambiguous, so this instead re-splits the data line directly on qualifying separators.
+pub fn split_log_stream_row<'a>(line: &'a str, header: &LogStreamHeader) -> Vec<&'a str> {
+    if header.tab_delimited {
+        return split_columns(line, Some(header.columns.len() + 1));
+    }
+    let mut result = Vec::with_capacity(header.columns.len() + 1);
+    for (i, column) in header.columns.iter().enumerate() {
+        match header.columns.get(i + 1) {
+            Some(next) => result.push(char_slice(line, column.start, Some(next.start)).trim()),
+            None => {
+                let tail = char_slice(line, column.start, None);
+                let pieces = split_columns(tail, Some(2));
+                result.push(pieces[0].trim());
+                if let Some(message) = pieces.get(1) {
+                    result.push(message);
+                }
+            }
+        }
+    }
+    result
+}
+
+/// Builds the `message_match_regex` for the `index`-th (0-based) of `header`'s named columns.
+///
+/// Fixed-width case: skips exactly `header.columns[index].start` chars, then captures this column's
+/// own value bounded to its known width (`bounded_column_value_pattern`) -- unbounded for the last
+/// column, which has no next column to bound against.
+///
+/// Tab-delimited case: skips `index` preceding columns' own value+separator pairs instead, since
+/// there's no shared fixed width to anchor a character-position skip against.
+///
+/// Anchored at the start of the line since a `log stream` line never carries this fixed-per-run
+/// column prefix anywhere but the very start.
+pub fn log_stream_column_pattern(header: &LogStreamHeader, index: usize, capture_group: &str) -> String {
+    if header.tab_delimited {
+        let mut pattern = String::from("^");
+        for _ in 0..index {
+            pattern.push_str(COLUMN_VALUE_PATTERN);
+            pattern.push_str(COLUMN_SEPARATOR_PATTERN);
+        }
+        pattern.push_str(&format!("(?P<{capture_group}>{COLUMN_VALUE_PATTERN})"));
+        return pattern;
+    }
+    let start = header.columns[index].start;
+    let value_pattern = match header.columns.get(index + 1) {
+        Some(next) => bounded_column_value_pattern(next.start.saturating_sub(start)),
+        None => COLUMN_VALUE_PATTERN.to_string(),
+    };
+    format!("^.{{{start}}}(?P<{capture_group}>{value_pattern})")
+}
+
+/// Builds the `message_match_regex` for the trailing unlabelled tail after `header`'s last named
+/// column: everything from the end of that column's own separator to end of line, verbatim -- its own
+/// internal whitespace, however irregular, is never treated as a further separator.
+pub fn log_stream_message_pattern(header: &LogStreamHeader, capture_group: &str) -> String {
+    if header.tab_delimited {
+        let mut pattern = String::from("^");
+        for _ in 0..header.columns.len() {
+            pattern.push_str(COLUMN_VALUE_PATTERN);
+            pattern.push_str(COLUMN_SEPARATOR_PATTERN);
+        }
+        pattern.push_str(&format!("(?P<{capture_group}>.*)$"));
+        return pattern;
+    }
+    let start = header.columns.last().map(|c| c.start).unwrap_or(0);
+    format!("^.{{{start}}}{COLUMN_VALUE_PATTERN}{COLUMN_SEPARATOR_PATTERN}(?P<{capture_group}>.*)$")
+}
 
 /// Go `log/slog` `NewJSONHandler` output (the `go-log` label): a single-line JSON object whose first
 /// three fields are always `time`, `level`, `msg` in that fixed order (slog's `JSONHandler` writes the
@@ -290,15 +497,90 @@ mod tests {
         assert!(!is_log_stream_line(header));
     }
 
+    fn column_names(header: &LogStreamHeader) -> Vec<&str> {
+        header.columns.iter().map(|c| c.name.as_str()).collect()
+    }
+
     #[test]
-    fn log_stream_patterns_capture_type_process_and_pid() {
+    fn detects_log_stream_header_columns_generically() {
+        let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap();
+        let header = detect_log_stream_header(header_line).expect("fixture header must be detected");
+        assert_eq!(column_names(&header), vec!["Timestamp", "Thread", "Type", "Activity", "PID", "TTL"]);
+        assert!(!header.tab_delimited);
+    }
+
+    #[test]
+    fn a_log_stream_data_line_is_never_detected_as_its_own_header() {
+        let data_line = LOG_STREAM_FIXTURE.lines().nth(1).unwrap();
+        assert!(detect_log_stream_header(data_line).is_none());
+    }
+
+    #[test]
+    fn plain_prose_is_not_detected_as_a_log_stream_header() {
+        assert!(detect_log_stream_header("just some plain prose here").is_none());
+        assert!(detect_log_stream_header("ts=1788648105773 msg=\"entity_moved\"").is_none());
+    }
+
+    #[test]
+    fn splits_log_stream_row_into_named_columns_plus_trailing_message() {
+        let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap();
+        let header = detect_log_stream_header(header_line).unwrap();
+        let data_line = "2026-09-06 01:48:54.518117+0100 0x4dfed2   Default     0x0                  606    0    WindowServer: (HSTouchHIDService) something happened";
+        let row = split_log_stream_row(data_line, &header);
+        assert_eq!(row.len(), header.columns.len() + 1);
+        assert_eq!(row[0], "2026-09-06 01:48:54.518117+0100");
+        assert_eq!(row[1], "0x4dfed2");
+        assert_eq!(row[2], "Default");
+        assert_eq!(row[3], "0x0");
+        assert_eq!(row[4], "606");
+        assert_eq!(row[5], "0");
+        // The message tail is never chopped up by its own internal single/double spacing.
+        assert_eq!(row[6], "WindowServer: (HSTouchHIDService) something happened");
+    }
+
+    #[test]
+    fn a_single_tab_separates_columns_exactly_like_a_run_of_spaces() {
+        let header = detect_log_stream_header("Timestamp\tThread\tType").unwrap();
+        assert_eq!(column_names(&header), vec!["Timestamp", "Thread", "Type"]);
+        assert!(header.tab_delimited);
+
+        // Two tabs in a row still qualify as a single separator (like a run of 2+ spaces), so this
+        // still cleanly separates Timestamp/Thread/Type despite the doubled tab before "Default".
+        let spaced = split_log_stream_row("2026-09-06 01:48:54\t0x4dfed2\t\tDefault trailing message", &header);
+        let tabbed = split_log_stream_row("2026-09-06 01:48:54\t0x4dfed2\tDefault trailing message", &header);
+        assert_eq!(spaced[0], "2026-09-06 01:48:54");
+        assert_eq!(spaced[1], "0x4dfed2");
+        assert_eq!(spaced[2], "Default trailing message");
+        assert_eq!(tabbed[0], "2026-09-06 01:48:54");
+        assert_eq!(tabbed[1], "0x4dfed2");
+        assert_eq!(tabbed[2], "Default trailing message");
+    }
+
+    #[test]
+    fn column_and_message_patterns_capture_the_same_fields_split_log_stream_row_finds() {
+        let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap();
+        let header = detect_log_stream_header(header_line).unwrap();
         let line = "2026-09-06 01:48:54.518117+0100 0x4dfed2   Default     0x0                  606    0    WindowServer: (HSTouchHIDService) something happened";
-        let type_re = Regex::new(LOG_STREAM_TYPE_PATTERN).unwrap();
+
-        let process_re = Regex::new(LOG_STREAM_PROCESS_PATTERN).unwrap();
+        let type_re = Regex::new(&log_stream_column_pattern(&header, 2, "Type")).unwrap();
-        let pid_re = Regex::new(LOG_STREAM_PID_PATTERN).unwrap();
+        assert_eq!(type_re.captures(line).unwrap().name("Type").unwrap().as_str(), "Default");
-        assert_eq!(type_re.captures(line).unwrap().name("log_type").unwrap().as_str(), "Default");
+
-        assert_eq!(process_re.captures(line).unwrap().name("process").unwrap().as_str(), "WindowServer");
+        let pid_re = Regex::new(&log_stream_column_pattern(&header, 4, "PID")).unwrap();
-        assert_eq!(pid_re.captures(line).unwrap().name("pid").unwrap().as_str(), "606");
+        assert_eq!(pid_re.captures(line).unwrap().name("PID").unwrap().as_str(), "606");
+
+        let message_re = Regex::new(&log_stream_message_pattern(&header, "message")).unwrap();
+        assert_eq!(
+            message_re.captures(line).unwrap().name("message").unwrap().as_str(),
+            "WindowServer: (HSTouchHIDService) something happened"
+        );
+
+        // Also exercise the Timestamp column specifically -- its own value contains one internal
+        // space, so a naive content-only pattern would previously over-consume into the next column.
+        let timestamp_re = Regex::new(&log_stream_column_pattern(&header, 0, "Timestamp")).unwrap();
+        assert_eq!(
+            timestamp_re.captures(line).unwrap().name("Timestamp").unwrap().as_str(),
+            "2026-09-06 01:48:54.518117+0100"
+        );
     }
 
     #[test]

tickets/059-log-stream-parsing-should-be-generic-whitespace-columnar.md

@@ -1,62 +0,0 @@
-# 059 — `log stream` parsing is one hardcoded regex, not generic whitespace-columnar detection
-
-## Summary
-
-Current `log stream` support is a single hardcoded regex,
-`LOG_STREAM_RE` (`src/logfmt.rs:93-113`), matching exactly six fixed
-columns (`Timestamp Thread Type Activity PID TTL`) by content shape
-(dates/hex/decimals) rather than by parsing the header row generically.
-`auto_declare_log_stream_params` (`src/engine.rs:451-472`) then extracts
-only three hardcoded fields (`log_type`, `process`, `pid`) via three more
-bespoke regexes (`LOG_STREAM_TYPE_PATTERN` etc., `logfmt.rs:108-113`).
-
-The fixture `src/fixtures/log_stream_sample.txt` has a real header row
-(`Timestamp                       Thread     Type        Activity             PID    TTL  `)
-whose columns are separated by runs of 2+ spaces — the general CSV-like
-shape the wishlist asks for — plus a trailing unlabelled `process:
-message` tail that must not get merged into the preceding column. Today
-nothing derives column names or positions from that header; every field
-is matched by a fixed, hand-tuned regex, so any deviation from this
-exact `log stream` output shape silently fails to parse.
-
-## Scope
-
-- Replace `LOG_STREAM_RE`/`LOG_STREAM_*_PATTERN`'s fixed-shape matching
-  with a generic whitespace-columnar detector: given a header line,
-  split on runs of duplicated whitespace (2+ spaces, or a single tab, or
-  " \t " — any whitespace repeated) to determine column names and
-  positions, then apply the same column boundaries to subsequent data
-  lines.
-  - The header's last named column's value must still capture through
-    to end-of-line without swallowing the trailing unlabelled message
-    column that `log stream` appends after the last header-named field
-    (i.e. that trailing text becomes its own unlabelled/message param,
-    not merged into the last header column).
-  - Single tabs are acceptable separators even where surrounding spaces
-    aren't doubled.
-- Wire the new column-driven extraction into
-  `auto_declare_log_stream_params` in place of the three bespoke
-  per-field regexes, so declared params are derived from whatever
-  columns the header actually names, not a fixed list of three.
-- Use `src/fixtures/log_stream_sample.txt` as the reference fixture and
-  add/extend tests in `src/logfmt.rs`/`src/engine.rs` asserting header
-  columns are correctly detected and the trailing message field is kept
-  separate.
-- Out of scope: changing param-cardinality bounding introduced in ticket
-  050/054 (commit f48d3c6f) — this is purely about how columns are
-  detected/split, not about pruning noisy auto-declarations.
-
-## Acceptance criteria
-
-- Piping real `log stream` output (as in `log_stream_sample.txt`)
-  through ottyl declares params matching the header's actual column
-  names, not the previous fixed three.
-- The trailing unlabelled message text on each `log stream` line is
-  captured as its own field and never merged into the last named
-  header column.
-- A header/data line whose columns are separated by a single tab is
-  parsed identically to one separated by 2+ spaces.
-
-## Depends on
-
-None.

tickets/060-timeline-custom-sort-needs-helper-btreemap-index.md

@@ -1,57 +0,0 @@
-# 060 — Timeline sort-by-property does a full materialize+sort per query instead of using a helper index
-
-## Summary
-
-Sorting the timeline by an arbitrary property does work today (from
-ticket 045): in the `EngineQuery::Logs` branch (`src/engine.rs:1408-1487`),
-when `sort_by` is set, every filtered line is materialized into a
-`Vec<LogEntryDto>` (deriving params for all of them), then sorted
-in-memory with `.sort_by(...)`, with `sort_missing_fallback` (default
-true) partitioning entries that have the sort param from those that
-don't and ordering the latter by `ottyl_seq` (`:1468-1482`).
-
-The problem is architectural: there is no incremental secondary index
-keyed by the sort property, only the primary `BTreeMap` keyed by
-`IndexKey::IngestedLog { sequence, timestamp }` (`src/index_key.rs`,
-inserted at `engine.rs:156-181`). Every sorted query re-derives params
-for and re-sorts the *entire* filtered range from scratch, and range
-selection (e.g. "give me the next page in this order") has no way to
-seek directly into a sorted position — it can only sort everything then
-slice.
-
-## Scope
-
-- Introduce a "helper index": a secondary `BTreeMap` keyed by
-  `(sort-param-value, ottyl_seq)` (or equivalent composite key so
-  `ottyl_seq` is the tie-break/fallback ordering baked into the key
-  itself, not a post-hoc partition step) per distinct param name that's
-  been sorted by.
-- Populate a helper index incrementally as lines are ingested (near
-  `engine.rs:156-181`, alongside the primary index insert), for any
-  param name that currently has a live helper index, rather than
-  rebuilding from scratch per query.
-- Lines missing the sort param must still be captured in the helper
-  index's ordering, falling back to `ottyl_seq` ordering — matching
-  current `sort_missing_fallback` semantics but expressed as part of the
-  index's key/ordering rather than a separate partition-and-append step.
-- Adapt the `EngineQuery::Logs` sorted branch (`engine.rs:1459-1483`) to
-  do a range walk over the relevant helper index instead of
-  materializing and sorting the full filtered set every time.
-- Out of scope: the eviction/purge policy for helper indexes that
-  haven't been used recently — that's ticket 061.
-
-## Acceptance criteria
-
-- Querying the timeline sorted by a declared param returns results
-  ordered correctly, including lines missing that param falling back to
-  `ottyl_seq` order, matching current behavior.
-- A sorted query no longer re-sorts the full filtered range from scratch
-  on every call once a helper index for that sort key exists — verified
-  by a test/benchmark showing incremental maintenance rather than
-  full materialize-then-sort.
-- Existing sort-related tests (from ticket 045) continue to pass
-  unmodified in their assertions.
-
-## Depends on
-
-None.

tickets/061-purge-unused-sort-helper-indexes.md

@@ -1,43 +0,0 @@
-# 061 — Sort helper indexes need to be purged after a period of no access
-
-## Summary
-
-Ticket 060 introduces per-sort-key helper `BTreeMap` indexes. Left
-unbounded, a user (or dashboard) that sorts by many distinct params over
-a session accumulates a helper index per param forever, each duplicating
-the full timeline's worth of keys. There's no existing time-since-access
-eviction pattern in `engine.rs` to model this on: the only eviction today
-is `evict_excess_logs` (`src/engine.rs:1356-1363`, called from ingest at
-`:181` and from `SetMaxLogs` at `:212`), which is a hard count-cap
-("drop oldest once over `max_logs`"), not an LRU/last-accessed scheme.
-
-## Scope
-
-- Track a last-accessed timestamp per helper index (e.g.
-  `HashMap<String, (BTreeMap<...>, Instant)>` or equivalent), updated
-  whenever a query actually reads from that index.
-- Add a sweep (triggered periodically, or opportunistically on
-  ingest/query — match whichever fits the existing engine actor's
-  message-loop structure) that drops any helper index whose
-  last-accessed time exceeds a fixed duration.
-- A dropped helper index must be transparently rebuildable: the next
-  query sorted by that param should lazily rebuild the index (a one-time
-  full scan) exactly as if it had never existed, rather than erroring or
-  falling back permanently to unsorted behavior.
-- Out of scope: making the purge duration user-configurable via
-  `Settings` — pick a reasonable fixed constant unless the ticket
-  implementer finds a strong reason to expose it.
-
-## Acceptance criteria
-
-- A helper index untouched for longer than the purge duration is
-  removed from engine memory.
-- Querying a sort key whose helper index was purged transparently
-  rebuilds it and returns correct results, with no user-visible error.
-- Actively-used helper indexes (queried more recently than the purge
-  duration) are never evicted.
-
-## Depends on
-
-Ticket 060 (timeline custom sort needs helper BTreeMap index) — this
-ticket assumes helper indexes already exist.