perf-sentinel-core 0.11.0

//! Pipeline: wires all stages together.

use crate::config::Config;
use crate::correlate;
use crate::detect;
use crate::detect::{Confidence, DetectConfig};
use crate::event::SpanEvent;
use crate::ingest::otlp::SpanConversionStats;
use crate::normalize;
use crate::report::{Analysis, Report};
use crate::score;

/// Run the full analysis pipeline on a batch of events.
#[must_use]
pub fn analyze(events: Vec<SpanEvent>, config: &Config) -> Report {
    analyze_with_traces(events, config, None).0
}

/// Run the full analysis pipeline, returning both the report and the correlated traces.
///
/// Use this when you need the intermediate `Trace` structures (e.g., for tree building
/// in the TUI inspect mode) without re-running normalization and correlation.
///
/// `ingest_stats` is the OTLP span-filter tally from
/// [`crate::ingest::json::JsonIngest::ingest_with_stats`]. When `Some` it
/// lands in `analysis.ingest` and feeds the opt-in `min_usable_span_ratio`
/// gate rule; pass `None` when the input carried no tally.
#[must_use]
pub fn analyze_with_traces(
    events: Vec<SpanEvent>,
    config: &Config,
    ingest_stats: Option<SpanConversionStats>,
) -> (Report, Vec<correlate::Trace>) {
    let start = std::time::Instant::now();
    let event_count = events.len();

    let normalized = normalize::normalize_all(events);
    let traces = correlate::correlate(normalized);
    let trace_count = traces.len();

    let detect_config = DetectConfig::from(config);
    let findings = detect::run_full_detection(&traces, &detect_config);

    let (mut findings, green_summary, per_endpoint_io_ops) = if config.green.enabled {
        let carbon_ctx = config.carbon_context();
        score::score_green(&traces, findings, Some(&carbon_ctx))
    } else {
        let total_io_ops = traces.iter().map(|t| t.spans.len()).sum();
        // Green disabled: skip the full scoring pass but still walk
        // the spans once for the per-endpoint counter. `score_green`
        // returns the same data as part of its own iteration when
        // enabled, so we never iterate twice in either branch.
        let per_endpoint_io_ops = crate::report::compute_per_endpoint_io_ops(&traces);
        (
            findings,
            crate::report::GreenSummary::disabled(total_io_ops),
            per_endpoint_io_ops,
        )
    };

    // Sort findings for deterministic output (HashMap iteration order is random)
    detect::sort_findings(&mut findings);

    // Stamp confidence on every finding. `analyze` is the batch path: it
    // stamps CiBatch when a CI environment is detected, otherwise LocalBatch
    // (a developer-machine run). The real daemon path
    // (daemon::process_traces) stamps Staging or Production from
    // Config::confidence() instead. Detectors themselves never reason about
    // confidence; they emit Confidence::default() and the pipeline overrides
    // it here. Env detection is the one impure step, kept to a helper.
    detect::apply_confidence(
        &mut findings,
        Confidence::batch_for_ci(ci_environment_detected()),
    );

    // Stamp the canonical signature so JSON consumers can copy-paste it
    // into `.perf-sentinel-acknowledgments.toml` without having to recompute.
    crate::acknowledgments::enrich_with_signatures(&mut findings);

    let ingest = ingest_stats.map(crate::report::IngestStats::from);
    let quality_gate = crate::quality_gate::evaluate(
        &findings,
        &green_summary,
        &config.thresholds,
        ingest.as_ref(),
    );
    let warning_details = skipped_usable_span_rule_warning(&config.thresholds, ingest.as_ref())
        .into_iter()
        .chain(daemon_only_green_backend_warning(&config.green))
        .collect();

    let report = Report {
        analysis: Analysis {
            duration_ms: start.elapsed().as_millis() as u64,
            events_processed: event_count,
            traces_analyzed: trace_count,
            ingest,
        },
        findings,
        green_summary,
        quality_gate,
        per_endpoint_io_ops,
        // Batch mode does not run the cross-trace correlator, whose
        // rolling window only exists in the daemon. Always empty here.
        correlations: vec![],
        embedded_traces: vec![],
        warnings: vec![],
        warning_details,
        acknowledged_findings: vec![],
        binary_version: env!("CARGO_PKG_VERSION").to_string(),
        // Batch analyze does not feed the periodic disclosure (that path is
        // daemon-archive-sourced), so the canonical avoidable tiers are only
        // computed at archive time, not here.
        detection_config: Some(detect_config),
        disclosure_waste: None,
    };

    (report, traces)
}

