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