k4bench.regression.report_builder

k4bench.regression.report_builder

Assemble the nightly regression report from the EOS run history.

Walks every (detector, platform, sample) triple found under the WebEOS data URL (the same hierarchy the dashboard's sidebar cascades through — these triples have independent baselines and are never pooled), pulls a trailing window of runs into the local cache, rebuilds the trend frames with :mod:k4bench.analysis.trend, attaches per-run reliability verdicts with :mod:k4bench.results.reliability_evidence, and runs the step detector in :mod:k4bench.regression.engine over every metric series.

workload_map

workload_map(results_df: DataFrame | None) -> dict[str, int]

{run_id: random_seed} for every run that recorded one.

A run absent from this map has an unknown workload (see :data:~k4bench.regression.engine.WORKLOAD_UNKNOWN) — it drew a fresh event mix, or predates the seed being recorded. The engine re-anchors its baseline when this value changes, so the map only has to say what was recorded, never why.

Source code in k4bench/regression/report_builder.py

def workload_map(results_df: pd.DataFrame | None) -> dict[str, int]:
    """``{run_id: random_seed}`` for every run that recorded one.

    A run absent from this map has an unknown workload (see
    :data:`~k4bench.regression.engine.WORKLOAD_UNKNOWN`) — it drew a fresh
    event mix, or predates the seed being recorded. The engine re-anchors its
    baseline when this value changes, so the map only has to say what was
    recorded, never why.
    """
    if (
        results_df is None or results_df.empty
        or "random_seed" not in results_df.columns
    ):
        return {}
    sub = results_df[["run_id", "random_seed"]].dropna().drop_duplicates("run_id")
    out: dict[str, int] = {}
    for row in sub.itertuples(index=False):
        try:
            out[str(row.run_id)] = int(row.random_seed)
        except (TypeError, ValueError):
            continue
    return out

unjudged_value_verdicts

unjudged_value_verdicts(*, detector: str, platform: str, sample: str, results_df: DataFrame | None, event_df: DataFrame | None, tonight: str, already: set[tuple[str, str]]) -> list[MetricVerdict]

Raw metric values for tonight's run as unjudged UNKNOWN verdicts.

Two different things end up here. The engine skips unreliable runs (they must not pollute baselines or flags), so their metrics get no verdict and their values would never reach the report the dashboard's Overview tab reads — leaving that tab unable to plot them even with "Exclude unreliable runs" off. And :data:REPORTED_ONLY_METRICS are never judged on any night by design, but are still worth being able to look up.

Either way this records tonight's raw value for every (label, metric) not already judged, marked UNKNOWN (never a flag), so the value is preserved for display. A normally-judged run is already covered.

Source code in k4bench/regression/report_builder.py

def unjudged_value_verdicts(
    *,
    detector: str,
    platform: str,
    sample: str,
    results_df: pd.DataFrame | None,
    event_df: pd.DataFrame | None,
    tonight: str,
    already: set[tuple[str, str]],
) -> list[MetricVerdict]:
    """Raw metric values for *tonight*'s run as unjudged ``UNKNOWN`` verdicts.

    Two different things end up here. The engine skips unreliable runs (they
    must not pollute baselines or flags), so their metrics get no verdict and
    their values would never reach the report the dashboard's Overview tab
    reads — leaving that tab unable to plot them even with "Exclude unreliable
    runs" off. And :data:`REPORTED_ONLY_METRICS` are never judged on any night
    by design, but are still worth being able to look up.

    Either way this records tonight's raw value for every ``(label, metric)``
    not *already* judged, marked ``UNKNOWN`` (never a flag), so the value is
    preserved for display. A normally-judged run is already covered.
    """
    out: list[MetricVerdict] = []

    def _emit(df: pd.DataFrame | None, metrics: dict[str, str]) -> None:
        if df is None or df.empty:
            return
        tonight_rows = df[df["run_id"] == tonight]
        for label in sorted(tonight_rows["label"].dropna().unique()):
            row = tonight_rows[tonight_rows["label"] == label]
            for metric, family in metrics.items():
                if metric not in row.columns or (str(label), metric) in already:
                    continue
                val = row[metric].iloc[0]
                if pd.isna(val) or not math.isfinite(float(val)):
                    continue
                out.append(MetricVerdict(
                    detector=detector, platform=platform, sample=sample,
                    label=str(label), metric_family=family, metric=metric,
                    sub_detector=None, run_id=tonight, run_date=tonight,
                    value=float(val), baseline_median=None, baseline_mad=None,
                    pct_change=None, z_score=None,
                    severity=Severity.UNKNOWN, direction=Direction.NONE,
                    reason=(
                        "recorded but not judged — reports the same measurement "
                        "as an already-judged metric"
                        if metric in REPORTED_ONLY_METRICS else
                        "unreliable host — value recorded but not judged"
                    ),
                ))

    results = with_cpu_efficiency(results_df) if results_df is not None else None
    _emit(results, RUN_VALUE_METRICS)
    _emit(event_df, EVENT_METRICS)
    return out

evaluate_group_series

evaluate_group_series(*, detector: str, platform: str, sample: str, results_df: DataFrame | None, event_df: DataFrame | None, reliability: dict[str, bool | None], hosts: dict[str, HostFact] | None = None) -> dict[SeriesId, list[MetricVerdict]]

Run the step detector over every run/event metric series of one run group. Region timings are not walked.

Returns the full verdict series per :class:SeriesId — the nightly report takes each series' verdict for the report night, while the dashboard drill-down and the retrospective threshold validation consume the whole walk.

hosts (from :func:~k4bench.regression.history.host_facts) names the machine behind each run, and only reaches the history tails attached to confirmed verdicts; it never enters the judgement itself. Omitted, the tails simply carry no host.

Where a run group has enough configurations to support it, each metric is first decomposed into a group-wide common mode and per-config residuals (:mod:k4bench.regression.common_mode). Both halves are walked: the shift as one group-level series under :data:~k4bench.regression.common_mode.COMMON_MODE_LABEL, and each config on what is left after it. A night where the whole group moved together therefore produces one finding instead of one per config, without the move going unjudged.

Source code in k4bench/regression/report_builder.py

def evaluate_group_series(
    *,
    detector: str,
    platform: str,
    sample: str,
    results_df: pd.DataFrame | None,
    event_df: pd.DataFrame | None,
    reliability: dict[str, bool | None],
    hosts: dict[str, HostFact] | None = None,
) -> dict[SeriesId, list[MetricVerdict]]:
    """Run the step detector over every run/event metric series of one run
    group. Region timings are not walked.

    Returns the **full verdict series** per :class:`SeriesId` — the nightly
    report takes each series' verdict for the report night, while the
    dashboard drill-down and the retrospective threshold validation consume
    the whole walk.

    *hosts* (from :func:`~k4bench.regression.history.host_facts`) names the
    machine behind each run, and only reaches the history tails attached to
    confirmed verdicts; it never enters the judgement itself. Omitted, the tails
    simply carry no host.

    Where a run group has enough configurations to support it, each metric is
    first decomposed into a group-wide common mode and per-config residuals
    (:mod:`k4bench.regression.common_mode`). Both halves are walked: the shift
    as one group-level series under
    :data:`~k4bench.regression.common_mode.COMMON_MODE_LABEL`, and each config
    on what is left after it. A night where the whole group moved together
    therefore produces one finding instead of one per config, without the move
    going unjudged.
    """
    out: dict[SeriesId, list[MetricVerdict]] = {}
    workloads = workload_map(results_df)

    def _walk(df: pd.DataFrame, metrics: dict[str, str]) -> None:
        labels = sorted(df["label"].dropna().unique())
        run_dates = dict(zip(df["run_id"].astype(str), df["x_date"], strict=True))
        # Too few configurations for a cross-config median to mean anything:
        # judge every series exactly as measured.
        decomposable = len(labels) >= MIN_COMMON_MODE_CONFIGS

        for metric, family in metrics.items():
            if metric not in df.columns:
                continue
            # A ratio in percentage points has no multiplicative common mode to
            # divide out, so it is judged as measured whatever the group size.
            shifts = (
                common_mode_shifts(df, metric)
                if decomposable and family not in ABSOLUTE_FLOOR_FAMILIES
                else {}
            )
            for label in labels:
                name = str(label)
                sid = SeriesId(detector, platform, sample, name, family, metric)
                history = _series_history(
                    df, df["label"] == label, metric, reliability,
                    shifts=shifts, workloads=workloads,
                )
                verdicts = evaluate_series(history, series=sid)
                if verdicts:
                    out[sid] = _with_history(history, verdicts, hosts or {})

            if not shifts:
                continue
            # The common mode itself, judged as its own series.
            group_sid = SeriesId(
                detector, platform, sample, COMMON_MODE_LABEL, family, metric,
            )
            group_history = shift_history(
                shifts, run_dates, reliability, workloads=workloads,
            )
            group_verdicts = evaluate_series(group_history, series=group_sid)
            if group_verdicts:
                out[group_sid] = _with_history(
                    group_history, group_verdicts, hosts or {},
                )

    if results_df is not None and not results_df.empty:
        _walk(with_cpu_efficiency(results_df), RUN_METRICS)

    if event_df is not None and not event_df.empty:
        _walk(event_df, EVENT_METRICS)

    return out

build_group_report

build_group_report(data_url: str, cache_dir: str | None, detector: str, platform: str, sample: str, *, fetch_window_runs: int = FETCH_WINDOW_RUNS, as_of: str | None = None) -> RunGroupReport | None

Build one triple's report from its trailing run window, or None when the triple has no fetchable runs at all.

as_of (a YYYY-MM-DD night) truncates the run history to runs on or before that night before the trailing window is taken, reproducing the report that night's runs would have produced — the seam the historical backfill drives. None judges the full history (the nightly CI case).

Source code in k4bench/regression/report_builder.py

def build_group_report(
    data_url: str,
    cache_dir: str | None,
    detector: str,
    platform: str,
    sample: str,
    *,
    fetch_window_runs: int = FETCH_WINDOW_RUNS,
    as_of: str | None = None,
) -> RunGroupReport | None:
    """Build one triple's report from its trailing run window, or ``None``
    when the triple has no fetchable runs at all.

    *as_of* (a ``YYYY-MM-DD`` night) truncates the run history to runs on or
    before that night before the trailing window is taken, reproducing the
    report that night's runs would have produced — the seam the historical
    backfill drives. ``None`` judges the full history (the nightly CI case).
    """
    stacks_dates = list_run_dates_all_stacks(data_url, detector, platform, sample)
    pairs = sorted(
        (date, stack) for stack, dates in stacks_dates.items() for date in dates
        if as_of is None or date <= as_of
    )[-fetch_window_runs:]
    if not pairs:
        return None
    window: dict[str, list[str]] = {}
    for date, stack in pairs:
        window.setdefault(stack, []).append(date)
    runs = fetch_runs_windowed(data_url, detector, platform, sample, window, cache_root=cache_dir)
    if not runs:
        return None
    run_dirs = tuple(r["run_dir"] for r in sorted(runs, key=lambda r: r["date"]))
    return group_report_from_run_dirs(detector, platform, sample, run_dirs)

group_report_from_run_dirs

group_report_from_run_dirs(detector: str, platform: str, sample: str, run_dirs: tuple[str, ...]) -> RunGroupReport | None

Build one triple's report from already-local run directories (ordered oldest → newest; each directory's name is its nightly date).

