Pick the best result.json from a dedup group for the report agent. Prefer passed-grade > rejected, then highest grade score, then smallest PoC (cleaner to analyze). Returns (result_path, result_dict, crash_dict). Unreadable entries are skipped; ValueError if nothing is readable.
(entries: list[tuple[Path, str, dict]])
| 1078 | |
| 1079 | |
| 1080 | def _pick_representative(entries: list[tuple[Path, str, dict]]) -> tuple[Path, dict, dict]: |
| 1081 | """Pick the best result.json from a dedup group for the report agent. |
| 1082 | |
| 1083 | Prefer passed-grade > rejected, then highest grade score, then smallest PoC |
| 1084 | (cleaner to analyze). Returns (result_path, result_dict, crash_dict). |
| 1085 | Unreadable entries are skipped; ValueError if nothing is readable. |
| 1086 | """ |
| 1087 | candidates: list[tuple[tuple[int, float, int, str], Path, dict, dict]] = [] |
| 1088 | for path, status, _reason in entries: |
| 1089 | try: |
| 1090 | r = json.loads(path.read_text()) |
| 1091 | except (OSError, json.JSONDecodeError): |
| 1092 | continue |
| 1093 | crash = r.get("crash") |
| 1094 | if not crash: |
| 1095 | continue |
| 1096 | score = (r.get("verdict") or {}).get("score") or 0.0 |
| 1097 | poc_len = len(crash.get("poc_bytes") or "") |
| 1098 | key = (_STATUS_ORDER.get(status, 2), -score, poc_len, str(path)) |
| 1099 | candidates.append((key, path, r, crash)) |
| 1100 | |
| 1101 | if not candidates: |
| 1102 | raise ValueError("no readable result.json in group") |
| 1103 | _k, path, result, crash = min(candidates, key=lambda c: c[0]) |
| 1104 | return path, result, crash |
| 1105 | |
| 1106 | |
| 1107 | async def _report_one( |
no outgoing calls