2dcbdbc75d
Preserve source, binary, and runtime evidence boundaries while making precision and performance regressions reproducible. - decompose analysis orchestration into bounded modules - add differential, metamorphic, organic, and Yams change-time gates - fix detector, P-Code, fuzz sanitizer, and report identity precision
76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
"""Bug-centric confirmation progress metrics inspired by Magma."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
from collections.abc import Mapping, Sequence
|
|
from typing import Any
|
|
|
|
from .metrics import distribution_summary
|
|
|
|
|
|
CONFIRMATION_STAGES = ("reached", "triggered", "detected", "replayed")
|
|
|
|
|
|
def validate_confirmation_progress(progress: Mapping[str, Any]) -> list[str]:
|
|
errors: list[str] = []
|
|
values: list[bool] = []
|
|
for stage in CONFIRMATION_STAGES:
|
|
value = progress.get(stage, False)
|
|
if not isinstance(value, bool):
|
|
errors.append(f"{stage} must be boolean")
|
|
value = False
|
|
values.append(value)
|
|
timing = progress.get(f"time_to_{stage}_ms")
|
|
if timing is not None and (
|
|
isinstance(timing, bool)
|
|
or not isinstance(timing, (int, float))
|
|
or float(timing) < 0
|
|
):
|
|
errors.append(f"time_to_{stage}_ms must be non-negative when present")
|
|
if timing is not None and not value:
|
|
errors.append(f"time_to_{stage}_ms requires {stage}=true")
|
|
for index in range(1, len(values)):
|
|
if values[index] and not values[index - 1]:
|
|
errors.append(
|
|
f"{CONFIRMATION_STAGES[index]} requires "
|
|
f"{CONFIRMATION_STAGES[index - 1]}"
|
|
)
|
|
return errors
|
|
|
|
|
|
def summarize_confirmation_results(results: Sequence[Mapping[str, Any]]) -> dict[str, Any]:
|
|
by_status: Counter[str] = Counter()
|
|
stage_counts = {stage: 0 for stage in CONFIRMATION_STAGES}
|
|
timings: dict[str, list[float]] = {stage: [] for stage in CONFIRMATION_STAGES}
|
|
errors: list[str] = []
|
|
finding_ids: set[str] = set()
|
|
for index, result in enumerate(results):
|
|
by_status[str(result.get("status", "unknown"))] += 1
|
|
finding_id = str(result.get("finding_id", ""))
|
|
if finding_id:
|
|
finding_ids.add(finding_id)
|
|
progress = result.get("progress", {})
|
|
if not isinstance(progress, dict):
|
|
errors.append(f"results[{index}].progress must be an object")
|
|
continue
|
|
errors.extend(
|
|
f"results[{index}].progress.{error}"
|
|
for error in validate_confirmation_progress(progress)
|
|
)
|
|
for stage in CONFIRMATION_STAGES:
|
|
if progress.get(stage, False):
|
|
stage_counts[stage] += 1
|
|
timing = progress.get(f"time_to_{stage}_ms")
|
|
if isinstance(timing, (int, float)) and not isinstance(timing, bool):
|
|
timings[stage].append(float(timing))
|
|
return {
|
|
"results": len(results),
|
|
"unique_finding_ids": len(finding_ids),
|
|
"by_status": dict(sorted(by_status.items())),
|
|
"stage_counts": stage_counts,
|
|
"time_to_stage_ms": {
|
|
stage: distribution_summary(timings[stage]) for stage in CONFIRMATION_STAGES
|
|
},
|
|
"validation_errors": errors,
|
|
}
|