Source code in k4bench/regression/report_builder.py

def group_report_from_run_dirs(
    detector: str,
    platform: str,
    sample: str,
    run_dirs: tuple[str, ...],
) -> RunGroupReport | None:
    """Build one triple's report from already-local run directories (ordered
    oldest → newest; each directory's name is its nightly date)."""
    if not run_dirs:
        return None
    tonight = max(Path(d).name for d in run_dirs)
    tonight_meta = parse_run_dir(
        next(Path(d) for d in run_dirs if Path(d).name == tonight)
    )
    results_df = build_results_trend(run_dirs)
    event_df = build_event_timing_trend(run_dirs)
    machine_df = build_machine_info_trend(run_dirs)
    reliability = run_reliability_map(results_df, machine_df)
    group = _group_report_from_frames(
        detector, platform, sample,
        results_df=results_df, event_df=event_df,
        reliability=reliability, tonight=tonight,
        hosts=host_facts(machine_df),
        configured_labels=tonight_meta["configured_labels"],
    )
    if group is None:
        return None
    # A night that wrote no result CSV has no release in its (absent) rows;
    # run_info still names the stack that failed.
    if not group.k4h_release:
        group.k4h_release = tonight_meta["k4h_release"] or ""
    return _with_region_deltas(
        group, run_dirs, judgeable_config_keys(results_df),
    )

