Add sort_by ordering for the log viewer, with a missing-param fallback/drop toggle (ticket 045)

/api/logs now accepts sort_by (an already-declared param name) and
sort_missing_fallback: lines carrying that param are ordered by its value,
while lines missing it are either appended in ottyl_seq order (fallback,
the default) or dropped from the response. Default ottyl_ts-tracking
order is unchanged when sort_by is omitted.

The log viewer (both the inline panel and its popout) exposes this as a
"Sort by" control next to the column picker, persisted to localStorage
under one shared key (ottyl.logPreview.sort) — the same global-viewer-
setting convention already used for visible param columns and the log
viewer's height, since there's no per-dashboard scope for this to divide
by.

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

src/engine.rs

@@ -1220,10 +1220,17 @@ fn handle_query(
                 },
                 None => None,
             };
+            // Ticket 045: `sort_by` must name an already-declared param — reject anything else up
+            // front rather than silently falling through to "every line is missing it".
+            if let Some(sort_by) = params.sort_by.as_deref() {
+                if !param_declarations_range(index).any(|(name, ..)| name == sort_by) {
+                    return EngineResponse::Error(format!("unknown param '{sort_by}' for sort_by"));
+                }
+            }
             let limit = params.limit.unwrap_or(500).min(10_000);
             // O(lines_returned * declared_params) per query — fine at v1 scale; if volume/params grow, move to incremental derivation at ingest time like apply_all_metrics does for buckets.
             let compiled_params = compiled_param_declarations(index);
-            let entries: Vec<LogEntryDto> = logs_range(index)
+            let filtered = logs_range(index)
                 .filter(|(sequence, timestamp, _)| {
                     params.since_sequence.is_none_or(|since| *sequence > since)
                         && params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
@@ -1233,15 +1240,50 @@ fn handle_query(
                     compiled_filter
                         .as_ref()
                         .is_none_or(|filter| eval_compiled(filter, *sequence, *timestamp, text))
-                })
-                .take(limit)
-                .map(|(sequence, timestamp, text)| LogEntryDto {
-                    sequence,
-                    timestamp_ms: timestamp.as_millis(),
-                    text: text.to_string(),
-                    params: derive_log_params(text, &compiled_params),
-                })
-                .collect();
+                });
+
+            let entries: Vec<LogEntryDto> = match params.sort_by.as_deref() {
+                // Default path: unchanged from pre-045 behavior — take(limit) before mapping, so
+                // only the entries actually returned pay the per-line param-derivation cost.
+                None => filtered
+                    .take(limit)
+                    .map(|(sequence, timestamp, text)| LogEntryDto {
+                        sequence,
+                        timestamp_ms: timestamp.as_millis(),
+                        text: text.to_string(),
+                        params: derive_log_params(text, &compiled_params),
+                    })
+                    .collect(),
+                // An alternate sort key needs every matching line's params derived up front (there's
+                // no way to know which lines will end up in the first `limit` before sorting), then
+                // reordered before the limit is finally applied.
+                Some(sort_by) => {
+                    let mut all: Vec<LogEntryDto> = filtered
+                        .map(|(sequence, timestamp, text)| LogEntryDto {
+                            sequence,
+                            timestamp_ms: timestamp.as_millis(),
+                            text: text.to_string(),
+                            params: derive_log_params(text, &compiled_params),
+                        })
+                        .collect();
+                    let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true);
+                    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
+                    }
+                }
+            };
+            let entries: Vec<LogEntryDto> = entries.into_iter().take(limit).collect();
             EngineResponse::Logs(entries)
         }
         EngineQuery::Timeline(params) => {
@@ -2972,6 +3014,21 @@ mod tests {
             end_ms: None,
             limit: None,
             filter: filter.map(str::to_string),
+            sort_by: None,
+            sort_missing_fallback: None,
+        })
+    }
+
+    /// Ticket 045: like `logs_query`, but with an alternate sort key and its missing-param handling.
+    fn logs_query_sorted(sort_by: &str, sort_missing_fallback: Option<bool>) -> EngineQuery {
+        EngineQuery::Logs(LogsParams {
+            since_sequence: None,
+            start_ms: None,
+            end_ms: None,
+            limit: None,
+            filter: None,
+            sort_by: Some(sort_by.to_string()),
+            sort_missing_fallback,
         })
     }
 
@@ -3037,6 +3094,125 @@ mod tests {
         }
     }
 
+    // -- Ticket 045: sort_by / sort_missing_fallback --
+
+    #[test]
+    fn logs_query_sort_by_orders_lines_by_declared_param_value() {
+        let mut index = BTreeMap::new();
+        declare_test_param(
+            &mut index,
+            "priority",
+            r"priority=(?P<priority>\d+)",
+            "priority",
+            MetricPrimitiveKind::Number,
+        );
+        // Ingested (and therefore default-ordered) as sequence 0, 1, 2, with priority values
+        // deliberately out of that order so a passing test can only mean the sort actually ran.
+        ingest_log(&mut index, 0, "priority=3 boom");
+        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) {
+            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"),
+        }
+    }
+
+    /// Seeded fixture mixing lines that carry the sort param with lines that don't, per the ticket
+    /// 045 acceptance criteria: with the fallback toggle on, lines missing the param are ordered by
+    /// `ottyl_seq` relative to each other (and appended after the sorted lines); with it off, they're
+    /// excluded from the response entirely.
+    #[test]
+    fn logs_query_sort_by_missing_param_fallback_orders_by_sequence_and_appends() {
+        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"); // has param
+        ingest_log(&mut index, 1, "no param here"); // missing
+        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) {
+            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
+                // missing-param lines in ottyl_seq order (1 then 3).
+                assert_eq!(sequences, vec![2, 0, 1, 3]);
+            }
+            _ => panic!("expected Logs response"),
+        }
+    }
+
+    #[test]
+    fn logs_query_sort_by_missing_param_drop_excludes_them() {
+        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, "no param here");
+        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) {
+            EngineResponse::Logs(entries) => {
+                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
+                assert_eq!(sequences, vec![2, 0]);
+            }
+            _ => panic!("expected Logs response"),
+        }
+    }
+
+    #[test]
+    fn logs_query_sort_by_unknown_param_returns_error() {
+        let mut index = BTreeMap::new();
+        ingest_log(&mut index, 0, "hello");
+
+        match handle_query(&index, logs_query_sorted("does_not_exist", None), 0) {
+            EngineResponse::Error(err) => assert!(err.contains("does_not_exist")),
+            _ => panic!("expected EngineResponse::Error"),
+        }
+    }
+
+    /// Explicit regression check for the ticket 045 acceptance criterion that default ordering
+    /// (no `sort_by`) is unchanged: same fixture as the fallback/drop tests above, but with
+    /// `sort_by: None` the response stays in plain ingest (`ottyl_seq`) order.
+    #[test]
+    fn logs_query_without_sort_by_keeps_default_ottyl_ts_order() {
+        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, "no param here");
+        ingest_log(&mut index, 2, "priority=1 boom");
+        ingest_log(&mut index, 3, "also missing");
+
+        match handle_query(&index, logs_query(None), 0) {
+            EngineResponse::Logs(entries) => {
+                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
+                assert_eq!(sequences, vec![0, 1, 2, 3]);
+            }
+            _ => panic!("expected Logs response"),
+        }
+    }
+
     #[test]
     fn logs_query_with_ottyl_fmt_comparison_filters_by_detected_format_for_mixed_format_lines() {
         let mut index = BTreeMap::new();