/// A configured `min_usable_span_ratio` that could not be evaluated is
/// worth saying out loud: the rule exists to stop a false green, so
/// silently skipping it reproduces exactly what it guards against. The
/// sample floor and a non-OTLP input are the two ways that happens.
fn skipped_usable_span_rule_warning(
    thresholds: &crate::config::ThresholdsConfig,
    ingest: Option<&crate::report::IngestStats>,
) -> Option<crate::report::Warning> {
    let threshold = thresholds.min_usable_span_ratio?;
    if ingest.is_some_and(|i| i.usable_span_ratio.is_some()) {
        return None;
    }
    let cause = match ingest {
        Some(i) => format!(
            "no I/O kind reached the {} span sample floor (received {}, filtered {})",
            crate::ingest::otlp::MIN_RATIO_SAMPLE,
            i.spans_received,
            i.spans_filtered
        ),
        None => {
            "the input carries no OTLP span tally (native, Jaeger, Zipkin or Tempo)".to_string()
        }
    };
    Some(crate::report::Warning::new(
        crate::report::warnings::TUNING,
        format!(
            "`min_usable_span_ratio` = {threshold} was not evaluated: {cause}. The quality gate says nothing about instrumentation quality for this run."
        ),
    ))
}

/// A green backend configured in the TOML that a batch run can never reach:
/// `analyze` starts no scraper, so the section is read and ignored. Only
/// compile-time table names are interpolated, so `Warning::new` is enough.
fn daemon_only_green_backend_warning(
    green: &crate::config::GreenConfig,
) -> Option<crate::report::Warning> {
    if !green.enabled {
        return None;
    }
    let configured: Vec<&str> = [
        ("`[green.alumet]`", green.alumet.is_some()),
        ("`[green.scaphandre]`", green.scaphandre.is_some()),
        ("`[green.kepler]`", green.kepler.is_some()),
        ("`[green.redfish]`", green.redfish.is_some()),
        ("`[green.cloud]`", green.cloud_energy.is_some()),
        ("`[green.broker_static]`", green.broker_static.is_some()),
        (
            "`[green.electricity_maps]`",
            green.electricity_maps.is_some(),
        ),
    ]
    .into_iter()
    .filter_map(|(name, set)| set.then_some(name))
    .collect();
    if configured.is_empty() {
        return None;
    }
    Some(crate::report::Warning::new(
        crate::report::warnings::TUNING,
        format!(
            "{} configured but this is a batch run: `analyze` starts no scraper, \
             so neither measured energy nor real-time grid intensity reached the \
             score. Every carbon figure here is the I/O proxy estimate over \
             embedded intensity data. Run `perf-sentinel watch` for measured figures.",
            configured.join(", ")
        ),
    ))
}