build_nightly_report

build_nightly_report(data_url: str, cache_dir: str | None = None, *, fetch_window_runs: int = FETCH_WINDOW_RUNS, as_of: str | None = None) -> NightlyReport

Build the cross-detector report for the most recent nightly.

The report night is the newest run date seen across all triples. A triple dated earlier is still reported normally when its CI run says it came from the report night's own batch, whatever the gap between the two dates (see :func:_same_batch); for a night whose runs carry no CI run at all, a lag of up to :data:SAME_BATCH_LAG_DAYS stands in for that. Anything else gets a missing run job failure (a hard crash uploads nothing, so absence is itself the failure signal) — unless it is stale by more than :data:MISSING_RUN_GRACE_DAYS, in which case it is treated as retired and dropped.

as_of truncates every triple's history to runs on or before that night (see :func:build_group_report), making the report night the newest run ≤ as_of — the historical-backfill seam.

Source code in k4bench/regression/report_builder.py

def build_nightly_report(
    data_url: str,
    cache_dir: str | None = None,
    *,
    fetch_window_runs: int = FETCH_WINDOW_RUNS,
    as_of: str | None = None,
) -> NightlyReport:
    """Build the cross-detector report for the most recent nightly.

    The report night is the newest run date seen across all triples. A triple
    dated earlier is still reported normally when its CI run says it came from
    the report night's own batch, whatever the gap between the two dates (see
    :func:`_same_batch`); for a night whose runs carry no CI run at all, a lag
    of up to :data:`SAME_BATCH_LAG_DAYS` stands in for that. Anything else gets
    a *missing run* job failure (a hard crash uploads nothing, so absence is
    itself the failure signal) — unless it is stale by more than
    :data:`MISSING_RUN_GRACE_DAYS`, in which case it is treated as retired and
    dropped.

    *as_of* truncates every triple's history to runs on or before that night
    (see :func:`build_group_report`), making the report night the newest run
    ≤ *as_of* — the historical-backfill seam.
    """
    groups: list[RunGroupReport] = []
    for detector in list_detectors(data_url):
        for platform in list_platforms(data_url, detector):
            stack_samples = scan_stack_samples(data_url, detector, platform)
            samples = sorted({s for ss in stack_samples.values() for s in ss})
            for sample in samples:
                try:
                    group = build_group_report(
                        data_url, cache_dir, detector, platform, sample,
                        fetch_window_runs=fetch_window_runs, as_of=as_of,
                    )
                except Exception:
                    _log.exception(
                        "build_nightly_report: failed for %s/%s/%s",
                        detector, platform, sample,
                    )
                    continue
                if group is not None:
                    groups.append(group)

    return _finalize_report(groups)

