Coverage for src/ai_jury/classification.py: 98%
113 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 23:18 +0000
1"""Deterministic PR-level classification derived from structured findings.
3The jury report already lists individual findings and consensus groups, but
4maintainers also want a compact, at-a-glance signal: how much review effort a PR
5needs, how risky it is, whether it touches security-sensitive code, and whether
6it warrants human attention. This module derives those four classifications as a
7PURE, fully deterministic function of the structured findings, the consensus
8groups, and (optionally) the unified diff.
10Nothing here calls an LLM or the network: identical inputs always produce
11identical output, which is what makes the classification safe to snapshot-test
12and to render in the deterministic mock report.
14Classifications
15---------------
16``review_effort`` : int, 1-5
17``risk_level`` : str, one of ``low`` / ``medium`` / ``high``
18``security_sensitive`` : bool
19``needs_human_attention`` : bool
21See :func:`classify` for the exact, documented formulas.
22"""
24from __future__ import annotations
26import re
27from typing import Any
29from .findings import SEVERITY_ORDER
31# Risk levels, ordered least to most severe.
32RISK_LOW = "low"
33RISK_MEDIUM = "medium"
34RISK_HIGH = "high"
36# Consensus buckets that mean "a human still needs to look at this": the verifier
37# could not confirm the finding, or flagged it as needing a human decision.
38_UNRESOLVED_BUCKETS = {"disputed"}
39_UNRESOLVED_STATUSES = {"needs_human_decision"}
41# Security keyword set. A finding is treated as security-sensitive if any of
42# these whole-word tokens (or multi-word phrases) appears in its claim, evidence,
43# suggested fix, or file path. Kept deliberately small and high-signal so benign
44# findings do not over-match. Matching is case-insensitive and word-boundary
45# anchored for single tokens (so "auth" does not fire inside "author").
46SECURITY_KEYWORDS: tuple[str, ...] = (
47 "injection",
48 "sql injection",
49 "xss",
50 "csrf",
51 "ssrf",
52 "rce",
53 "remote code execution",
54 "traversal",
55 "path traversal",
56 "directory traversal",
57 "secret",
58 "credential",
59 "password",
60 "token",
61 "api key",
62 "private key",
63 "auth",
64 "authentication",
65 "authorization",
66 "deserialization",
67 "sanitize",
68 "sanitization",
69 "escape",
70 "vulnerab",
71 "exploit",
72 "privilege",
73 "sandbox escape",
74)
76# Prefix stems: entries that should match any word starting with them (e.g.
77# "vulnerab" -> vulnerability/vulnerable/vulnerabilities; "exploit" ->
78# exploit/exploitable/exploited). Issue v1.5.0/L-2: these were anchored with a
79# trailing ``\b`` like full words, so ``\bvulnerab\b`` never matched
80# "vulnerability" (the ``\b`` fails before the following letter). Compile them
81# with a trailing ``\w*`` instead.
82_PREFIX_STEMS: frozenset[str] = frozenset({"vulnerab", "exploit"})
84# Pre-compiled, word-boundary anchored matchers for each keyword. Multi-word
85# phrases match on a relaxed boundary (spaces inside the phrase are literal).
86# Prefix stems use a trailing ``\w*`` so they match the whole word family.
87_KEYWORD_RES: tuple[re.Pattern[str], ...] = tuple(
88 re.compile(
89 r"\b" + re.escape(kw) + (r"\w*" if kw in _PREFIX_STEMS else r"\b"),
90 re.IGNORECASE,
91 )
92 for kw in SECURITY_KEYWORDS
93)
95# A single combined regex containing all security keyword patterns.
96# Evaluating one compound regex `(A|B|C)` in the C regex engine is ~4x faster
97# than iterating over 27 separate regexes in Python via `any()`.
98_COMBINED_RX = re.compile("|".join(rx.pattern for rx in _KEYWORD_RES), re.IGNORECASE)
101def _severity_rank(severity: str) -> int:
102 """Lower number = more severe (mirrors findings.SEVERITY_ORDER)."""
103 return SEVERITY_ORDER.get(severity, len(SEVERITY_ORDER))
106def _resolved_findings(outcome: Any, findings: Any) -> list:
107 """Pick the finding list to classify on.
109 Prefers an explicit ``findings`` argument, then ``outcome.findings``. The
110 list is returned as-is (callers pass already-aggregated findings).
111 """
112 if findings is not None:
113 return list(findings)
114 if outcome is not None and getattr(outcome, "findings", None) is not None:
115 return list(outcome.findings)
116 return []
119def _resolved_groups(outcome: Any, groups: Any) -> list:
120 if groups is not None:
121 return list(groups)
122 if outcome is not None and getattr(outcome, "groups", None) is not None:
123 return list(outcome.groups)
124 return []
127def diff_lines_changed(diff: str | None) -> int:
128 """Count added/removed lines in a unified diff (deterministic).
130 Counts lines beginning with a single ``+`` or ``-`` that are NOT part of the
131 file header (``+++`` / ``---``). Returns 0 for an empty or missing diff.
132 """
133 if not diff:
134 return 0
135 # bolt: avoid allocating a huge list of strings from splitlines()
136 # and generator overhead by using C-optimized string counting.
137 c = diff.count("\n+") + diff.count("\n-") - diff.count("\n+++") - diff.count("\n---")
138 if diff.startswith("+") and not diff.startswith("+++"): 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 c += 1
140 elif diff.startswith("-") and not diff.startswith("---"): 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true
141 c += 1
142 return c
145def _text_blob(finding: Any) -> str:
146 """Concatenate the human-text fields of a finding for keyword scanning."""
147 parts = [
148 getattr(finding, "claim", "") or "",
149 getattr(finding, "evidence", "") or "",
150 getattr(finding, "suggested_fix", "") or "",
151 getattr(finding, "file", "") or "",
152 getattr(finding, "reviewer", "") or "",
153 ]
154 return " ".join(parts)
157def is_security_finding(finding: Any) -> bool:
158 """True if a single finding looks security-related.
160 A finding is security-sensitive when EITHER its severity is ``critical`` OR
161 any :data:`SECURITY_KEYWORDS` token appears in its text fields. The
162 injection-scanner's synthetic finding (reviewer ``injection-scanner``,
163 claim mentioning "injection") is therefore caught by the keyword path.
164 """
165 if getattr(finding, "severity", "") == "critical":
166 return True
167 blob = _text_blob(finding)
168 return bool(_COMBINED_RX.search(blob))
171def _risk_level_from_stats(has_critical: bool, has_major: bool, has_minor: bool, groups: list) -> str:
172 """Derive the risk level from precomputed severity stats.
174 Thresholds (deterministic):
175 * ``high`` — any ``critical`` finding, OR any ``major`` finding that is
176 part of a confirmed consensus group (consensus/majority bucket and not
177 rejected/unsupported).
178 * ``medium`` — any ``major`` finding (single-reviewer / unverified), OR any
179 ``minor`` finding.
180 * ``low`` — only ``nit`` / ``info`` findings, or no findings at all.
181 """
182 if has_critical:
183 return RISK_HIGH
185 if has_major:
186 # A confirmed (consensus/majority, not rejected) major finding is high
187 # risk; an isolated or rejected one is medium.
188 for g in groups:
189 if (
190 g.severity == "major"
191 and g.bucket in ("consensus", "majority")
192 and (getattr(g, "status", "") or "") != "unsupported"
193 ):
194 return RISK_HIGH
195 return RISK_MEDIUM
197 if has_minor:
198 return RISK_MEDIUM
200 return RISK_LOW
203def _review_effort_from_stats(n: int, most_severe: int, lines_changed: int) -> int:
204 """Map precomputed stats + diff size onto a 1-5 review-effort score (deterministic)."""
205 score = 1
207 if n >= 8:
208 score += 2
209 elif n >= 3:
210 score += 1
211 elif n >= 1:
212 score += 0 # presence is captured by the severity term below
214 if most_severe <= _severity_rank("major"):
215 score += 2
216 elif most_severe <= _severity_rank("minor"):
217 score += 1
219 if lines_changed > 400:
220 score += 2
221 elif lines_changed > 80:
222 score += 1
224 return max(1, min(5, score))
227def _has_unresolved_groups(groups: list) -> bool:
228 """True if any consensus group is disputed or needs a human decision."""
229 for g in groups:
230 if getattr(g, "bucket", "") in _UNRESOLVED_BUCKETS:
231 return True
232 if getattr(g, "status", "") in _UNRESOLVED_STATUSES:
233 return True
234 return False
237def classify(
238 outcome: Any = None,
239 *,
240 findings: Any = None,
241 groups: Any = None,
242 diff: str | None = None,
243) -> dict:
244 """Return the deterministic PR-level classification dict."""
245 fs = _resolved_findings(outcome, findings)
246 gs = _resolved_groups(outcome, groups)
247 lines_changed = diff_lines_changed(diff)
249 has_critical = False
250 has_major = False
251 has_minor = False
252 security = False
253 most_severe = 99
255 # bolt: single-pass iteration to collect finding statistics
256 for f in fs:
257 rank = _severity_rank(f.severity)
258 if rank < most_severe:
259 most_severe = rank
261 if f.severity == "critical":
262 has_critical = True
263 elif f.severity == "major":
264 has_major = True
265 elif f.severity == "minor":
266 has_minor = True
268 if not security and is_security_finding(f):
269 security = True
271 risk = _risk_level_from_stats(has_critical, has_major, has_minor, gs)
272 effort = _review_effort_from_stats(len(fs), most_severe, lines_changed)
273 needs_human = risk == RISK_HIGH or security or _has_unresolved_groups(gs)
275 return {
276 "review_effort": effort,
277 "risk_level": risk,
278 "security_sensitive": bool(security),
279 "needs_human_attention": bool(needs_human),
280 }
283def label_strings(classification: dict) -> list[str]:
284 """Derive GitHub label strings from a classification dict (deterministic).
286 Mirrors the labels suggested in issue #7, e.g.::
288 ["review effort: 3/5", "risk: high", "possible security issue",
289 "needs human attention"]
291 The security and human-attention labels are only emitted when their flag is
292 true. Order is stable.
293 """
294 labels = [
295 f"review effort: {classification['review_effort']}/5",
296 f"risk: {classification['risk_level']}",
297 ]
298 if classification.get("security_sensitive"):
299 labels.append("possible security issue")
300 if classification.get("needs_human_attention"):
301 labels.append("needs human attention")
302 return labels
305def summary_line(classification: dict) -> str:
306 """Render a compact one-line human summary of the classification."""
307 return (
308 f"review effort: {classification['review_effort']}/5"
309 f" · risk: {classification['risk_level']}"
310 f" · security-sensitive: {'yes' if classification['security_sensitive'] else 'no'}"
311 f" · needs human attention: "
312 f"{'yes' if classification['needs_human_attention'] else 'no'}"
313 )