Bound auto-declared param cardinality and recognize log stream format (ticket 050/054)

auto_declare_params (src/engine.rs) previously never overwrote or pruned a
declaration, so a burst of unusual key=value-shaped lines could permanently
inflate the params UI forever. A real macOS `log stream` dump
(huig_dump.txt, copied into src/fixtures/log_stream_sample.txt) discovered
220 distinct auto-declared params from under 700 lines, mostly one-off
WiFi/radio stat counters (20+ pairs on a single line), plus base64-blob
false positives from key()'s greedy alnum matching landing right before
`==` padding.

Fix, combining two bounds per the ticket's options:
- src/logfmt.rs: extract_pairs/pair now rejects a candidate pair whose
  bareword value starts with a bare '=' (looks_like_base64_padding_artifact)
  — the exact shape produced when a base64/hash blob's padding sits right
  after a run of alnum chars that key() consumed as if it were a key.
- src/engine.rs: auto_declare_params now enforces a per-line cap
  (MAX_NEW_AUTO_PARAMS_PER_LINE = 8) on new params from one line, and a
  total ceiling (MAX_AUTO_DECLARED_PARAMS = 64) on auto-derived params
  overall — the per-line cap alone doesn't bound unbounded key *variety*
  across many lines, so both are needed.

Also adds structured extraction for log stream's fixed-column shape
(Timestamp Thread Type Activity PID TTL, then `process: message`):
detect_format gains a new "log_stream" ottyl_fmt value (checked first,
since it's the most specific of the four), and
auto_declare_log_stream_params declares log_type/process/pid as standalone
queryable params the first time such a line is seen — derived positionally
from the column layout rather than =-scanning, so there's no ambiguity or
false-positive risk. This is a separate always-attempted extraction
alongside the existing key=value pass, not a replacement for it: a log
stream line's message tail can still contain real key=value pairs (e.g.
the WiFi stat lines) that auto_declare_params keeps discovering
independently. This also fully addresses ticket 054, confirming huig_dump.txt
is space-padded fixed-width columns, not TSV, and that no separate TSV
format is needed.

Throughput (src/fixtures/log_stream_sample.txt x50, release build):
before (caps disabled) ~369k lines/sec, 206 auto-declared params;
after (caps enabled) ~1.20M lines/sec, 64 auto-declared params — about
3.2x faster, confirming the ever-growing declaration set was the
bottleneck rather than per-line regex/nom scanning cost.

Removes tickets/050 and tickets/054 now that both are implemented.

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

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_log_stream_params(&mut index, &text);
                 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);
@@ -298,14 +299,46 @@ fn build_auto_param_pattern(key: &str, group: &str, quoted: bool) -> String {
     }
 }
 
+/// Hard ceiling on how many params `auto_declare_params` will ever create automatically over the
+/// life of the engine. Without this, one unusual input can permanently inflate the declared-param
+/// set forever — `auto_declare_params` never overwrites or prunes a declaration, so every distinct
+/// key ever seen sticks around. Ticket 050's real `log stream` dump discovered 220 distinct keys from
+/// under 700 lines, mostly one-off WiFi/radio stat counters (e.g. `txRTSFrm`/`rxCTSUcast`/...) that
+/// will never recur meaningfully. Chosen alongside `MAX_NEW_AUTO_PARAMS_PER_LINE` (below) rather than
+/// on its own: a per-line cap alone still lets an unbounded *variety* of lines each contribute a few
+/// new keys forever, so this total ceiling is the actual backstop.
+const MAX_AUTO_DECLARED_PARAMS: usize = 64;
+
+/// Per-line ceiling on how many *new* params a single ingested line may contribute to the auto-derived
+/// set. Without this, one unusually wide line (the fixture's WiFi stat dump has 20+ `key=value` pairs
+/// in a single line) could burn most of `MAX_AUTO_DECLARED_PARAMS` in one shot, crowding out keys from
+/// more representative lines ingested later. Pairs beyond the cap are simply never declared (from this
+/// line or any other, since a later line with fewer pairs would still need to reach them within its own
+/// per-line budget) — acceptable since they're exactly the low-value, one-off counters ticket 050 flags.
+const MAX_NEW_AUTO_PARAMS_PER_LINE: usize = 8;
+
+fn auto_derived_param_count(index: &BTreeMap<IndexKey, DataValue>) -> usize {
+    param_declarations_range(index).filter(|declaration| declaration.6).count()
+}
+
 /// 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.
