skills/ce-code-review/scripts/findings-mechanics.py

#!/usr/bin/env python3
"""Apply deterministic validation, exact dedup, confidence gates, and numbering."""

from __future__ import annotations

import json
import sys
from collections import Counter
from typing import Any


SEVERITIES = ("P0", "P1", "P2", "P3")
CONFIDENCES = (0, 25, 50, 75, 100)
AUTOFIX_CLASSES = ("gated_auto", "manual", "advisory")
OWNERS = ("downstream-resolver", "human", "release")
REQUIRED_TOP = {
    "reviewer": str,
    "findings": list,
    "residual_risks": list,
    "testing_gaps": list,
}
REQUIRED_FINDING = {
    "title": str,
    "severity": str,
    "file": str,
    "line": (int, str),
    "confidence": int,
    "autofix_class": str,
    "owner": str,
    "requires_verification": bool,
    "pre_existing": bool,
}


def valid_return(value: Any) -> bool:
    return isinstance(value, dict) and all(
        isinstance(value.get(key), expected) for key, expected in REQUIRED_TOP.items()
    )


def nonempty_string(value: Any) -> bool:
    return isinstance(value, str) and bool(value.strip())


def valid_finding(value: Any) -> bool:
    if not isinstance(value, dict):
        return False
    if not all(isinstance(value.get(key), expected) for key, expected in REQUIRED_FINDING.items()):
        return False
    if type(value["confidence"]) is not int:
        return False
    if "first_evidence" in value and not nonempty_string(value["first_evidence"]):
        return False
    line = value["line"]
    line_valid = (type(line) is int and line > 0) or (
        isinstance(line, str) and bool(line.strip())
    )
    return (
        value["severity"] in SEVERITIES
        and value["confidence"] in CONFIDENCES
        and value["autofix_class"] in AUTOFIX_CLASSES
        and value["owner"] in OWNERS
        and line_valid
    )


def fingerprint(finding: dict[str, Any]) -> tuple[str, str, str]:
    return (
        finding["file"].strip().lower(),
        str(finding["line"]).strip(),
        " ".join(finding["title"].lower().split()),
    )


def independent_reviewer(name: str, source: dict[str, Any]) -> bool:
    if name == "fast-pass":
        return False
    if name.startswith("adversarial-"):
        return source.get("independence_verified") is True
    return True


def promote(confidence: int) -> int:
    return {50: 75, 75: 100, 100: 100}.get(confidence, confidence)


def merge_group(group: list[tuple[dict[str, Any], str, tuple[str, ...]]]) -> dict[str, Any]:
    # Start with the most urgent/high-confidence representation, then merge conservatively.
    group.sort(key=lambda item: (SEVERITIES.index(item[0]["severity"]), -item[0]["confidence"]))
    merged = dict(group[0][0])
    # A current-diff classification wins a disagreement so an exact duplicate
    # cannot disappear merely because the pre-existing reviewer arrived first.
    merged["pre_existing"] = all(item[0]["pre_existing"] for item in group)
    reviewer_names: list[str] = []
    independent: set[str] = set()
    for finding, reviewer, independent_names in group:
        supplied = finding.get("reviewers")
        names = supplied if isinstance(supplied, list) else [reviewer]
        for name in names:
            if isinstance(name, str) and name not in reviewer_names:
                reviewer_names.append(name)
        independent.update(independent_names)

        if AUTOFIX_CLASSES.index(finding["autofix_class"]) > AUTOFIX_CLASSES.index(merged["autofix_class"]):
            merged["autofix_class"] = finding["autofix_class"]
        if OWNERS.index(finding["owner"]) > OWNERS.index(merged["owner"]):
            merged["owner"] = finding["owner"]
        merged["requires_verification"] = (
            merged["requires_verification"] or finding["requires_verification"]
        )
        if not nonempty_string(merged.get("first_evidence")) and nonempty_string(
            finding.get("first_evidence")
        ):
            merged["first_evidence"] = finding["first_evidence"]
        if not merged.get("suggested_fix") and finding.get("suggested_fix"):
            merged["suggested_fix"] = finding["suggested_fix"]
        if not merged.get("settled_conflict") and finding.get("settled_conflict"):
            merged["settled_conflict"] = finding["settled_conflict"]

    confidence = max(item[0]["confidence"] for item in group)
    has_first_evidence = nonempty_string(merged.get("first_evidence"))
    if confidence >= 75 and not has_first_evidence:
        confidence = 50
    if len(independent) >= 2 and has_first_evidence:
        confidence = promote(confidence)
    merged["confidence"] = confidence
    merged["reviewers"] = reviewer_names
    merged["independent_reviewers"] = [
        reviewer for reviewer in reviewer_names if reviewer in independent
    ]
    return merged


