fold/scripts/benchmarks/fold_eval/performance.py

"""Performance attribution shared by real-project benchmark runners."""

from __future__ import annotations

from typing import Any, Mapping

from .metrics import paired_comparison


SOURCE_PHASE_METRICS = (
    "function_fingerprint_ms",
    "call_graph_ms",
    "security_candidate_ms",
    "cross_tu_property_ms",
    "symbolic_summary_ms",
)

GRAPH_SUBPHASE_METRICS = (
    "project_index_load_ms",
    "project_graph_collect_ms",
    "project_graph_merge_ms",
    "project_graph_fallback_ms",
    "project_index_contribution_update_ms",
    "project_index_snapshot_update_ms",
    "security_candidate_ms",
)


def optimization_surface(profile: dict[str, Any]) -> dict[str, Any]:
    """Attribute analysis time without treating the residual as detector cost.

    Detector timers and the source phase timers are disjoint by construction.
    The remaining time is still intentionally labelled unattributed: it
    includes the main Clang traversal, detector setup not covered by detector
    timers, and orchestration overhead.
    """

    scan = profile.get("scan", {})
    analyze_ms = float(scan.get("analyze_ms", {}).get("median", 0))
    detector_rows: list[dict[str, Any]] = []
    detector_total = 0.0
    for name, metrics in profile.get("detectors", {}).items():
        wall_ms = float(metrics.get("wall_ms", {}).get("median", 0))
        detector_total += wall_ms
        detector_rows.append(
            {
                "detector": name,
                "wall_ms": wall_ms,
                "analyze_share_pct": (
                    round(100.0 * wall_ms / analyze_ms, 2) if analyze_ms else 0.0
                ),
                "z3_queries": metrics.get("z3_queries", {}).get("median", 0),
                "z3_timeouts": metrics.get("z3_timeouts", {}).get("max", 0),
            }
        )
    detector_rows.sort(key=lambda item: (-float(item["wall_ms"]), item["detector"]))

    phase_rows: list[dict[str, Any]] = []
    phase_total = 0.0
    phase_metrics: list[str] = list(SOURCE_PHASE_METRICS)
    # Rust, Swift, and binary routes do their expensive frontend work outside
    # the Clang source phases. Keep source reports unchanged when it is zero,
    # but surface the phase when a non-Clang profile recorded it.
    if float(scan.get("frontend_extraction_ms", {}).get("median", 0)) > 0:
        phase_metrics.append("frontend_extraction_ms")
    for name in phase_metrics:
        wall_ms = float(scan.get(name, {}).get("median", 0))
        phase_total += wall_ms
        phase_rows.append(
            {
                "phase": name.removesuffix("_ms"),
                "wall_ms": wall_ms,
                "analyze_share_pct": (
                    round(100.0 * wall_ms / analyze_ms, 2) if analyze_ms else 0.0
                ),
            }
        )
    phase_rows.sort(key=lambda item: (-float(item["wall_ms"]), item["phase"]))
    graph_rows = [
        {
            "phase": name.removesuffix("_ms"),
            "wall_ms": float(scan.get(name, {}).get("median", 0)),
        }
        for name in GRAPH_SUBPHASE_METRICS
    ]
    graph_rows.sort(key=lambda item: (-float(item["wall_ms"]), item["phase"]))

    non_detector_ms = max(0.0, analyze_ms - detector_total)
    return {
        "ranked_detectors": detector_rows,
        "ranked_source_phases": phase_rows,
        "ranked_graph_subphases": graph_rows,
        # Retained for baseline compatibility: all time not attributed to a
        # detector, including the newly named source phases.
        "shared_frontend_and_orchestration_ms": round(non_detector_ms, 3),
        "known_source_phase_ms": round(phase_total, 3),
        "unattributed_frontend_and_orchestration_ms": round(
            max(0.0, non_detector_ms - phase_total), 3
        ),
        "peak_rss_mb": round(
            float(scan.get("peak_rss_kb", {}).get("median", 0)) / 1024.0, 3
        ),
    }


def compare_performance_lanes(
    current: Mapping[str, Any],
    baseline: Mapping[str, Any],
    *,
    max_wall_percent: float,
) -> tuple[dict[str, Any], list[str]]:
    """Compare matching lanes and state whether evidence is paired.

    New summaries retain repetition samples, allowing trial ``n`` to be paired
    with the same target/configuration trial in the baseline. Older summaries
    remain readable: when samples are absent or counts differ, the comparison
    is explicitly labelled as a median-only fallback.
    """

    comparisons: dict[str, Any] = {}
    failures: list[str] = []
    current_performance = current.get("performance", {})
    baseline_performance = baseline.get("performance", {})
    if not isinstance(current_performance, Mapping) or not isinstance(
        baseline_performance, Mapping
    ):
        return comparisons, failures

    for lane in ("profiled_serial", "throughput"):
        current_lane = current_performance.get(lane)
        baseline_lane = baseline_performance.get(lane)
        if not isinstance(current_lane, Mapping) or not isinstance(
            baseline_lane, Mapping
        ):
            continue

        current_wall = current_lane.get("wall_seconds", {})
        baseline_wall = baseline_lane.get("wall_seconds", {})
        before = (
            float(baseline_wall.get("median", 0))
            if isinstance(baseline_wall, Mapping)
            else 0.0
        )
        after = (
            float(current_wall.get("median", 0))
            if isinstance(current_wall, Mapping)
            else 0.0
        )
        median_change = 100.0 * (after - before) / before if before > 0 else None
        comparison: dict[str, Any] = {
            "method": "median-only",
            "baseline_median_seconds": before,
            "candidate_median_seconds": after,
            "median_change_pct": median_change,
        }

        current_samples = current_lane.get("wall_samples_seconds")
        baseline_samples = baseline_lane.get("wall_samples_seconds")
        if (
            isinstance(current_samples, list)
            and isinstance(baseline_samples, list)
            and current_samples
            and len(current_samples) == len(baseline_samples)
        ):
            comparison["method"] = "paired-by-repetition"
            comparison["paired"] = paired_comparison(
                [float(value) for value in baseline_samples],
                [float(value) for value in current_samples],
            )
        elif isinstance(current_samples, list) or isinstance(baseline_samples, list):
            comparison["fallback_reason"] = "missing or unequal repetition samples"

        comparisons[lane] = comparison
        if median_change is not None and median_change > max_wall_percent:
            failures.append(
                f"{lane} wall time regressed {median_change:.1f}% "
                f"(limit {max_wall_percent:.1f}%)"
            )
    return comparisons, failures