/// Detect a CI environment. GitHub Actions, GitLab, Travis and most runners
/// export a truthy `CI` env var. `CI=false` and `CI=0` count as "not CI" so
/// an operator can force a local context. Jenkins does not set `CI` (it
/// exports `JENKINS_URL`), so a present `JENKINS_URL` also counts as CI
/// unless `CI` is explicitly false. The single source of truth: the batch
/// [`Confidence`] here and the disclosure `generated_by` attribution both
/// call it.
#[must_use]
pub fn ci_environment_detected() -> bool {
    match std::env::var("CI") {
        Ok(v) if v == "false" || v == "0" => false,
        Ok(v) if !v.is_empty() => true,
        _ => std::env::var_os("JENKINS_URL").is_some(),
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use super::*;
    use crate::event::SpanEvent;

    #[test]
    fn empty_pipeline_produces_empty_report() {
        let config = Config::default();
        let report = analyze(vec![], &config);
        assert!(report.findings.is_empty());
        assert_eq!(report.analysis.events_processed, 0);
        assert_eq!(report.analysis.traces_analyzed, 0);
        assert!(report.quality_gate.passed);
    }

    #[test]
    fn waste_dedup_no_double_count() {
        use crate::test_helpers::{make_sql_event, make_sql_series_events};
        // 5 different params + 2 duplicates of param 1 = 7 events, same template
        // N+1 sees 7 occurrences with 5 distinct params -> finding (avoidable = 6)
        // Redundant sees 3 occurrences of order_id=1 -> finding (avoidable = 2)
        // Without dedup: 6 + 2 = 8. With dedup: max(6, 2) = 6.
        let mut events: Vec<SpanEvent> = make_sql_series_events(5);
        // Add 2 more with order_id = 1 (duplicates)
        for i in 6..=7 {
            events.push(make_sql_event(
                "trace-1",
                &format!("span-{i}"),
                "SELECT * FROM order_item WHERE order_id = 1",
                &format!("2025-07-10T14:32:01.{:03}Z", i * 40),
            ));
        }

        let config = Config::default();
        let report = analyze(events, &config);
        assert!(!report.findings.is_empty());
        assert_eq!(report.green_summary.avoidable_io_ops, 6);
    }

    #[test]
    fn zero_events_waste_ratio_is_zero() {
        let config = Config::default();
        let report = analyze(vec![], &config);
        assert!((report.green_summary.io_waste_ratio - 0.0).abs() < f64::EPSILON);
        assert_eq!(report.green_summary.total_io_ops, 0);
        assert_eq!(report.green_summary.avoidable_io_ops, 0);
    }

    #[test]
    fn clean_events_zero_waste_ratio() {
        use crate::test_helpers::make_sql_event;
        // 4 events with different templates -> no N+1 (below threshold), no redundant
        let events = vec![
            make_sql_event(
                "trace-1",
                "span-1",
                "SELECT * FROM users WHERE id = 1",
                "2025-07-10T14:32:01.000Z",
            ),
            make_sql_event(
                "trace-1",
                "span-2",
                "SELECT * FROM orders WHERE id = 2",
                "2025-07-10T14:32:01.050Z",
            ),
            make_sql_event(
                "trace-1",
                "span-3",
                "SELECT * FROM products WHERE id = 3",
                "2025-07-10T14:32:01.100Z",
            ),
            make_sql_event(
                "trace-1",
                "span-4",
                "INSERT INTO logs (msg) VALUES ('ok')",
                "2025-07-10T14:32:01.150Z",
            ),
        ];

        let config = Config::default();
        let report = analyze(events, &config);

        assert!(report.findings.is_empty());
        assert_eq!(report.green_summary.total_io_ops, 4);
        assert_eq!(report.green_summary.avoidable_io_ops, 0);
        assert!((report.green_summary.io_waste_ratio - 0.0).abs() < f64::EPSILON);
    }

    #[test]
    fn pipeline_with_findings_computes_green_summary() {
        use crate::test_helpers::make_n_plus_one_events;
        // 6 events with different params -> N+1 finding
        let events = make_n_plus_one_events();

        let config = Config::default();
        let report = analyze(events, &config);

        assert!(!report.findings.is_empty());
        assert_eq!(report.green_summary.avoidable_io_ops, 5);
        assert!((report.green_summary.io_waste_ratio - 5.0_f64 / 6.0).abs() < f64::EPSILON);
        assert_eq!(report.green_summary.total_io_ops, 6);
    }

    #[test]
    fn dedup_across_traces() {
        use crate::test_helpers::make_sql_event;
        // Two traces, each with redundant queries on different templates
        let mut events = Vec::new();
        for i in 1..=3 {
            events.push(make_sql_event(
                "trace-A",
                &format!("span-a{i}"),
                "SELECT * FROM order_item WHERE order_id = 42",
                &format!("2025-07-10T14:32:01.{:03}Z", i * 50),
            ));
        }
        for i in 1..=3 {
            events.push(make_sql_event(
                "trace-B",
                &format!("span-b{i}"),
                "SELECT * FROM orders WHERE user_id = 7",
                &format!("2025-07-10T14:32:02.{:03}Z", i * 50),
            ));
        }

        let config = Config::default();
        let report = analyze(events, &config);

        // Each trace has 3 redundant -> avoidable = 2 each -> total = 4
        assert_eq!(report.green_summary.avoidable_io_ops, 4);
        assert_eq!(report.green_summary.total_io_ops, 6);
    }

    #[test]
    fn pipeline_with_green_default_region_produces_co2() {
        use crate::test_helpers::make_n_plus_one_events;
        let events = make_n_plus_one_events();

        let config = Config {
            green: crate::config::GreenConfig {
                default_region: Some("eu-west-3".to_string()),
                ..crate::config::GreenConfig::default()
            },
            ..Config::default()
        };
        let report = analyze(events, &config);

        let co2 = report
            .green_summary
            .co2
            .as_ref()
            .expect("co2 should be Some when default_region is configured");
        assert!(co2.total.mid > 0.0);
        assert!(co2.avoidable.mid > 0.0);
    }

    #[test]
    fn pipeline_empty_traces_no_co2() {
        // With 0 events, compute_carbon_report early-returns
        // (None, vec![]), nothing meaningful to report.
        // Avoids emitting a noisy all-zeros co2 object for empty daemon ticks.
        let config = Config::default();
        let report = analyze(vec![], &config);
        assert!(
            report.green_summary.co2.is_none(),
            "co2 should be None for empty traces"
        );
        assert!(report.green_summary.regions.is_empty());
    }

    #[test]
    fn green_disabled_skips_scoring() {
        use crate::test_helpers::make_n_plus_one_events;
        // 6 events -> N+1 finding, but green scoring disabled
        let events = make_n_plus_one_events();

        let config = Config {
            green: crate::config::GreenConfig {
                enabled: false,
                ..crate::config::GreenConfig::default()
            },
            ..Config::default()
        };
        let report = analyze(events, &config);

        // Findings are still detected
        assert!(!report.findings.is_empty());
        // But green scoring is bypassed
        assert_eq!(report.green_summary.avoidable_io_ops, 0);
        assert!((report.green_summary.io_waste_ratio - 0.0).abs() < f64::EPSILON);
        assert!(report.green_summary.top_offenders.is_empty());
        assert!(report.green_summary.co2.is_none());
        assert!(report.green_summary.regions.is_empty());
        // total_io_ops still counted
        assert_eq!(report.green_summary.total_io_ops, 6);
        // green_impact on findings should be None
        for f in &report.findings {
            assert!(f.green_impact.is_none());
        }
    }

    #[test]
    fn green_disabled_with_region_still_no_co2() {
        let config = Config {
            green: crate::config::GreenConfig {
                enabled: false,
                default_region: Some("eu-west-3".to_string()),
                ..crate::config::GreenConfig::default()
            },
            ..Config::default()
        };
        let report = analyze(vec![], &config);
        assert!(report.green_summary.co2.is_none());
    }

    #[test]
    fn configured_rule_that_cannot_run_emits_a_warning() {
        // Silently skipping the rule reproduces the false green it
        // guards against, so the run must say the gate stayed silent.
        let config = Config {
            thresholds: crate::config::ThresholdsConfig {
                min_usable_span_ratio: Some(0.9),
                ..crate::config::ThresholdsConfig::default()
            },
            ..Config::default()
        };
        // Under the sample floor: 5 I/O-shaped spans.
        let stats = SpanConversionStats {
            received: 5,
            filtered_missing_db_statement: 2,
            retained_sql: 3,
            ..SpanConversionStats::default()
        };
        let (report, _) = analyze_with_traces(vec![], &config, Some(stats));
        assert!(
            report.analysis.ingest.unwrap().usable_span_ratio.is_none(),
            "the floor must suppress the ratio"
        );
        let warning = report
            .warning_details
            .iter()
            .find(|w| w.message.contains("min_usable_span_ratio"))
            .expect("a configured rule that cannot run is reported");
        assert!(
            warning.message.contains("sample floor"),
            "{}",
            warning.message
        );
    }

    /// A configured backend that batch cannot scrape must say so: the run
    /// silently produces estimated figures where the operator expected
    /// measured ones. Built through the real TOML path rather than by hand,
    /// so the test breaks if a backend key is renamed.
    #[test]
    fn configured_energy_backend_in_batch_emits_a_tuning_warning() {
        let config = crate::config::load_from_str(
            "[green]\nenabled = true\n\n[green.scaphandre]\nendpoint = \"http://127.0.0.1:8080/metrics\"\n",
        )
        .unwrap();

        let report = analyze(vec![], &config);

        assert_eq!(
            report.warning_details.len(),
            1,
            "{:?}",
            report.warning_details
        );
        let warning = &report.warning_details[0];
        assert_eq!(warning.kind, crate::report::warnings::TUNING);
        assert!(
            warning.message.contains("[green.scaphandre]"),
            "{warning:?}"
        );
        assert!(warning.message.contains("watch"), "{warning:?}");
    }

    /// One cause, one entry: the remedy is the same for every backend, so
    /// two configured backends must not produce two lines.
    #[test]
    fn several_configured_backends_produce_one_warning() {
        let config = crate::config::load_from_str(
            "[green]\nenabled = true\n\n\
             [green.scaphandre]\nendpoint = \"http://127.0.0.1:8080/metrics\"\n\n\
             [green.electricity_maps]\napi_key = \"k\"\n\n\
             [green.electricity_maps.region_map]\n\"eu-west-3\" = \"FR\"\n",
        )
        .unwrap();

        let report = analyze(vec![], &config);

        assert_eq!(report.warning_details.len(), 1);
        let message = &report.warning_details[0].message;
        assert!(message.contains("[green.scaphandre]"), "{message}");
        assert!(message.contains("[green.electricity_maps]"), "{message}");
    }

    /// Green off means the whole scoring pass is skipped, so the backend was
    /// already inert for a more obvious reason.
    /// The Rust field is `cloud_energy` but the TOML section is
    /// `[green.cloud]`: a message naming the field sends the operator
    /// grepping for a section that does not exist.
    #[test]
    fn every_named_section_is_a_real_toml_section() {
        let toml = "[green]\nenabled = true\n\n\
                    [green.cloud]\nprometheus_endpoint = \"http://127.0.0.1:9090\"\n";
        let config = crate::config::load_from_str(toml).unwrap();

        let message = &analyze(vec![], &config).warning_details[0].message;

        assert!(message.contains("[green.cloud]"), "{message}");
        // The section name must round-trip through the parser.
        assert!(crate::config::load_from_str(toml).is_ok());
    }

    /// Backticks live in the data, like `suggested_fix.recommendation`:
    /// the HTML renders them as code chips, the terminal and TUI strip them
    /// with `strip_code_ticks`. A warning naming a command must mark it.
    #[test]
    fn warning_marks_its_command_as_code() {
        let config = crate::config::load_from_str(
            "[green]\nenabled = true\n\n\
             [green.scaphandre]\nendpoint = \"http://127.0.0.1:8080/metrics\"\n",
        )
        .unwrap();

        let message = &analyze(vec![], &config).warning_details[0].message;

        assert!(message.contains("`analyze`"), "{message}");
        assert!(message.contains("`perf-sentinel watch`"), "{message}");
        assert_eq!(
            crate::text_safety::strip_code_ticks(message)
                .matches('`')
                .count(),
            0,
            "the terminal path must render it tick-free"
        );
    }

    #[test]
    fn green_disabled_does_not_warn_about_backends() {
        let config = crate::config::load_from_str(
            "[green]\nenabled = false\n\n[green.scaphandre]\nendpoint = \"http://127.0.0.1:8080/metrics\"\n",
        )
        .unwrap();

        assert!(analyze(vec![], &config).warning_details.is_empty());
    }

    #[test]
    fn no_warning_when_the_rule_is_not_configured() {
        let report = analyze(vec![], &Config::default());
        assert!(report.warning_details.is_empty());
    }

    // --- ingest stats propagation ---

    #[test]
    fn ingest_stats_land_in_analysis_and_gate() {
        // 4 of 5 I/O-shaped spans unusable: analysis.ingest carries the
        // tally and a configured min_usable_span_ratio fails the gate on
        // an otherwise finding-free run (the false-green scenario).
        let stats = SpanConversionStats {
            received: 30,
            filtered_not_io: 5,
            filtered_missing_db_statement: 20,
            retained_sql: 5,
            ..SpanConversionStats::default()
        };
        let config = Config {
            thresholds: crate::config::ThresholdsConfig {
                min_usable_span_ratio: Some(0.9),
                ..crate::config::ThresholdsConfig::default()
            },
            ..Config::default()
        };
        let (report, _) = analyze_with_traces(vec![], &config, Some(stats));
        let ingest = report.analysis.ingest.expect("tally propagated");
        assert_eq!(ingest.spans_received, 30);
        assert_eq!(ingest.spans_filtered, 25);
        assert!((ingest.usable_span_ratio.unwrap() - 0.2).abs() < f64::EPSILON);
        assert!(!report.quality_gate.passed, "0.2 < 0.9 must fail the gate");
    }

    #[test]
    fn no_ingest_stats_keeps_report_shape_unchanged() {
        let config = Config::default();
        let report = analyze(vec![], &config);
        assert!(report.analysis.ingest.is_none());
        let json = serde_json::to_value(&report).unwrap();
        assert!(
            json["analysis"].get("ingest").is_none(),
            "absent tally must not serialize"
        );
    }

    // --- batch mode stamps a batch confidence (local or CI) ---

    #[test]
    fn batch_analyze_stamps_a_batch_confidence() {
        use crate::test_helpers::make_n_plus_one_events;
        let events = make_n_plus_one_events();
        // Even with a production environment in config, batch analyze must
        // stamp a batch confidence (CiBatch in CI, LocalBatch otherwise),
        // never a daemon level: confidence is mode-driven for `analyze`, the
        // config `daemon.environment` only affects `watch` daemon mode. The
        // exact batch variant depends on the host env (CI var), so assert
        // the family, not a specific value (see batch_for_ci for the map).
        let config = Config {
            daemon: crate::config::DaemonConfig {
                environment: crate::config::DaemonEnvironment::Production,
                ..crate::config::DaemonConfig::default()
            },
            ..Config::default()
        };
        let report = analyze(events, &config);
        assert!(!report.findings.is_empty());
        for f in &report.findings {
            assert!(f.confidence.is_batch(), "got {:?}", f.confidence);
        }
    }

    // ---------------------------------------------------------------
    // Sharded trace routing correctness
    // ---------------------------------------------------------------

    /// Simulate sticky `trace_id` sharding across N instances.
    /// Split events by FNV-1a hash of `trace_id` (same algorithm as
    /// daemon sampling), run the pipeline on each shard independently,
    /// then verify that every per-trace finding from the baseline
    /// (non-sharded) run also appears in the sharded results.
    ///
    /// This validates the claim that horizontal scaling via an `OTel`
    /// Collector `loadbalancingexporter` produces the same per-trace
    /// detection results as a single daemon instance.
    #[test]
    fn sharded_detection_matches_single_instance() {
        use std::collections::{HashMap, HashSet};
        const NUM_SHARDS: u64 = 2;

        // Build a dataset with 4 distinct traces, each containing an
        // N+1 SQL pattern (6 similar queries). Use different services
        // so the shards are non-trivial.
        let traces_data = [
            ("trace-A", "svc-alpha"),
            ("trace-B", "svc-beta"),
            ("trace-C", "svc-gamma"),
            ("trace-D", "svc-delta"),
        ];
        let mut all_events: Vec<SpanEvent> = Vec::new();
        for (trace_id, service) in &traces_data {
            for i in 0..6 {
                let ts = format!("2025-07-10T14:32:01.{i:03}Z");
                let mut ev = crate::test_helpers::make_sql_event(
                    trace_id,
                    &format!("span-{trace_id}-{i}"),
                    &format!("SELECT * FROM orders WHERE id = {}", 100 + i),
                    &ts,
                );
                ev.service = Arc::from(*service);
                all_events.push(ev);
            }
        }

        let config = Config::default();

        // Baseline: all events in a single instance
        let baseline = analyze(all_events.clone(), &config);
        assert!(
            baseline.findings.len() >= 4,
            "expected at least 4 findings (one N+1 per trace), got {}",
            baseline.findings.len()
        );

        // Shard into 2 buckets by FNV-1a hash of trace_id
        let mut shards: Vec<Vec<SpanEvent>> = vec![vec![]; NUM_SHARDS as usize];
        for event in &all_events {
            let hash = fnv1a_hash(event.trace_id.as_bytes());
            let bucket = (hash % NUM_SHARDS) as usize;
            shards[bucket].push(event.clone());
        }

        // Verify each shard got at least one trace (hash distribution)
        for (i, shard) in shards.iter().enumerate() {
            assert!(
                !shard.is_empty(),
                "shard {i} is empty, hash distribution failed"
            );
        }

        // Verify no trace is split across shards
        let mut trace_to_shard: HashMap<String, usize> = HashMap::new();
        for (i, shard) in shards.iter().enumerate() {
            for ev in shard {
                if let Some(&prev) = trace_to_shard.get(&ev.trace_id) {
                    assert_eq!(
                        prev, i,
                        "trace {} split across shards {prev} and {i}",
                        ev.trace_id
                    );
                }
                trace_to_shard.insert(ev.trace_id.clone(), i);
            }
        }

        // Run pipeline on each shard independently
        let mut sharded_findings = Vec::new();
        for shard in shards {
            let report = analyze(shard, &config);
            sharded_findings.extend(report.findings);
        }

        // Build a set of (trace_id, finding_type) for comparison.
        // Cross-trace findings (slow percentiles) are excluded because
        // they depend on seeing all traces, which sharding splits.
        let baseline_set: HashSet<(String, String)> = baseline
            .findings
            .iter()
            .filter(|f| {
                !matches!(
                    f.finding_type,
                    detect::FindingType::SlowSql | detect::FindingType::SlowHttp
                )
            })
            .map(|f| (f.trace_id.clone(), f.finding_type.as_str().to_string()))
            .collect();
        let sharded_set: HashSet<(String, String)> = sharded_findings
            .iter()
            .filter(|f| {
                !matches!(
                    f.finding_type,
                    detect::FindingType::SlowSql | detect::FindingType::SlowHttp
                )
            })
            .map(|f| (f.trace_id.clone(), f.finding_type.as_str().to_string()))
            .collect();

        assert_eq!(
            baseline_set, sharded_set,
            "sharded findings differ from baseline.\n\
             baseline: {baseline_set:?}\n\
             sharded:  {sharded_set:?}"
        );
    }

    /// FNV-1a hash (same algorithm as `daemon::hash_trace_id`).
    fn fnv1a_hash(bytes: &[u8]) -> u64 {
        let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
        for &b in bytes {
            hash ^= u64::from(b);
            hash = hash.wrapping_mul(0x0100_0000_01b3);
        }
        hash
    }
}