task_runs/
user_beholders.rs

1//! @arch:layer(kg_store)
2//! @arch:role(substrate)
3//! @arch:see(.yah/docs/working/yah-task-runs.md)
4//!
5//! User-extensible drop-in beholder definitions.
6//!
7//! Users place `.toml` files in `~/.yah/beholders/`. Each file describes one
8//! beholder: which command it matches, its mode (parser | rewriter), and —
9//! for parser mode — a set of regex patterns that turn output lines into
10//! structured events.
11//!
12//! ## File format
13//!
14//! ```toml
15//! name    = "my-linter"
16//! version = "1.0"
17//!
18//! # argv0 after wrapper stripping; string or array of strings
19//! argv0 = "my-linter"
20//!
21//! # Decline when any of these flags appear in argv (optional)
22//! decline_if_has = ["--version", "--help"]
23//!
24//! mode = "parser"         # "parser" | "rewriter"
25//!
26//! # Rewriter only: args to append to argv when not already present
27//! add_args = ["--json"]
28//!
29//! # Parser: one or more line-matching patterns (tried in order; all matches fire)
30//! [[patterns]]
31//! regex  = '(.+):(\d+): (error|warning): (.+)'
32//! level  = "$3"           # literal "error"/"warn"/etc., or "$N" capture group
33//! target = "my-linter"   # optional; defaults to beholder name
34//! msg    = "$4"           # optional; defaults to the full matched line
35//! [patterns.fields]
36//! "file.path" = "$1"
37//! "file.line" = "$2"
38//! ```
39//!
40//! ## Discovery
41//!
42//! [`load_user_beholders`] reads every `*.toml` file in the given directory.
43//! Files that fail to parse or are semantically invalid are skipped; a warning
44//! is written to stderr so users can debug their definitions without crashing
45//! the daemon.
46
47use std::collections::HashMap;
48use std::path::Path;
49use std::str::FromStr;
50use std::sync::Arc;
51
52use regex::Regex;
53use serde::Deserialize;
54
55use crate::beholders::{Beholder, BeholderFactory, BeholderMode};
56use crate::types::{ChunkRef, Event, EventSource, Level, OutputChunk};
57
58// ─── TOML config shapes ───────────────────────────────────────────────────────
59
60#[derive(Debug, Deserialize)]
61struct UserBeholderFile {
62    name: String,
63    version: String,
64    #[serde(default)]
65    argv0: Argv0Spec,
66    #[serde(default)]
67    decline_if_has: Vec<String>,
68    mode: ModeStr,
69    #[serde(default)]
70    add_args: Vec<String>,
71    #[serde(default)]
72    patterns: Vec<PatternDef>,
73}
74
75/// argv0 match spec: a single command name or a list of aliases.
76#[derive(Debug, Deserialize, Default)]
77#[serde(untagged)]
78enum Argv0Spec {
79    #[default]
80    None,
81    One(String),
82    Many(Vec<String>),
83}
84
85impl Argv0Spec {
86    fn is_match(&self, argv0: &str) -> bool {
87        match self {
88            Argv0Spec::None => false,
89            Argv0Spec::One(s) => s == argv0,
90            Argv0Spec::Many(v) => v.iter().any(|s| s == argv0),
91        }
92    }
93}
94
95#[derive(Debug, Deserialize)]
96#[serde(rename_all = "lowercase")]
97enum ModeStr {
98    Parser,
99    Rewriter,
100}
101
102#[derive(Debug, Deserialize)]
103struct PatternDef {
104    regex: String,
105    level: String,
106    #[serde(default)]
107    target: String,
108    /// Explicit message text. `"$N"` captures group N; literal otherwise.
109    /// Defaults to the full matched line when absent.
110    msg: Option<String>,
111    #[serde(default)]
112    fields: HashMap<String, String>,
113}
114
115// ─── Compiled representations ─────────────────────────────────────────────────
116
117struct CompiledPattern {
118    re: Regex,
119    level: FieldSpec,
120    target: String,
121    msg: Option<FieldSpec>,
122    fields: Vec<(String, FieldSpec)>,
123}
124
125/// A value that is either a literal string or a regex capture-group reference.
126enum FieldSpec {
127    Literal(String),
128    /// Index into regex captures (1-based, matching `$N` syntax).
129    Capture(usize),
130}
131
132fn parse_field_spec(s: &str) -> FieldSpec {
133    if let Some(rest) = s.strip_prefix('$') {
134        if let Ok(n) = rest.parse::<usize>() {
135            return FieldSpec::Capture(n);
136        }
137    }
138    FieldSpec::Literal(s.to_owned())
139}
140
141fn resolve_field(spec: &FieldSpec, caps: &regex::Captures) -> Option<String> {
142    match spec {
143        FieldSpec::Literal(s) => Some(s.clone()),
144        FieldSpec::Capture(n) => caps.get(*n).map(|m| m.as_str().to_owned()),
145    }
146}
147
148fn parse_level_str(s: &str) -> Level {
149    match s.to_ascii_lowercase().as_str() {
150        "error" | "err" | "fatal" => Level::Error,
151        "warn" | "warning" => Level::Warn,
152        "info" | "information" | "notice" => Level::Info,
153        "debug" | "verbose" => Level::Debug,
154        "trace" => Level::Trace,
155        _ => Level::Info,
156    }
157}
158
159impl CompiledPattern {
160    fn try_compile(def: &PatternDef, beholder_name: &str) -> Result<Self, String> {
161        let re = Regex::new(&def.regex)
162            .map_err(|e| format!("invalid regex {:?}: {e}", def.regex))?;
163        let level = parse_field_spec(&def.level);
164        let target = if def.target.is_empty() {
165            beholder_name.to_owned()
166        } else {
167            def.target.clone()
168        };
169        let msg = def.msg.as_deref().map(parse_field_spec);
170        let fields = def.fields.iter()
171            .map(|(k, v)| (k.clone(), parse_field_spec(v)))
172            .collect();
173        Ok(Self { re, level, target, msg, fields })
174    }
175
176    fn apply(&self, line: &str, chunk: &OutputChunk, source: &EventSource) -> Option<Event> {
177        let caps = self.re.captures(line)?;
178
179        let level = match &self.level {
180            FieldSpec::Literal(s) => Level::from_str(s).unwrap_or_else(|_| parse_level_str(s)),
181            FieldSpec::Capture(n) => {
182                let s = caps.get(*n)?.as_str();
183                parse_level_str(s)
184            }
185        };
186
187        let msg = match &self.msg {
188            Some(spec) => resolve_field(spec, &caps).unwrap_or_else(|| line.to_owned()),
189            None => line.to_owned(),
190        };
191
192        let mut fields_map = serde_json::Map::new();
193        for (key, spec) in &self.fields {
194            if let Some(val) = resolve_field(spec, &caps) {
195                insert_nested(&mut fields_map, key, val);
196            }
197        }
198
199        Some(Event {
200            run_id: chunk.run_id.clone(),
201            seq: 0,
202            offset_ms: chunk.offset_ms,
203            level,
204            target: self.target.clone(),
205            msg,
206            fields: serde_json::Value::Object(fields_map),
207            anchor: Some(ChunkRef { seq: chunk.seq }),
208            source: source.clone(),
209        })
210    }
211}
212
213/// Insert `val` at a dot-delimited `key` path into `map`.
214///
215/// `"file.path"` becomes `{"file": {"path": val}}`.
216/// Existing intermediate objects are merged; non-object intermediates are
217/// replaced.
218fn insert_nested(map: &mut serde_json::Map<String, serde_json::Value>, key: &str, val: String) {
219    match key.split_once('.') {
220        None => {
221            map.insert(key.to_owned(), serde_json::Value::String(val));
222        }
223        Some((head, tail)) => {
224            let inner = map
225                .entry(head.to_owned())
226                .or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));
227            if let serde_json::Value::Object(ref mut m) = inner {
228                insert_nested(m, tail, val);
229            } else {
230                // Replace non-object with a fresh object.
231                let mut m = serde_json::Map::new();
232                insert_nested(&mut m, tail, val);
233                *inner = serde_json::Value::Object(m);
234            }
235        }
236    }
237}
238
239// ─── UserBeholderFactory ──────────────────────────────────────────────────────
240
241/// Dynamic beholder factory loaded from a user TOML drop-in.
242pub struct UserBeholderFactory {
243    name: &'static str,
244    version: &'static str,
245    argv0: Argv0Spec,
246    decline_if_has: Vec<String>,
247    mode: BeholderMode,
248    patterns: Arc<Vec<CompiledPattern>>,
249}
250
251fn leak_str(s: String) -> &'static str {
252    Box::leak(s.into_boxed_str())
253}
254
255impl BeholderFactory for UserBeholderFactory {
256    fn name(&self) -> &'static str { self.name }
257    fn version(&self) -> &'static str { self.version }
258
259    fn matches(&self, resolved_argv: &[String]) -> bool {
260        let argv0 = match resolved_argv.first() {
261            Some(s) => s.as_str(),
262            None => return false,
263        };
264        if !self.argv0.is_match(argv0) {
265            return false;
266        }
267        // Decline if any of the user-listed flags appear in argv.
268        if self.decline_if_has.iter().any(|f| resolved_argv.contains(f)) {
269            return false;
270        }
271        true
272    }
273
274    fn mode(&self) -> BeholderMode { self.mode.clone() }
275
276    fn create(&self) -> Box<dyn Beholder> {
277        Box::new(UserBeholder {
278            name: self.name,
279            version: self.version,
280            patterns: Arc::clone(&self.patterns),
281            buf: Vec::new(),
282        })
283    }
284}
285
286// ─── UserBeholder ─────────────────────────────────────────────────────────────
287
288/// Per-run instance for a user-defined beholder. Buffers output and applies
289/// all compiled regex patterns to each complete line.
290struct UserBeholder {
291    name: &'static str,
292    version: &'static str,
293    patterns: Arc<Vec<CompiledPattern>>,
294    buf: Vec<u8>,
295}
296
297impl Beholder for UserBeholder {
298    fn name(&self) -> &'static str { self.name }
299    fn version(&self) -> &'static str { self.version }
300    fn mode(&self) -> BeholderMode { BeholderMode::Parser }
301
302    fn parse_chunk(&mut self, chunk: &OutputChunk) -> Vec<Event> {
303        self.buf.extend_from_slice(&chunk.bytes);
304        let mut events = Vec::new();
305        let source = EventSource::Beholder {
306            name: self.name.to_owned(),
307            version: self.version.to_owned(),
308        };
309
310        while let Some(nl) = self.buf.iter().position(|&b| b == b'\n') {
311            let raw: Vec<u8> = self.buf.drain(..=nl).collect();
312            let line = match std::str::from_utf8(&raw) {
313                Ok(s) => s.trim_end(),
314                Err(_) => continue,
315            };
316            if line.is_empty() { continue }
317
318            for pat in self.patterns.as_ref() {
319                if let Some(ev) = pat.apply(line, chunk, &source) {
320                    events.push(ev);
321                }
322            }
323        }
324
325        events
326    }
327}
328
329// ─── Loader ───────────────────────────────────────────────────────────────────
330
331/// Load all user-defined beholder factories from `*.toml` files in `dir`.
332///
333/// Missing or unreadable directories return an empty list. Files that fail
334/// to parse or compile are skipped with a warning on stderr.
335pub fn load_user_beholders(dir: &Path) -> Vec<Box<dyn BeholderFactory>> {
336    let entries = match std::fs::read_dir(dir) {
337        Ok(e) => e,
338        Err(_) => return Vec::new(),
339    };
340
341    let mut factories: Vec<Box<dyn BeholderFactory>> = Vec::new();
342
343    for entry in entries.flatten() {
344        let path = entry.path();
345        if path.extension().and_then(|e| e.to_str()) != Some("toml") {
346            continue;
347        }
348
349        let src = match std::fs::read_to_string(&path) {
350            Ok(s) => s,
351            Err(e) => {
352                eprintln!("[yah beholders] could not read {}: {e}", path.display());
353                continue;
354            }
355        };
356
357        let def: UserBeholderFile = match toml::from_str(&src) {
358            Ok(d) => d,
359            Err(e) => {
360                eprintln!("[yah beholders] parse error in {}: {e}", path.display());
361                continue;
362            }
363        };
364
365        match compile_factory(def, &path) {
366            Ok(f) => factories.push(f),
367            Err(e) => {
368                eprintln!("[yah beholders] invalid definition in {}: {e}", path.display());
369            }
370        }
371    }
372
373    factories
374}
375
376fn compile_factory(
377    def: UserBeholderFile,
378    path: &Path,
379) -> Result<Box<dyn BeholderFactory>, String> {
380    if def.name.is_empty() {
381        return Err("name must not be empty".into());
382    }
383
384    let mode = match def.mode {
385        ModeStr::Parser => BeholderMode::Parser,
386        ModeStr::Rewriter => BeholderMode::DynamicRewriter { add_args: def.add_args },
387    };
388
389    let mut compiled_patterns = Vec::new();
390    for (i, pat_def) in def.patterns.iter().enumerate() {
391        let cp = CompiledPattern::try_compile(pat_def, &def.name)
392            .map_err(|e| format!("patterns[{i}]: {e}"))?;
393        compiled_patterns.push(cp);
394    }
395
396    if matches!(def.mode, ModeStr::Parser) && compiled_patterns.is_empty() {
397        eprintln!(
398            "[yah beholders] warning: parser beholder {:?} in {} has no patterns — \
399             it will attach but emit no events",
400            def.name,
401            path.display()
402        );
403    }
404
405    Ok(Box::new(UserBeholderFactory {
406        name: leak_str(def.name),
407        version: leak_str(def.version),
408        argv0: def.argv0,
409        decline_if_has: def.decline_if_has,
410        mode,
411        patterns: Arc::new(compiled_patterns),
412    }))
413}
414
415// ─── Tests ────────────────────────────────────────────────────────────────────
416
417#[cfg(test)]
418mod tests {
419    use super::*;
420    use crate::beholders::BeholderSelect;
421    use crate::types::{Initiator, RunStatus, Stream, TaskRunId};
422
423    fn make_chunk(run_id: &TaskRunId, bytes: &[u8]) -> OutputChunk {
424        OutputChunk {
425            run_id: run_id.clone(),
426            seq: 0,
427            offset_ms: 0,
428            stream: Stream::Stdout,
429            bytes: bytes.to_vec(),
430        }
431    }
432
433    fn toml_factory(src: &str) -> Result<Box<dyn BeholderFactory>, String> {
434        let def: UserBeholderFile = toml::from_str(src)
435            .map_err(|e| e.to_string())?;
436        compile_factory(def, std::path::Path::new("<test>"))
437    }
438
439    #[test]
440    fn parser_beholder_matches_argv0() {
441        let f = toml_factory(r#"
442            name = "mytool"
443            version = "1.0"
444            argv0 = "mytool"
445            mode = "parser"
446        "#).unwrap();
447
448        assert!(f.matches(&["mytool".to_owned(), "--check".to_owned()]));
449        assert!(!f.matches(&["cargo".to_owned()]));
450        assert!(matches!(f.mode(), BeholderMode::Parser));
451    }
452
453    #[test]
454    fn parser_beholder_argv0_list() {
455        let f = toml_factory(r#"
456            name = "mytool"
457            version = "1.0"
458            argv0 = ["mytool", "mt"]
459            mode = "parser"
460        "#).unwrap();
461
462        assert!(f.matches(&["mt".to_owned()]));
463        assert!(f.matches(&["mytool".to_owned()]));
464        assert!(!f.matches(&["other".to_owned()]));
465    }
466
467    #[test]
468    fn decline_if_has_flag() {
469        let f = toml_factory(r#"
470            name = "mytool"
471            version = "1.0"
472            argv0 = "mytool"
473            decline_if_has = ["--version", "--help"]
474            mode = "parser"
475        "#).unwrap();
476
477        assert!(f.matches(&["mytool".to_owned(), "--check".to_owned()]));
478        assert!(!f.matches(&["mytool".to_owned(), "--version".to_owned()]));
479        assert!(!f.matches(&["mytool".to_owned(), "--help".to_owned()]));
480    }
481
482    #[test]
483    fn rewriter_mode_add_args() {
484        let f = toml_factory(r#"
485            name = "mytool"
486            version = "1.0"
487            argv0 = "mytool"
488            mode = "rewriter"
489            add_args = ["--json"]
490        "#).unwrap();
491
492        assert!(matches!(f.mode(), BeholderMode::DynamicRewriter { .. }));
493        if let BeholderMode::DynamicRewriter { add_args } = f.mode() {
494            assert_eq!(add_args, vec!["--json"]);
495        }
496    }
497
498    #[test]
499    fn parser_extracts_events_from_chunk() {
500        let f = toml_factory(r#"
501            name = "mytool"
502            version = "1.0"
503            argv0 = "mytool"
504            mode = "parser"
505
506            [[patterns]]
507            regex = '^(.+):(\d+): (error|warning): (.+)$'
508            level = "$3"
509            msg = "$4"
510            [patterns.fields]
511            "file.path" = "$1"
512            "file.line" = "$2"
513        "#).unwrap();
514
515        let run_id = TaskRunId::new();
516        let mut beholder = f.create();
517        let line = b"src/main.rs:42: error: type mismatch\n";
518        let chunk = make_chunk(&run_id, line);
519        let events = beholder.parse_chunk(&chunk);
520
521        assert_eq!(events.len(), 1);
522        let ev = &events[0];
523        assert!(matches!(ev.level, Level::Error));
524        assert_eq!(ev.msg, "type mismatch");
525        assert_eq!(ev.fields["file"]["path"], "src/main.rs");
526        assert_eq!(ev.fields["file"]["line"], "42");
527    }
528
529    #[test]
530    fn parser_defaults_msg_to_full_line() {
531        let f = toml_factory(r#"
532            name = "mytool"
533            version = "1.0"
534            argv0 = "mytool"
535            mode = "parser"
536
537            [[patterns]]
538            regex = 'ERROR'
539            level = "error"
540        "#).unwrap();
541
542        let run_id = TaskRunId::new();
543        let mut beholder = f.create();
544        let chunk = make_chunk(&run_id, b"ERROR: something went wrong\n");
545        let events = beholder.parse_chunk(&chunk);
546
547        assert_eq!(events.len(), 1);
548        assert_eq!(events[0].msg, "ERROR: something went wrong");
549    }
550
551    #[test]
552    fn parser_all_patterns_fire_on_same_line() {
553        let f = toml_factory(r#"
554            name = "mytool"
555            version = "1.0"
556            argv0 = "mytool"
557            mode = "parser"
558
559            [[patterns]]
560            regex = 'error'
561            level = "error"
562
563            [[patterns]]
564            regex = 'warning'
565            level = "warn"
566        "#).unwrap();
567
568        let run_id = TaskRunId::new();
569        let mut beholder = f.create();
570        // Line matches neither — one event only from the first pattern? No, "error warning" matches both.
571        let chunk = make_chunk(&run_id, b"error warning foo\n");
572        let events = beholder.parse_chunk(&chunk);
573        assert_eq!(events.len(), 2);
574    }
575
576    #[test]
577    fn insert_nested_dot_path() {
578        let mut map = serde_json::Map::new();
579        insert_nested(&mut map, "file.path", "src/main.rs".to_owned());
580        insert_nested(&mut map, "file.line", "42".to_owned());
581        insert_nested(&mut map, "error.code", "E001".to_owned());
582        insert_nested(&mut map, "top", "value".to_owned());
583
584        assert_eq!(map["file"]["path"], "src/main.rs");
585        assert_eq!(map["file"]["line"], "42");
586        assert_eq!(map["error"]["code"], "E001");
587        assert_eq!(map["top"], "value");
588    }
589
590    #[test]
591    fn load_from_dir_skips_non_toml() {
592        let dir = tempfile::tempdir().unwrap();
593        std::fs::write(dir.path().join("not-toml.txt"), "irrelevant").unwrap();
594        std::fs::write(dir.path().join("beholder.toml"), r#"
595            name = "loaded"
596            version = "1.0"
597            argv0 = "loaded"
598            mode = "parser"
599        "#).unwrap();
600
601        let factories = load_user_beholders(dir.path());
602        assert_eq!(factories.len(), 1);
603        assert_eq!(factories[0].name(), "loaded");
604    }
605
606    #[test]
607    fn load_from_missing_dir_returns_empty() {
608        let factories = load_user_beholders(Path::new("/nonexistent/path/to/beholders"));
609        assert!(factories.is_empty());
610    }
611
612    #[test]
613    fn load_skips_malformed_toml() {
614        let dir = tempfile::tempdir().unwrap();
615        std::fs::write(dir.path().join("bad.toml"), "not valid toml [[[").unwrap();
616        std::fs::write(dir.path().join("good.toml"), r#"
617            name = "good"
618            version = "1.0"
619            argv0 = "good"
620            mode = "parser"
621        "#).unwrap();
622
623        let factories = load_user_beholders(dir.path());
624        assert_eq!(factories.len(), 1);
625        assert_eq!(factories[0].name(), "good");
626    }
627}
task_runs/user_beholders.rs

task_runs/
user_beholders.rs
