fold/scripts/benchmarks/fold_eval/dispositions.py

"""Reviewed finding dispositions and change-time delta reporting."""

from __future__ import annotations

import hashlib
import json
from collections import Counter
from datetime import datetime
from typing import Any, Mapping, Sequence


DISPOSITION_SCHEMA_VERSION = 1
DISPOSITIONS = frozenset(
    {"actionable", "accepted-risk", "duplicate", "false-positive", "unsupported"}
)


def configuration_fingerprint(configuration: Mapping[str, Any]) -> str:
    encoded = json.dumps(configuration, sort_keys=True, separators=(",", ":")).encode()
    return hashlib.sha256(encoded).hexdigest()


def validate_disposition_baseline(document: Mapping[str, Any]) -> list[str]:
    errors: list[str] = []
    if document.get("schema_version") != DISPOSITION_SCHEMA_VERSION:
        errors.append(f"unsupported disposition schema_version: {document.get('schema_version')}")
    for field in ("target_name", "reviewed_at", "reviewer"):
        if not isinstance(document.get(field), str) or not str(document[field]).strip():
            errors.append(f"{field} must be a non-empty string")
    reviewed_at = document.get("reviewed_at")
    if isinstance(reviewed_at, str):
        try:
            datetime.fromisoformat(reviewed_at.replace("Z", "+00:00"))
        except ValueError:
            errors.append("reviewed_at must be an ISO-8601 timestamp")

    target = document.get("target")
    if not isinstance(target, dict):
        errors.append("target must be an object")
    else:
        for field in ("revision", "tu_set_fingerprint", "configuration_fingerprint"):
            if not isinstance(target.get(field), str) or not target[field]:
                errors.append(f"target.{field} must be a non-empty string")

    findings = document.get("findings")
    if not isinstance(findings, list):
        errors.append("findings must be an array")
        return errors
    seen: set[str] = set()
    for index, finding in enumerate(findings):
        if not isinstance(finding, dict):
            errors.append(f"findings[{index}] must be an object")
            continue
        finding_id = finding.get("id")
        if not isinstance(finding_id, str) or not finding_id:
            errors.append(f"findings[{index}].id must be a non-empty string")
        elif finding_id in seen:
            errors.append(f"duplicate disposition finding id: {finding_id}")
        else:
            seen.add(finding_id)
        if finding.get("status") not in DISPOSITIONS:
            errors.append(f"findings[{index}].status is not a recognized disposition")
        for field in ("rule_id", "rationale"):
            if not isinstance(finding.get(field), str) or not finding[field].strip():
                errors.append(f"findings[{index}].{field} must be a non-empty string")
        location = finding.get("location")
        if location is not None and (
            not isinstance(location, dict)
            or not isinstance(location.get("uri"), str)
            or not isinstance(location.get("line"), int)
            or isinstance(location.get("line"), bool)
            or location.get("line", 0) <= 0
        ):
            errors.append(
                f"findings[{index}].location must contain a URI and positive line"
            )
    return errors


def compare_dispositions(
    current_findings: Sequence[Mapping[str, Any]],
    baseline: Mapping[str, Any],
    *,
    changed_files: Sequence[str] | None = None,
    scope_kloc: float = 0.0,
) -> dict[str, Any]:
    errors = validate_disposition_baseline(baseline)
    if errors:
        raise ValueError("; ".join(errors))
    current: dict[str, Mapping[str, Any]] = {}
    for finding in current_findings:
        finding_id = str(finding.get("finding_id", ""))
        if not finding_id:
            raise ValueError("current finding is missing properties.id")
        if finding_id in current:
            raise ValueError(f"duplicate current finding id: {finding_id}")
        current[finding_id] = finding
    reviewed = {str(item["id"]): item for item in baseline["findings"]}
    current_ids = set(current)
    reviewed_ids = set(reviewed)
    new_ids = sorted(current_ids - reviewed_ids)
    persisting_ids = sorted(current_ids & reviewed_ids)
    resolved_ids = sorted(reviewed_ids - current_ids)
    changed = set(changed_files or [])
    new_in_scope = [
        finding_id
        for finding_id in new_ids
        if not changed or str(current[finding_id].get("uri", "")) in changed
    ]
    current_in_scope = [
        finding_id
        for finding_id in current_ids
        if not changed or str(current[finding_id].get("uri", "")) in changed
    ]
    current_status = Counter(str(reviewed[item]["status"]) for item in persisting_ids)
    resolved_status = Counter(str(reviewed[item]["status"]) for item in resolved_ids)
    new_rules = Counter(str(current[item].get("rule_id", "")) for item in new_ids)
    identity_mismatches = [
        {
            "id": finding_id,
            "reviewed_rule_id": reviewed[finding_id]["rule_id"],
            "current_rule_id": current[finding_id].get("rule_id", ""),
        }
        for finding_id in persisting_ids
        if reviewed[finding_id]["rule_id"] != current[finding_id].get("rule_id", "")
    ]
    moved = []
    for finding_id in persisting_ids:
        location = reviewed[finding_id].get("location")
        if not isinstance(location, dict):
            continue
        before = {"uri": location.get("uri", ""), "line": int(location.get("line", 0))}
        after = {
            "uri": current[finding_id].get("uri", ""),
            "line": int(current[finding_id].get("line", 0)),
        }
        if before != after:
            moved.append({"id": finding_id, "reviewed": before, "current": after})
    current_count = len(current_ids)
    return {
        "new": new_ids,
        "persisting": persisting_ids,
        "resolved": resolved_ids,
        "counts": {
            "current": current_count,
            "reviewed": len(reviewed_ids),
            "new": len(new_ids),
            "persisting": len(persisting_ids),
            "resolved": len(resolved_ids),
            "current_in_scope": len(current_in_scope),
            "new_in_scope": len(new_in_scope),
            "new_in_changed_files": len(new_in_scope) if changed else 0,
        },
        "new_in_changed_files": new_in_scope if changed else [],
        "changed_files": sorted(changed),
        "changed_scope": "explicit-files" if changed else "selected-and-finding-files",
        "by_disposition": dict(sorted(current_status.items())),
        "resolved_by_disposition": dict(sorted(resolved_status.items())),
        "new_by_rule": dict(sorted(new_rules.items())),
        "identity_mismatches": identity_mismatches,
        "moved": moved,
        "review_coverage": round(len(persisting_ids) / current_count, 6)
        if current_count
        else 1.0,
        "scope_kloc": round(scope_kloc, 3),
        "alert_density_per_kloc": round(len(current_in_scope) / scope_kloc, 3)
        if scope_kloc > 0
        else None,
        "new_alert_density_per_kloc": round(len(new_in_scope) / scope_kloc, 3)
        if scope_kloc > 0
        else None,
    }