yarp_cli/reducers/
mod.rs

1mod bounded;
2mod build;
3mod classify;
4mod diff;
5mod evidence;
6pub(crate) mod filter;
7mod list;
8mod log;
9mod search;
10mod status;
11mod test;
12
13use bounded::{LineAccumulator, LineView, ShortRaw};
14pub use evidence::RecoveryMarker;
15use evidence::{EvidenceClass, EvidenceCollector};
16use filter::{AnsiStripper, line_filter_keeps};
17
18use crate::rewrite::TransformDiagnostics;
19use crate::rule_pack::{self as yarp_rule_pack, Action, OutputPolicy, Reducer, Rule};
20
21const DIAGNOSTIC_CONTEXT_LINES: usize = 6;
22
23#[derive(Debug, Default)]
24struct RegisteredDiagnosticTracker {
25    matched: [bool; 5],
26    window: u128,
27}
28
29impl RegisteredDiagnosticTracker {
30    fn push(&mut self, byte: u8) {
31        let lower = byte.to_ascii_lowercase();
32        self.window = (self.window << 8) | u128::from(lower);
33        match lower {
34            b'e' if tail_matches(self.window, FAILURE) => self.matched[0] = true,
35            b'c' if tail_matches(self.window, PANIC) => self.matched[1] = true,
36            b'r' if tail_matches(self.window, ERROR) => self.matched[2] = true,
37            b'g' if tail_matches(self.window, WARNING) => self.matched[3] = true,
38            b't' if tail_matches(self.window, TEST_RESULT) => self.matched[4] = true,
39            _ => {}
40        }
41    }
42
43    fn observe_slice(&mut self, body: &[u8]) {
44        for (matched, term) in self
45            .matched
46            .iter_mut()
47            .zip(classify::REGISTERED_DIAGNOSTICS)
48        {
49            if !*matched && classify::contains_ascii_insensitive(body, term) {
50                *matched = true;
51            }
52        }
53    }
54
55    fn end_line(&mut self) {
56        self.window = 0;
57    }
58}
59
60const FAILURE: (u128, usize) = (0x66_61_69_6c_75_72_65, 56);
61const PANIC: (u128, usize) = (0x70_61_6e_69_63, 40);
62const ERROR: (u128, usize) = (0x65_72_72_6f_72, 40);
63const WARNING: (u128, usize) = (0x77_61_72_6e_69_6e_67, 56);
64const TEST_RESULT: (u128, usize) = (0x74_65_73_74_20_72_65_73_75_6c_74, 88);
65
66fn tail_matches(window: u128, term: (u128, usize)) -> bool {
67    let mask = (1_u128 << term.1) - 1;
68    window & mask == term.0
69}
70
71#[derive(Debug)]
72pub struct StreamReducer {
73    kind: Reducer,
74    success_policy: OutputPolicy,
75    failure_policy: OutputPolicy,
76    success: EvidenceCollector,
77    failure: EvidenceCollector,
78    raw: ShortRaw,
79    line: LineAccumulator,
80    ansi: Option<AnsiStripper>,
81    line_number: u64,
82    diagnostic_context: usize,
83    registered_diagnostics: RegisteredDiagnosticTracker,
84    transform_diagnostics: TransformDiagnostics,
85}
86
87impl StreamReducer {
88    /// Build bounded state for one validated reduction rule.
89    ///
90    /// # Errors
91    ///
92    /// Returns an error when the supplied rule is not a complete reduction rule.
93    pub fn new(rule: &Rule) -> Result<Self, String> {
94        Self::new_with_transform_diagnostics(rule, TransformDiagnostics::default())
95    }
96
97    pub(crate) fn new_with_transform_diagnostics(
98        rule: &Rule,
99        transform_diagnostics: TransformDiagnostics,
100    ) -> Result<Self, String> {
101        if rule.action != Action::Reduce {
102            return Err("cannot create a reducer for a passthrough rule".to_owned());
103        }
104        let memory_bound = yarp_rule_pack::stream_memory_bound(rule)?;
105        if memory_bound > yarp_rule_pack::MAX_STREAM_MEMORY_BYTES {
106            return Err(format!(
107                "rule requires {memory_bound} bytes per stream, above the {}-byte limit",
108                yarp_rule_pack::MAX_STREAM_MEMORY_BYTES
109            ));
110        }
111        let kind = rule
112            .reducer
113            .clone()
114            .ok_or_else(|| "reduction rule is missing a reducer".to_owned())?;
115        let success_policy = rule
116            .success
117            .ok_or_else(|| "reduction rule is missing a success policy".to_owned())?;
118        let failure_policy = rule
119            .failure
120            .ok_or_else(|| "reduction rule is missing a failure policy".to_owned())?;
121        let max_line = success_policy
122            .max_line_bytes
123            .max(failure_policy.max_line_bytes);
124        let strip_ansi = !matches!(
125            kind,
126            Reducer::LineFilter {
127                strip_ansi: false,
128                ..
129            }
130        );
131        let family = family_name(&kind);
132        Ok(Self {
133            kind,
134            success_policy,
135            failure_policy,
136            success: EvidenceCollector::new(family, success_policy),
137            failure: EvidenceCollector::new(family, failure_policy),
138            raw: ShortRaw::new(success_policy, failure_policy),
139            line: LineAccumulator::new(max_line, max_line),
140            ansi: strip_ansi.then(AnsiStripper::new),
141            line_number: 0,
142            diagnostic_context: 0,
143            registered_diagnostics: RegisteredDiagnosticTracker::default(),
144            transform_diagnostics,
145        })
146    }
147
148    pub fn push(&mut self, chunk: &[u8]) {
149        self.raw.push(chunk);
150        for byte in chunk {
151            if self.line.observe_source(*byte) {
152                self.registered_diagnostics.push(*byte);
153            }
154            let output = self
155                .ansi
156                .as_mut()
157                .map_or(Some(*byte), |stripper| stripper.push_byte(*byte));
158            if let Some(output) = output {
159                self.line.push_output(output);
160            }
161            if *byte == b'\n' {
162                let line = self.line.take();
163                self.process_line(&line);
164            }
165        }
166    }
167
168    #[must_use]
169    pub fn finish(mut self, success: bool, recovery: Option<RecoveryMarker<'_>>) -> Vec<u8> {
170        if !self.line.is_empty() {
171            let line = self.line.take();
172            self.process_line(&line);
173        }
174        let (collector, policy) = if success {
175            (self.success, self.success_policy)
176        } else {
177            (self.failure, self.failure_policy)
178        };
179        self.raw.choose(
180            collector.render(recovery, self.registered_diagnostics.matched, success),
181            policy,
182        )
183    }
184
185    fn process_line(&mut self, line: &LineView) {
186        self.line_number = self.line_number.saturating_add(1);
187        if classify::has_registered_diagnostic(&line.prefix) {
188            self.registered_diagnostics.observe_slice(&line.prefix);
189        }
190        if line.truncated {
191            let tail = line.tail_bytes();
192            if classify::has_registered_diagnostic(&tail) {
193                self.registered_diagnostics.observe_slice(&tail);
194            }
195        }
196        let mut class = if classify::is_tool_outcome(&line.prefix) {
197            EvidenceClass::Outcome
198        } else if self.transform_diagnostics.matches_line(&line.prefix) {
199            EvidenceClass::Diagnostic
200        } else {
201            classify_line(&self.kind, line)
202        };
203        if class == EvidenceClass::Diagnostic {
204            self.diagnostic_context = DIAGNOSTIC_CONTEXT_LINES;
205        } else if self.diagnostic_context > 0 {
206            if class == EvidenceClass::Noise || class == EvidenceClass::Example {
207                class = EvidenceClass::Context;
208            }
209            self.diagnostic_context -= 1;
210        }
211        self.registered_diagnostics.end_line();
212        self.success.observe(self.line_number, line, class);
213        self.failure.observe(self.line_number, line, class);
214    }
215}
216
217fn family_name(kind: &Reducer) -> &'static str {
218    match kind {
219        Reducer::SearchSummary => "search",
220        Reducer::DiffSummary => "diff",
221        Reducer::TestSummary => "test",
222        Reducer::BuildSummary => "build",
223        Reducer::LogSummary => "log",
224        Reducer::StatusSummary => "status",
225        Reducer::ListSummary => "list",
226        Reducer::LineFilter { .. } => "line filter",
227    }
228}
229
230fn classify_line(kind: &Reducer, line: &LineView) -> EvidenceClass {
231    match kind {
232        Reducer::SearchSummary => search::classify(&line.prefix),
233        Reducer::DiffSummary => diff::classify(&line.prefix),
234        Reducer::TestSummary => test::classify(&line.prefix),
235        Reducer::BuildSummary => build::classify(&line.prefix),
236        Reducer::LogSummary => log::classify(&line.prefix),
237        Reducer::StatusSummary => status::classify(&line.prefix),
238        Reducer::ListSummary => list::classify(&line.prefix),
239        Reducer::LineFilter { drop, keep, .. } => {
240            if line_filter_keeps(line, drop, keep) {
241                EvidenceClass::Example
242            } else {
243                EvidenceClass::Noise
244            }
245        }
246    }
247}
248
249/// Reduce one in-memory byte stream with the same engine used by child capture.
250///
251/// # Errors
252///
253/// Returns an error when the supplied rule is not a complete reduction rule.
254pub fn reduce_bytes(rule: &Rule, input: &[u8], success: bool) -> Result<Vec<u8>, String> {
255    reduce_bytes_with_recovery(rule, input, success, None)
256}
257
258/// Reduce one in-memory stream and include a committed recovery reference in the savings decision.
259///
260/// # Errors
261///
262/// Returns an error when the supplied rule is not a complete reduction rule.
263pub fn reduce_bytes_with_recovery(
264    rule: &Rule,
265    input: &[u8],
266    success: bool,
267    recovery: Option<RecoveryMarker<'_>>,
268) -> Result<Vec<u8>, String> {
269    reduce_bytes_with_recovery_and_transform_diagnostics(
270        rule,
271        input,
272        success,
273        recovery,
274        TransformDiagnostics::default(),
275    )
276}
277
278/// Reduce one in-memory stream while treating selected transform prefixes as typed diagnostics.
279///
280/// # Errors
281///
282/// Returns an error when the supplied rule is not a complete reduction rule.
283pub fn reduce_bytes_with_recovery_and_transform_diagnostics(
284    rule: &Rule,
285    input: &[u8],
286    success: bool,
287    recovery: Option<RecoveryMarker<'_>>,
288    transform_diagnostics: TransformDiagnostics,
289) -> Result<Vec<u8>, String> {
290    let mut reducer = StreamReducer::new_with_transform_diagnostics(rule, transform_diagnostics)?;
291    for chunk in input.chunks(8 * 1024) {
292        reducer.push(chunk);
293    }
294    Ok(reducer.finish(success, recovery))
295}
296
297/// Calculate a conservative retained-memory bound for one stream.
298///
299/// # Errors
300///
301/// Returns an error when the rule is incomplete or the bound overflows `usize`.
302pub fn configured_memory_bound(rule: &Rule) -> Result<usize, String> {
303    yarp_rule_pack::stream_memory_bound(rule)
304}
305
306#[cfg(test)]
307mod tests {
308    use std::fmt::Write as _;
309
310    use crate::rule_pack::{CommandMatcher, OutputPolicy};
311
312    use super::*;
313
314    fn policy(max: usize, savings: usize) -> OutputPolicy {
315        OutputPolicy {
316            max_line_bytes: 256,
317            max_output_bytes: max,
318            min_savings_bytes: savings,
319            min_savings_basis_points: 1_000,
320        }
321    }
322
323    fn rule(reducer: Reducer) -> Rule {
324        Rule {
325            id: "test/rule".to_owned(),
326            matcher: CommandMatcher {
327                program: vec!["test".to_owned()],
328                argv_prefix: Vec::new(),
329                argv_contains_all: Vec::new(),
330            },
331            action: Action::Reduce,
332            transform: None,
333            reducer: Some(reducer),
334            success: Some(policy(1_024, 8)),
335            failure: Some(policy(2_048, 8)),
336        }
337    }
338
339    #[test]
340    fn uses_real_exit_status_policy() {
341        let input = numbered_lines("line", 200);
342        let success = reduce_bytes(&rule(Reducer::ListSummary), input.as_bytes(), true)
343            .expect("success reduction");
344        let failure = reduce_bytes(&rule(Reducer::ListSummary), input.as_bytes(), false)
345            .expect("failure reduction");
346        assert!(
347            failure.len() > success.len(),
348            "success={} failure={}",
349            success.len(),
350            failure.len()
351        );
352    }
353
354    #[test]
355    fn keeps_tool_session_identifiers_needed_for_follow_up() {
356        let input = format!(
357            "Chunk ID: abc\nProcess running with session ID 12345\n{}",
358            numbered_lines("path", 2_000)
359        );
360        let output =
361            reduce_bytes(&rule(Reducer::ListSummary), input.as_bytes(), true).expect("reduction");
362        let text = String::from_utf8(output).expect("UTF-8");
363        assert!(text.contains("Process running with session ID 12345"));
364    }
365
366    #[test]
367    fn rejects_rules_above_the_stream_memory_limit() {
368        let mut oversized = rule(Reducer::ListSummary);
369        let policy = OutputPolicy {
370            max_line_bytes: 1_048_576,
371            max_output_bytes: 4_194_304,
372            min_savings_bytes: 1_048_576,
373            min_savings_basis_points: 1_000,
374        };
375        oversized.success = Some(policy);
376        oversized.failure = Some(policy);
377        let error = StreamReducer::new(&oversized).expect_err("memory limit");
378        assert!(error.contains("per stream"));
379    }
380
381    #[test]
382    fn output_is_independent_of_chunk_boundaries() {
383        let input = numbered_lines("path:1:match", 400);
384        let rule = rule(Reducer::SearchSummary);
385        let expected = reduce_bytes(&rule, input.as_bytes(), true).expect("reduction");
386        for chunk_size in 1..32 {
387            let mut reducer = StreamReducer::new(&rule).expect("reducer");
388            for chunk in input.as_bytes().chunks(chunk_size) {
389                reducer.push(chunk);
390            }
391            assert_eq!(
392                reducer.finish(true, None),
393                expected,
394                "chunk size {chunk_size}"
395            );
396        }
397    }
398
399    #[test]
400    fn keeps_diagnostics_from_selected_transform_stages() {
401        let plan = crate::rewrite::select_result_plan("cargo test | tee -p /dev/full")
402            .expect("tee pipeline");
403        let input = format!(
404            "{}tee: /dev/full: No space left on device\n",
405            "test routine ... ok\n".repeat(1_000)
406        );
407        let output = reduce_bytes_with_recovery_and_transform_diagnostics(
408            &plan.rule,
409            input.as_bytes(),
410            false,
411            None,
412            plan.transform_diagnostics,
413        )
414        .expect("reduction");
415        assert!(String::from_utf8_lossy(&output).contains("tee: /dev/full"));
416    }
417
418    #[test]
419    fn test_summary_keeps_failures_and_drops_progress() {
420        let input = format!(
421            "{}error: build failed\ntest result: FAILED\n",
422            "test routine ... ok\n".repeat(1_000)
423        );
424        let output =
425            reduce_bytes(&rule(Reducer::TestSummary), input.as_bytes(), false).expect("reduction");
426        let text = String::from_utf8(output).expect("UTF-8");
427        assert!(!text.contains("test routine"));
428        assert!(text.contains("error: build failed"));
429        assert!(text.contains("test result: FAILED"));
430    }
431
432    #[test]
433    fn keeps_registered_diagnostics_from_the_tail_of_a_bounded_long_line() {
434        let mut compact = rule(Reducer::SearchSummary);
435        compact.success.as_mut().expect("policy").max_line_bytes = 256;
436        compact.failure.as_mut().expect("policy").max_line_bytes = 256;
437        let input = format!("{} warning: tail evidence\n", "x".repeat(1_024));
438        let output = reduce_bytes(&compact, input.as_bytes(), true).expect("reduction");
439        assert!(String::from_utf8_lossy(&output).contains("warning"));
440    }
441
442    #[test]
443    fn records_registered_diagnostics_from_the_middle_of_an_oversized_line() {
444        let mut compact = rule(Reducer::SearchSummary);
445        compact.success.as_mut().expect("policy").max_line_bytes = 256;
446        compact.failure.as_mut().expect("policy").max_line_bytes = 256;
447        let input = format!(
448            "{} warning: middle evidence {}\n",
449            "x".repeat(1_024),
450            "y".repeat(1_024)
451        );
452        let output = reduce_bytes(&compact, input.as_bytes(), true).expect("reduction");
453        assert!(String::from_utf8_lossy(&output).contains("source_terms=warning"));
454    }
455
456    #[test]
457    fn applies_the_line_limit_before_stripping_ansi() {
458        let mut input = b"\x1b]".to_vec();
459        input.extend(std::iter::repeat_n(b'x', 400));
460        input.extend_from_slice(b"\x07visible\n");
461        input.extend_from_slice(numbered_lines("tail", 200).as_bytes());
462        let rule = rule(Reducer::LineFilter {
463            strip_ansi: true,
464            drop: Vec::new(),
465            keep: Vec::new(),
466        });
467        let output = reduce_bytes(&rule, &input, true).expect("reduction");
468        assert!(
469            String::from_utf8_lossy(&output).contains("line truncated"),
470            "raw source length must control line truncation"
471        );
472    }
473
474    #[test]
475    fn every_builtin_reducer_obeys_both_hard_output_budgets() {
476        let input = numbered_lines("diagnostic line", 4_000);
477        let mut registry = crate::rules::Registry::builtins_only();
478        for summary in registry.summaries().expect("built-in summaries") {
479            if summary.rule.action != Action::Reduce {
480                continue;
481            }
482            assert!(
483                configured_memory_bound(&summary.rule).expect("memory bound") < 4 * 1024 * 1024,
484                "{} exceeds the per-stream memory target",
485                summary.rule.id
486            );
487            for success in [true, false] {
488                let output = reduce_bytes(&summary.rule, input.as_bytes(), success)
489                    .unwrap_or_else(|error| panic!("{}: {error}", summary.rule.id));
490                let policy = if success {
491                    summary.rule.success.expect("success policy")
492                } else {
493                    summary.rule.failure.expect("failure policy")
494                };
495                assert!(
496                    output.len() <= policy.max_output_bytes,
497                    "{} emitted {} bytes with a {} byte budget",
498                    summary.rule.id,
499                    output.len(),
500                    policy.max_output_bytes
501                );
502            }
503        }
504    }
505
506    #[test]
507    fn every_builtin_reducer_keeps_short_output_exact() {
508        let input = b"one useful line\n";
509        let mut registry = crate::rules::Registry::builtins_only();
510        for summary in registry.summaries().expect("built-in summaries") {
511            if summary.rule.action != Action::Reduce {
512                continue;
513            }
514            let output = reduce_bytes(&summary.rule, input, true)
515                .unwrap_or_else(|error| panic!("{}: {error}", summary.rule.id));
516            assert_eq!(output, input, "{}", summary.rule.id);
517        }
518    }
519
520    fn numbered_lines(prefix: &str, count: usize) -> String {
521        let mut output = String::new();
522        for index in 0..count {
523            writeln!(&mut output, "{prefix} {index}: repeated output")
524                .expect("write synthetic output");
525        }
526        output
527    }
528
529    #[test]
530    fn unknown_binary_bytes_are_processed_without_utf8() {
531        let mut input = vec![0xff; 4_096];
532        input.push(b'\n');
533        let output = reduce_bytes(&rule(Reducer::SearchSummary), &input, true).expect("reduction");
534        assert!(output.len() <= 1_024);
535    }
536}
yarp_cli/reducers/mod.rs

yarp_cli/reducers/
mod.rs