build_nightly_report_local

build_nightly_report_local(data_dir: str, *, fetch_window_runs: int = FETCH_WINDOW_RUNS, as_of: str | None = None) -> NightlyReport

Like :func:build_nightly_report, but over a local directory tree with the same {detector}/{platform}/{stack}/{sample}/{date} layout as EOS (used by the integration test and for offline dry-runs; no network). as_of truncates each sample's runs the same way.

Source code in k4bench/regression/report_builder.py

def build_nightly_report_local(
    data_dir: str,
    *,
    fetch_window_runs: int = FETCH_WINDOW_RUNS,
    as_of: str | None = None,
) -> NightlyReport:
    """Like :func:`build_nightly_report`, but over a local directory tree with
    the same ``{detector}/{platform}/{stack}/{sample}/{date}`` layout as EOS
    (used by the integration test and for offline dry-runs; no network).
    *as_of* truncates each sample's runs the same way."""
    root = Path(data_dir)
    groups: list[RunGroupReport] = []
    for det_dir in sorted(p for p in root.iterdir() if p.is_dir()):
        if det_dir.name.startswith(("_", ".")):
            continue
        for plat_dir in sorted(p for p in det_dir.iterdir() if p.is_dir()):
            # Collect each sample's run dirs across all stacks.
            per_sample: dict[str, list[Path]] = {}
            for stack_dir in sorted(p for p in plat_dir.iterdir() if p.is_dir()):
                for sample_dir in sorted(p for p in stack_dir.iterdir() if p.is_dir()):
                    per_sample.setdefault(sample_dir.name, []).extend(
                        p for p in sample_dir.iterdir() if p.is_dir()
                    )
            for sample, run_paths in sorted(per_sample.items()):
                run_dirs = tuple(
                    str(p) for p in sorted(run_paths, key=lambda p: p.name)
                    if as_of is None or p.name <= as_of
                )[-fetch_window_runs:]
                group = group_report_from_run_dirs(
                    det_dir.name, plat_dir.name, sample, run_dirs
                )
                if group is not None:
                    groups.append(group)
    return _finalize_report(groups)