def main() -> int:
    try:
        payload = json.load(sys.stdin)
    except (json.JSONDecodeError, OSError) as error:
        print(json.dumps({"status": "failed", "reason": str(error)}))
        return 2

    if not isinstance(payload, list):
        print(json.dumps({"status": "failed", "reason": "expected an array of reviewer returns"}))
        return 2

    malformed_returns = 0
    malformed_findings = 0
    grouped: dict[
        tuple[str, str, str], list[tuple[dict[str, Any], str, tuple[str, ...]]]
    ] = {}
    residual_risks: list[Any] = []
    testing_gaps: list[Any] = []

    for source in payload:
        if not valid_return(source):
            malformed_returns += 1
            continue
        reviewer = source["reviewer"]
        residual_risks.extend(source["residual_risks"])
        testing_gaps.extend(source["testing_gaps"])
        for finding in source["findings"]:
            if not valid_finding(finding):
                malformed_findings += 1
                continue
            finding = dict(finding)
            if reviewer == "fast-pass":
                finding["confidence"] = min(finding["confidence"], 50)
            independent_names: tuple[str, ...]
            if reviewer == "synthesis":
                supplied_reviewers = finding.get("reviewers")
                supplied_independent = finding.get("independent_reviewers")
                reviewer_set = (
                    {name for name in supplied_reviewers if isinstance(name, str)}
                    if isinstance(supplied_reviewers, list)
                    else set()
                )
                independent_list = (
                    supplied_independent if isinstance(supplied_independent, list) else []
                )
                independent_names = tuple(
                    name
                    for name in independent_list
                    if isinstance(name, str) and name in reviewer_set
                )
            elif independent_reviewer(reviewer, source):
                independent_names = (reviewer,)
            else:
                independent_names = ()
            grouped.setdefault(fingerprint(finding), []).append(
                (finding, reviewer, independent_names)
            )

    merged = [merge_group(group) for group in grouped.values()]
    suppressed: Counter[str] = Counter()
    suppressed_findings: list[dict[str, Any]] = []
    survivors: list[dict[str, Any]] = []
    pre_existing: list[dict[str, Any]] = []
    for finding in merged:
        if finding["pre_existing"]:
            pre_existing.append(finding)
            continue
        if (
            finding["confidence"] < 75
            and finding["severity"] != "P0"
            and not finding.get("settled_conflict")
        ):
            suppressed[str(finding["confidence"])] += 1
            suppressed_findings.append(finding)
            continue
        survivors.append(finding)

    suppressed_findings.sort(
        key=lambda item: (
            SEVERITIES.index(item["severity"]),
            -item["confidence"],
            item["file"].lower(),
            str(item["line"]),
            item["title"].lower(),
        )
    )
    survivors.sort(
        key=lambda item: (
            SEVERITIES.index(item["severity"]),
            -item["confidence"],
            item["file"].lower(),
            str(item["line"]),
            item["title"].lower(),
        )
    )
    for number, finding in enumerate(survivors, 1):
        finding["#"] = number

    print(
        json.dumps(
            {
                "status": "complete",
                "findings": survivors,
                "suppressed_findings": suppressed_findings,
                "pre_existing_findings": pre_existing,
                "residual_risks": residual_risks,
                "testing_gaps": testing_gaps,
                "suppressed_by_confidence": dict(sorted(suppressed.items())),
                "malformed_returns": malformed_returns,
                "malformed_findings": malformed_findings,
            },
            sort_keys=True,
        )
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())