+/// Bounded by `MAX_AUTO_DECLARED_PARAMS`/`MAX_NEW_AUTO_PARAMS_PER_LINE` (see their doc comments) so a
+/// single burst of unusual lines can't permanently inflate the declared-param set (ticket 050).
 fn auto_declare_params(index: &mut BTreeMap<IndexKey, DataValue>, text: &str) {
+    let mut total = auto_derived_param_count(index);
+    if total >= MAX_AUTO_DECLARED_PARAMS {
+        return;
+    }
+    let mut new_this_line = 0;
     for (key, value, quoted) in crate::logfmt::extract_pairs(text) {
+        if new_this_line >= MAX_NEW_AUTO_PARAMS_PER_LINE || total >= MAX_AUTO_DECLARED_PARAMS {
+            break;
+        }
         let index_key = IndexKey::ParamDeclaration { name: key.to_string() };
         if index.contains_key(&index_key) {
             continue;
@@ -323,6 +356,47 @@ fn auto_declare_params(index: &mut BTreeMap<IndexKey, DataValue>, text: &str) {
                 auto_derived: true,
             }),
         );
+        new_this_line += 1;
+        total += 1;
+    }
+}
+
+/// Declares `log_type`/`process`/`pid` as standalone params the first time a `log stream`-shaped line
+/// is seen (ticket 050/054) — `Type`, the process name, and `PID` are low-cardinality, genuinely
+/// useful columns derivable directly from `log stream`'s fixed-column shape
+/// (`crate::logfmt::is_log_stream_line`), with none of the `=`-scanning ambiguity `auto_declare_params`
+/// has to guard against. Gated on that shape so ordinary logfmt/JSON/prose input never gets these three
+/// params. Unlike `auto_declare_params`, adding these is never *skipped* on account of
+/// `MAX_AUTO_DECLARED_PARAMS` — it can only ever add exactly these three names, once each — though
+/// since they're stored as `auto_derived` too, once present they do count against that shared ceiling,
+/// tightening (slightly) the budget left for generic `key=value` pairs. Like `auto_declare_params`,
+/// never overwrites an existing declaration (manual or auto) for any of the three names.
+fn auto_declare_log_stream_params(index: &mut BTreeMap<IndexKey, DataValue>, text: &str) {
+    if !crate::logfmt::is_log_stream_line(text) {
+        return;
+    }
+    let candidates: [(&str, &str, &str, MetricPrimitiveKind); 3] = [
+        ("log_type", crate::logfmt::LOG_STREAM_TYPE_PATTERN, "log_type", MetricPrimitiveKind::String),
+        ("process", crate::logfmt::LOG_STREAM_PROCESS_PATTERN, "process", MetricPrimitiveKind::String),
+        ("pid", crate::logfmt::LOG_STREAM_PID_PATTERN, "pid", MetricPrimitiveKind::Number),
+    ];
+    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(),
+                message_match_regex: pattern.to_string(),
+                capture_group: group.to_string(),
+                json_field_path: None,
+                kind,
+                elapsed_since_last: false,
+                auto_derived: true,
+            }),
+        );
     }
 }
 
@@ -1782,6 +1856,94 @@ mod tests {
         assert!(declarations[0].6, "the original auto-derived declaration must be untouched");
     }
 
+    const LOG_STREAM_FIXTURE: &str = include_str!("fixtures/log_stream_sample.txt");
+
+    /// Runs both auto-declaration passes over every fixture line, same order as `EngineMessage::Ingest`.
+    fn ingest_log_stream_fixture(index: &mut BTreeMap<IndexKey, DataValue>) {
+        for line in LOG_STREAM_FIXTURE.lines() {
+            auto_declare_log_stream_params(index, line);
+            auto_declare_params(index, line);
+        }
+    }
+
+    #[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
+        // 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
+        // `MAX_AUTO_DECLARED_PARAMS` ceiling covers all of them together (see
+        // `auto_declare_log_stream_params`'s doc comment).
+        assert!(
+            declared <= MAX_AUTO_DECLARED_PARAMS,
+            "expected at most {MAX_AUTO_DECLARED_PARAMS} declared params, got {declared}"
+        );
+        assert!(declared < 100, "declared param count ({declared}) should stay well under the old ~220");
+    }
+
+    #[test]
+    fn base64_padding_fragments_in_log_stream_fixture_are_never_auto_declared() {
+        let mut index = BTreeMap::new();
+        ingest_log_stream_fixture(&mut index);
+
+        let names: HashSet<&str> = param_declarations_range(&index).map(|d| d.0).collect();
+        // The mDNSResponder lines' `<mask.hash: '...=='>` blobs would otherwise surface as params
+        // named after the tail of the base64 payload (see `looks_like_base64_padding_artifact`).
+        assert!(!names.contains("rKTEtG7JmnThhoKPbZiVrA"));
+        assert!(!names.contains("gEdA8N4g"));
+    }
+
+    #[test]
+    fn log_stream_shaped_lines_expose_type_process_and_pid_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"));
+        assert!(declared.contains("process"));
+        assert!(declared.contains("pid"));
+
+        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("process"), Some(&MetricPrimitive::String("kitty".to_string())));
+        assert_eq!(params.get("pid"), Some(&MetricPrimitive::Number(Number::Int(1343))));
+    }
+
+    #[test]
+    fn non_log_stream_lines_do_not_get_log_type_process_or_pid_declared() {
+        let mut index = BTreeMap::new();
+        auto_declare_log_stream_params(&mut index, r#"ts=1788648105773 msg="entity_moved" x1=-4"#);
+
+        assert!(param_declarations_range(&index).next().is_none());
+    }
+
+    #[test]
+    #[ignore] // manual throughput measurement, not part of the regular suite
+    fn bench_ingest_fixture_throughput() {
+        let repeats = 50;
+        let mut index = BTreeMap::new();
+        let start = std::time::Instant::now();
+        for _ in 0..repeats {
+            for line in LOG_STREAM_FIXTURE.lines() {
+                auto_declare_log_stream_params(&mut index, line);
+                auto_declare_params(&mut index, line);
+            }
+        }
+        let elapsed = start.elapsed();
+        let total_lines = repeats * LOG_STREAM_FIXTURE.lines().count();
+        eprintln!(
+            "ingested {total_lines} lines in {elapsed:?} ({:.0} lines/sec), {} params declared",
+            total_lines as f64 / elapsed.as_secs_f64(),
+            param_declarations_range(&index).count(),
+        );
+    }
+
     #[test]
     fn rejects_invalid_regex() {
         let mut index = BTreeMap::new();