Skills › Compound Engineering › ce-code-review › findings-mechanics.py
findings-mechanics.py
274 lines · skills/ce-code-review/scripts/findings-mechanics.py at 3449230
1#!/usr/bin/env python3
2"""Apply deterministic validation, exact dedup, confidence gates, and numbering."""
4from __future__ import annotations
6import json
7import sys
8from collections import Counter
9from typing import Any
12SEVERITIES = ("P0", "P1", "P2", "P3")
13CONFIDENCES = (0, 25, 50, 75, 100)
14AUTOFIX_CLASSES = ("gated_auto", "manual", "advisory")
15OWNERS = ("downstream-resolver", "human", "release")
16REQUIRED_TOP = {
17 "reviewer": str,
18 "findings": list,
19 "residual_risks": list,
20 "testing_gaps": list,
21}
22REQUIRED_FINDING = {
23 "title": str,
24 "severity": str,
25 "file": str,
26 "line": (int, str),
27 "confidence": int,
28 "autofix_class": str,
29 "owner": str,
30 "requires_verification": bool,
31 "pre_existing": bool,
32}
35def valid_return(value: Any) -> bool:
36 return isinstance(value, dict) and all(
37 isinstance(value.get(key), expected) for key, expected in REQUIRED_TOP.items()
38 )
41def nonempty_string(value: Any) -> bool:
42 return isinstance(value, str) and bool(value.strip())
45def valid_finding(value: Any) -> bool:
46 if not isinstance(value, dict):
47 return False
48 if not all(isinstance(value.get(key), expected) for key, expected in REQUIRED_FINDING.items()):
49 return False
50 if type(value["confidence"]) is not int:
51 return False
52 if "first_evidence" in value and not nonempty_string(value["first_evidence"]):
53 return False
54 line = value["line"]
55 line_valid = (type(line) is int and line > 0) or (
56 isinstance(line, str) and bool(line.strip())
57 )
58 return (
59 value["severity"] in SEVERITIES
60 and value["confidence"] in CONFIDENCES
61 and value["autofix_class"] in AUTOFIX_CLASSES
62 and value["owner"] in OWNERS
63 and line_valid
64 )
67def fingerprint(finding: dict[str, Any]) -> tuple[str, str, str]:
68 return (
69 finding["file"].strip().lower(),
70 str(finding["line"]).strip(),
71 " ".join(finding["title"].lower().split()),
72 )
75def independent_reviewer(name: str, source: dict[str, Any]) -> bool:
76 if name == "fast-pass":
77 return False
78 if cross_model_peer(name):
79 return source.get("independence_verified") is True
80 return True
83def cross_model_peer(name: str) -> bool:
84 return name.startswith("adversarial-")
87def promote(confidence: int) -> int:
88 return {50: 75, 75: 100, 100: 100}.get(confidence, confidence)
91def merge_group(group: list[tuple[dict[str, Any], str, tuple[str, ...]]]) -> dict[str, Any]:
92 # Start with the most urgent/high-confidence representation, then merge conservatively.
93 group.sort(key=lambda item: (SEVERITIES.index(item[0]["severity"]), -item[0]["confidence"]))
94 merged = dict(group[0][0])
95 # A current-diff classification wins a disagreement so an exact duplicate
96 # cannot disappear merely because the pre-existing reviewer arrived first.
97 merged["pre_existing"] = all(item[0]["pre_existing"] for item in group)
98 reviewer_names: list[str] = []
99 independent: set[str] = set()
100 for finding, reviewer, independent_names in group:
101 supplied = finding.get("reviewers")
102 names = supplied if isinstance(supplied, list) else [reviewer]
103 for name in names:
104 if isinstance(name, str) and name not in reviewer_names:
105 reviewer_names.append(name)
106 independent.update(independent_names)
108 if AUTOFIX_CLASSES.index(finding["autofix_class"]) > AUTOFIX_CLASSES.index(merged["autofix_class"]):
109 merged["autofix_class"] = finding["autofix_class"]
110 if OWNERS.index(finding["owner"]) > OWNERS.index(merged["owner"]):
111 merged["owner"] = finding["owner"]
112 merged["requires_verification"] = (
113 merged["requires_verification"] or finding["requires_verification"]
114 )
115 if not nonempty_string(merged.get("first_evidence")) and nonempty_string(
116 finding.get("first_evidence")
117 ):
118 merged["first_evidence"] = finding["first_evidence"]
119 if not merged.get("suggested_fix") and finding.get("suggested_fix"):
120 merged["suggested_fix"] = finding["suggested_fix"]
121 if not merged.get("settled_conflict") and finding.get("settled_conflict"):
122 merged["settled_conflict"] = finding["settled_conflict"]
124 confidence = max(item[0]["confidence"] for item in group)
125 has_first_evidence = nonempty_string(merged.get("first_evidence"))
126 if confidence >= 75 and not has_first_evidence:
127 confidence = 50
128 # In-process reviewers share one serving model, so their agreement is one
129 # reading repeated; only a verified cross-model peer corroborates.
130 if len(independent) >= 2 and has_first_evidence and any(cross_model_peer(n) for n in independent):
131 confidence = promote(confidence)
132 merged["confidence"] = confidence
133 merged["reviewers"] = reviewer_names
134 merged["independent_reviewers"] = [
135 reviewer for reviewer in reviewer_names if reviewer in independent
136 ]
137 return merged
140def main() -> int:
141 try:
142 payload = json.load(sys.stdin)
143 except (json.JSONDecodeError, OSError) as error:
144 print(json.dumps({"status": "failed", "reason": str(error)}))
145 return 2
147 if not isinstance(payload, list):
148 print(json.dumps({"status": "failed", "reason": "expected an array of reviewer returns"}))
149 return 2
151 malformed_returns = 0
152 malformed_findings = 0
153 first_evidence_backfilled = 0
154 grouped: dict[
155 tuple[str, str, str], list[tuple[dict[str, Any], str, tuple[str, ...]]]
156 ] = {}
157 residual_risks: list[Any] = []
158 testing_gaps: list[Any] = []
160 for source in payload:
161 if not valid_return(source):
162 malformed_returns += 1
163 continue
164 reviewer = source["reviewer"]
165 residual_risks.extend(source["residual_risks"])
166 testing_gaps.extend(source["testing_gaps"])
167 for finding in source["findings"]:
168 backfilled = False
169 if isinstance(finding, dict):
170 finding = dict(finding)
171 evidence = finding.get("evidence")
172 if (
173 finding.get("confidence") in (75, 100)
174 and ("first_evidence" not in finding or isinstance(finding["first_evidence"], str))
175 and not nonempty_string(finding.get("first_evidence"))
176 and isinstance(evidence, list)
177 and evidence
178 and nonempty_string(evidence[0])
179 ):
180 finding["first_evidence"] = evidence[0]
181 backfilled = True
182 if not valid_finding(finding):
183 malformed_findings += 1
184 continue
185 first_evidence_backfilled += int(backfilled)
186 if reviewer == "fast-pass":
187 finding["confidence"] = min(finding["confidence"], 50)
188 independent_names: tuple[str, ...]
189 if reviewer == "synthesis":
190 supplied_reviewers = finding.get("reviewers")
191 supplied_independent = finding.get("independent_reviewers")
192 reviewer_set = (
193 {name for name in supplied_reviewers if isinstance(name, str)}
194 if isinstance(supplied_reviewers, list)
195 else set()
196 )
197 independent_list = (
198 supplied_independent if isinstance(supplied_independent, list) else []
199 )
200 independent_names = tuple(
201 name
202 for name in independent_list
203 if isinstance(name, str) and name in reviewer_set
204 )
205 elif independent_reviewer(reviewer, source):
206 independent_names = (reviewer,)
207 else:
208 independent_names = ()
209 grouped.setdefault(fingerprint(finding), []).append(
210 (finding, reviewer, independent_names)
211 )
213 merged = [merge_group(group) for group in grouped.values()]
214 suppressed: Counter[str] = Counter()
215 suppressed_findings: list[dict[str, Any]] = []
216 survivors: list[dict[str, Any]] = []
217 pre_existing: list[dict[str, Any]] = []
218 for finding in merged:
219 if finding["pre_existing"]:
220 pre_existing.append(finding)
221 continue
222 if (
223 finding["confidence"] < 75
224 and finding["severity"] != "P0"
225 and not finding.get("settled_conflict")
226 ):
227 suppressed[str(finding["confidence"])] += 1
228 suppressed_findings.append(finding)
229 continue
230 survivors.append(finding)
232 suppressed_findings.sort(
233 key=lambda item: (
234 SEVERITIES.index(item["severity"]),
235 -item["confidence"],
236 item["file"].lower(),
237 str(item["line"]),
238 item["title"].lower(),
239 )
240 )
241 survivors.sort(
242 key=lambda item: (
243 SEVERITIES.index(item["severity"]),
244 -item["confidence"],
245 item["file"].lower(),
246 str(item["line"]),
247 item["title"].lower(),
248 )
249 )
250 for number, finding in enumerate(survivors, 1):
251 finding["#"] = number
253 print(
254 json.dumps(
255 {
256 "status": "complete",
257 "findings": survivors,
258 "suppressed_findings": suppressed_findings,
259 "pre_existing_findings": pre_existing,
260 "residual_risks": residual_risks,
261 "testing_gaps": testing_gaps,
262 "suppressed_by_confidence": dict(sorted(suppressed.items())),
263 "malformed_returns": malformed_returns,
264 "malformed_findings": malformed_findings,
265 "first_evidence_backfilled": first_evidence_backfilled,
266 },
267 sort_keys=True,
268 )
269 )
270 return 0
273if __name__ == "__main__":
274 raise SystemExit(main())