Coverage for src/ai_jury/classification.py: 99%
111 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-30 06:29 +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 ( 138 ↛ 144line 138 didn't jump to line 144 because the condition on line 138 was never true
139 diff.startswith("+")
140 and not diff.startswith("+++")
141 or diff.startswith("-")
142 and not diff.startswith("---")
143 ):
144 c += 1
145 return c
148def _text_blob(finding: Any) -> str:
149 """Concatenate the human-text fields of a finding for keyword scanning."""
150 parts = [
151 getattr(finding, "claim", "") or "",
152 getattr(finding, "evidence", "") or "",
153 getattr(finding, "suggested_fix", "") or "",
154 getattr(finding, "file", "") or "",
155 getattr(finding, "reviewer", "") or "",
156 ]
157 return " ".join(parts)
160def is_security_finding(finding: Any) -> bool:
161 """True if a single finding looks security-related.
163 A finding is security-sensitive when EITHER its severity is ``critical`` OR
164 any :data:`SECURITY_KEYWORDS` token appears in its text fields. The
165 injection-scanner's synthetic finding (reviewer ``injection-scanner``,
166 claim mentioning "injection") is therefore caught by the keyword path.
167 """
168 if getattr(finding, "severity", "") == "critical":
169 return True
170 blob = _text_blob(finding)
171 return bool(_COMBINED_RX.search(blob))
174def _risk_level_from_stats(
175 has_critical: bool, has_major: bool, has_minor: bool, groups: list
176) -> str:
177 """Derive the risk level from precomputed severity stats.
179 Thresholds (deterministic):
180 * ``high`` — any ``critical`` finding, OR any ``major`` finding that is
181 part of a confirmed consensus group (consensus/majority bucket and not
182 rejected/unsupported).
183 * ``medium`` — any ``major`` finding (single-reviewer / unverified), OR any
184 ``minor`` finding.
185 * ``low`` — only ``nit`` / ``info`` findings, or no findings at all.
186 """
187 if has_critical:
188 return RISK_HIGH
190 if has_major:
191 # A confirmed (consensus/majority, not rejected) major finding is high
192 # risk; an isolated or rejected one is medium.
193 for g in groups:
194 if (
195 g.severity == "major"
196 and g.bucket in ("consensus", "majority")
197 and (getattr(g, "status", "") or "") != "unsupported"
198 ):
199 return RISK_HIGH
200 return RISK_MEDIUM
202 if has_minor:
203 return RISK_MEDIUM
205 return RISK_LOW
208def _review_effort_from_stats(n: int, most_severe: int, lines_changed: int) -> int:
209 """Map precomputed stats + diff size onto a 1-5 review-effort score (deterministic)."""
210 score = 1
212 if n >= 8:
213 score += 2
214 elif n >= 3:
215 score += 1
216 elif n >= 1:
217 score += 0 # presence is captured by the severity term below
219 if most_severe <= _severity_rank("major"):
220 score += 2
221 elif most_severe <= _severity_rank("minor"):
222 score += 1
224 if lines_changed > 400:
225 score += 2
226 elif lines_changed > 80:
227 score += 1
229 return max(1, min(5, score))
232def _has_unresolved_groups(groups: list) -> bool:
233 """True if any consensus group is disputed or needs a human decision."""
234 for g in groups:
235 if getattr(g, "bucket", "") in _UNRESOLVED_BUCKETS:
236 return True
237 if getattr(g, "status", "") in _UNRESOLVED_STATUSES:
238 return True
239 return False
242def classify(
243 outcome: Any = None,
244 *,
245 findings: Any = None,
246 groups: Any = None,
247 diff: str | None = None,
248) -> dict:
249 """Return the deterministic PR-level classification dict."""
250 fs = _resolved_findings(outcome, findings)
251 gs = _resolved_groups(outcome, groups)
252 lines_changed = diff_lines_changed(diff)
254 has_critical = False
255 has_major = False
256 has_minor = False
257 security = False
258 most_severe = 99
260 # bolt: single-pass iteration to collect finding statistics
261 for f in fs:
262 rank = _severity_rank(f.severity)
263 if rank < most_severe:
264 most_severe = rank
266 if f.severity == "critical":
267 has_critical = True
268 elif f.severity == "major":
269 has_major = True
270 elif f.severity == "minor":
271 has_minor = True
273 if not security and is_security_finding(f):
274 security = True
276 risk = _risk_level_from_stats(has_critical, has_major, has_minor, gs)
277 effort = _review_effort_from_stats(len(fs), most_severe, lines_changed)
278 needs_human = risk == RISK_HIGH or security or _has_unresolved_groups(gs)
280 return {
281 "review_effort": effort,
282 "risk_level": risk,
283 "security_sensitive": bool(security),
284 "needs_human_attention": bool(needs_human),
285 }
288def label_strings(classification: dict) -> list[str]:
289 """Derive GitHub label strings from a classification dict (deterministic).
291 Mirrors the labels suggested in issue #7, e.g.::
293 ["review effort: 3/5", "risk: high", "possible security issue",
294 "needs human attention"]
296 The security and human-attention labels are only emitted when their flag is
297 true. Order is stable.
298 """
299 labels = [
300 f"review effort: {classification['review_effort']}/5",
301 f"risk: {classification['risk_level']}",
302 ]
303 if classification.get("security_sensitive"):
304 labels.append("possible security issue")
305 if classification.get("needs_human_attention"):
306 labels.append("needs human attention")
307 return labels
310def summary_line(classification: dict) -> str:
311 """Render a compact one-line human summary of the classification."""
312 return (
313 f"review effort: {classification['review_effort']}/5"
314 f" · risk: {classification['risk_level']}"
315 f" · security-sensitive: {'yes' if classification['security_sensitive'] else 'no'}"
316 f" · needs human attention: "
317 f"{'yes' if classification['needs_human_attention'] else 'no'}"
318 )