fold/scripts/benchmarks/fold_eval/flaky_evidence.py

"""Sanitized repeated-test evidence and flake classification."""

from __future__ import annotations

import json
from pathlib import Path

VALID_OUTCOMES = {"pass", "fail", "skip", "missing"}


def parse_meson_testlog(path: Path, attempt: int, exit_code: int) -> dict[str, object]:
    tests: list[dict[str, object]] = []
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except OSError as error:
        return {
            "attempt": attempt,
            "exit_code": exit_code,
            "tests": tests,
            "parse_error": str(error),
        }
    for line_number, line in enumerate(lines, start=1):
        if not line.strip():
            continue
        try:
            record = json.loads(line)
        except json.JSONDecodeError as error:
            return {
                "attempt": attempt,
                "exit_code": exit_code,
                "tests": tests,
                "parse_error": f"line {line_number}: {error}",
            }
        if not isinstance(record, dict) or not isinstance(record.get("name"), str):
            continue
        is_fail = record.get("is_fail")
        result = str(record.get("result", "")).upper()
        if isinstance(is_fail, bool) and is_fail:
            outcome = "fail"
        elif result in {"SKIP", "IGNORED"}:
            outcome = "skip"
        elif isinstance(is_fail, bool) and not is_fail:
            outcome = "pass"
        else:
            outcome = "fail"
        duration = record.get("duration")
        tests.append(
            {
                "name": record["name"],
                "outcome": outcome,
                "duration_seconds": (
                    round(duration, 6)
                    if isinstance(duration, (int, float))
                    and not isinstance(duration, bool)
                    else None
                ),
                "return_code": record.get("returncode"),
                "result": result or "UNKNOWN",
            }
        )
    return {"attempt": attempt, "exit_code": exit_code, "tests": tests}


def classify(outcomes: list[str]) -> str:
    values = set(outcomes)
    if values == {"pass"}:
        return "stable-pass"
    if values == {"fail"}:
        return "stable-fail"
    if values == {"skip"}:
        return "stable-skip"
    if values <= {"pass", "fail"} and values == {"pass", "fail"}:
        return "flaky"
    return "inconsistent"


def summarize_attempts(
    attempts: list[dict[str, object]], *, expected_tests: list[str] | None = None
) -> dict[str, object]:
    names = set(expected_tests or [])
    attempt_maps: list[dict[str, dict[str, object]]] = []
    for attempt in attempts:
        records = attempt.get("tests", [])
        records = records if isinstance(records, list) else []
        by_name: dict[str, dict[str, object]] = {}
        for record in records:
            if isinstance(record, dict) and isinstance(record.get("name"), str):
                by_name[str(record["name"])] = record
                names.add(str(record["name"]))
        attempt_maps.append(by_name)

    tests = []
    counts = {
        "stable-pass": 0,
        "stable-fail": 0,
        "stable-skip": 0,
        "flaky": 0,
        "inconsistent": 0,
    }
    for name in sorted(names):
        outcomes: list[str] = []
        durations: list[float | None] = []
        for by_name in attempt_maps:
            record = by_name.get(name)
            outcome = record.get("outcome") if record is not None else "missing"
            if not isinstance(outcome, str) or outcome not in VALID_OUTCOMES:
                outcome = "missing"
            duration = record.get("duration_seconds") if record is not None else None
            outcomes.append(outcome)
            durations.append(
                duration
                if isinstance(duration, (int, float)) and not isinstance(duration, bool)
                else None
            )
        classification = classify(outcomes)
        counts[classification] += 1
        tests.append(
            {
                "name": name,
                "classification": classification,
                "outcomes": outcomes,
                "duration_seconds": durations,
            }
        )

    exit_codes = []
    parse_errors = []
    for attempt in attempts:
        exit_code = attempt.get("exit_code")
        exit_codes.append(exit_code if isinstance(exit_code, int) else None)
        error = attempt.get("parse_error")
        if isinstance(error, str):
            parse_errors.append({"attempt": attempt.get("attempt"), "error": error})
    all_attempts_passed = bool(attempts) and all(code == 0 for code in exit_codes)
    all_attempts_passed = all_attempts_passed and not parse_errors
    all_attempts_passed = all_attempts_passed and all(
        item["classification"] in {"stable-pass", "stable-skip"} for item in tests
    )
    return {
        "schema_version": 1,
        "attempt_count": len(attempts),
        "exit_codes": exit_codes,
        "all_attempts_passed": all_attempts_passed,
        "counts": counts,
        "flaky_tests": [
            item["name"] for item in tests if item["classification"] == "flaky"
        ],
        "parse_errors": parse_errors,
        "tests": tests,
    }