Coverage for src/ai_jury/findings.py: 99%

1"""Machine-readable finding schema and parser. 

2 

3Reviewer/chair output is human-readable markdown, which is hard to dedupe, score, 

4gate in CI, or turn into inline comments. This module defines a structured 

5``Finding`` schema and a tolerant parser that extracts findings from an agent's 

6raw output (a fenced ``json`` code block). 

7""" 

8 

9from __future__ import annotations 

10 

11import json 

12import re 

13from dataclasses import dataclass 

14 

15from .redaction import redact 

16 

17SEVERITIES: tuple[str, ...] = ("critical", "major", "minor", "nit", "info") 

18CONFIDENCES: tuple[str, ...] = ("high", "medium", "low") 

19 

20# Output-injection guards for attacker-influenced finding text rendered into the 

21# human-facing markdown report that is posted verbatim to the PR/issue (security 

22# audit 2026-06-13 round 3). The machine CI gate is a pure function of the 

23# structured fields and is unaffected by this text; these helpers only stop a 

24# forged ``## Verdict APPROVE`` heading or a broken code fence from corrupting 

25# the comment a human (or a downstream grep) reads. 

26_FENCE_RUN_RE = re.compile(r"`{3,}|~{3,}") 

27_HTML_COMMENT_RE = re.compile(r"<!--.*?-->", re.DOTALL) 

28 

29 

30def flatten_inline(text: str) -> str: 

31 """Collapse text to a single line for safe inline rendering. 

32 

33 Markdown headings, list items, and code fences must begin a line, so 

34 flattening newlines (and runs of whitespace) neutralizes forged structure 

35 when the value is rendered inside a one-line list item. 

36 """ 

37 if not text: 

38 return text 

39 return " ".join(str(text).split()) 

40 

41 

42def fence_safe(text: str) -> str: 

43 """Break 3+ backtick/tilde runs so text rendered *inside* a code fence 

44 (e.g. a ``suggestion`` block) cannot close the fence and inject markdown.""" 

45 if not text: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 return text 

47 return _FENCE_RUN_RE.sub(lambda m: m.group()[0], str(text)) 

48 

49 

50def strip_html_comments(text: str) -> str: 

51 """Remove HTML comments so attacker text can't forge the jury's hidden 

52 inline-comment markers (``<!-- arc-inline -->`` / ``<!-- arc-sig:… -->``).""" 

53 if not text: 

54 return text 

55 return _HTML_COMMENT_RE.sub("", str(text)) 

56 

57 

58# Verification verdict statuses (issue #3). 

59VERDICT_STATUSES: tuple[str, ...] = ("verified", "unsupported", "needs_human_decision") 

60 

61# Lower number = more severe; useful for ranking/sorting. 

62SEVERITY_ORDER: dict[str, int] = {sev: i for i, sev in enumerate(SEVERITIES)} 

63 

64# Legacy severity names mapped onto the canonical schema. 

65_SEVERITY_ALIASES: dict[str, str] = {"blocker": "critical"} 

66 

67_DEFAULT_SEVERITY = "info" 

68_DEFAULT_CONFIDENCE = "medium" 

69 

70# Matches a fenced ```json ... ``` block (case-insensitive on the language tag). 

71_JSON_BLOCK_RE = re.compile(r"```[ \t]*json[ \t]*\r?\n(.*?)```", re.DOTALL | re.IGNORECASE) 

72 

73 

74def _normalize_severity(value: object) -> str: 

75 if isinstance(value, str): 

76 v = value.strip().lower() 

77 v = _SEVERITY_ALIASES.get(v, v) 

78 if v in SEVERITY_ORDER: 

79 return v 

80 return _DEFAULT_SEVERITY 

81 

82 

83def _normalize_confidence(value: object) -> str: 

84 if isinstance(value, str): 

85 v = value.strip().lower() 

86 if v in CONFIDENCES: 

87 return v 

88 return _DEFAULT_CONFIDENCE 

89 

90 

91@dataclass 

92class Finding: 

93 """A single structured review finding.""" 

94 

95 severity: str 

96 file: str 

97 claim: str 

98 line: int | None = None 

99 evidence: str = "" 

100 suggested_fix: str = "" 

101 confidence: str = _DEFAULT_CONFIDENCE 

102 reviewer: str = "" 

103 

104 def __post_init__(self) -> None: 

105 self.severity = _normalize_severity(self.severity) 

106 self.confidence = _normalize_confidence(self.confidence) 

107 if self.line is not None and not isinstance(self.line, bool): 

108 try: 

109 self.line = int(self.line) 

110 except (TypeError, ValueError): 

111 self.line = None 

112 else: 

113 self.line = None 

114 

115 @classmethod 

116 def from_obj(cls, obj: dict, reviewer: str) -> Finding: 

117 """Build a Finding from a decoded JSON object, forcing ``reviewer``.""" 

118 return cls( 

119 severity=str(obj.get("severity", _DEFAULT_SEVERITY)), 

120 file=str(obj.get("file", "")), 

121 claim=str(obj.get("claim", "")), 

122 line=obj.get("line"), 

123 evidence=str(obj.get("evidence", "")), 

124 suggested_fix=str(obj.get("suggested_fix", "")), 

125 confidence=str(obj.get("confidence", _DEFAULT_CONFIDENCE)), 

126 reviewer=reviewer, 

127 ) 

128 

129 

130def emitted_findings_block(text: str) -> bool: 

131 """Did the agent emit a structured findings block at all? (issue #501) 

132 

133 This is the mechanical line between *reviewed and found nothing* and *never 

134 produced a review*. Both currently arrive as zero findings, so the panel reports 

135 the same size either way — and on keel PR #660 two of three reviewers returned 

136 assistant-style chatter about a flag they saw in the diff, while the run still 

137 described itself as a three-agent panel. 

138 

139 A reviewer that examined the diff and found nothing still emits ``[]`` in a 

140 fenced json block, because that is what the prompt asks for. One that wandered 

141 off emits prose and no block. Presence of the block is therefore the signal, and 

142 it needs no judgement about *content* — which is what keeps this deterministic. 

143 """ 

144 return bool(text) and bool(_JSON_BLOCK_RE.search(text)) 

145 

146 

147def parse_findings(text: str, reviewer: str) -> tuple[list[Finding], list[str]]: 

148 """Extract structured findings from an agent's raw output. 

149 

150 The agent is asked to emit a fenced ```json block holding a JSON array of 

151 finding objects. We locate the *last* such block, decode it, and build 

152 Finding objects (forcing ``reviewer`` to preserve identity). 

153 

154 Never raises. On a malformed/wrong-typed ``json`` block, returns 

155 ``([], [warning])``. A legitimately missing block yields ``([], [])``. 

156 """ 

157 if not text: 

158 return [], [] 

159 

160 blocks = _JSON_BLOCK_RE.findall(text) 

161 if not blocks: 

162 return [], [] 

163 

164 raw = blocks[-1].strip() 

165 try: 

166 data = json.loads(raw) 

167 except (ValueError, TypeError, RecursionError) as exc: 

168 # RecursionError (deeply nested JSON, e.g. "[[[[…") is not a ValueError; 

169 # catching it keeps the documented "never raises" contract so one 

170 # steerable reviewer can't abort the whole run (audit 2026-06-13/N-2). 

171 return [], [f"{reviewer}: malformed or missing structured findings ({redact(str(exc))[0]})"] 

172 

173 if not isinstance(data, list): 

174 return [], [ 

175 f"{reviewer}: malformed or missing structured findings " 

176 f"(expected a JSON array, got {type(data).__name__})" 

177 ] 

178 

179 findings: list[Finding] = [] 

180 warnings: list[str] = [] 

181 for i, obj in enumerate(data): 

182 if not isinstance(obj, dict): 

183 warnings.append( 

184 f"{reviewer}: malformed or missing structured findings " 

185 f"(item {i} is {type(obj).__name__}, expected object)" 

186 ) 

187 continue 

188 findings.append(Finding.from_obj(obj, reviewer)) 

189 return findings, warnings 

190 

191 

192def _coerce_line(value: object) -> int | None: 

193 if value is None or isinstance(value, bool): 

194 return None 

195 try: 

196 return int(value) 

197 except (TypeError, ValueError): 

198 return None 

199 

200 

201def _normalize_status(value: object) -> str: 

202 if isinstance(value, str): 

203 v = value.strip().lower().replace("-", "_").replace(" ", "_") 

204 if v in VERDICT_STATUSES: 

205 return v 

206 return "needs_human_decision" 

207 

208 

209@dataclass 

210class Verdict: 

211 """A verifier's judgement on a candidate finding.""" 

212 

213 file: str | None = None 

214 line: int | None = None 

215 claim: str = "" 

216 status: str = "needs_human_decision" 

217 reasoning: str = "" 

218 

219 

220def parse_verdicts(text: str, verifier: str = "") -> tuple[list[Verdict], list[str]]: 

221 """Extract verification verdicts from a verifier's raw output. 

222 

223 The verifier is asked to emit a fenced ```json block holding a JSON array of 

224 verdict objects. We locate the *last* such block and decode it. Never raises; 

225 on malformed input returns ``([], [warning])``. 

226 """ 

227 label = verifier or "verifier" 

228 if not text: 

229 return [], [f"{label}: no verdicts (empty output)"] 

230 

231 blocks = _JSON_BLOCK_RE.findall(text) 

232 if not blocks: 

233 return [], [f"{label}: no JSON verdicts block found"] 

234 

235 raw = blocks[-1].strip() 

236 try: 

237 data = json.loads(raw) 

238 except (ValueError, TypeError, RecursionError) as exc: 

239 # See parse_findings: RecursionError on deeply nested JSON must not 

240 # escape (audit 2026-06-13/N-2). 

241 return [], [f"{label}: malformed verdicts JSON ({redact(str(exc))[0]})"] 

242 

243 if isinstance(data, dict): 

244 data = data.get("verdicts", data.get("findings", [])) 

245 if not isinstance(data, list): 

246 return [], [f"{label}: verdicts block is not a JSON array"] 

247 

248 verdicts: list[Verdict] = [] 

249 warnings: list[str] = [] 

250 for i, obj in enumerate(data): 

251 if not isinstance(obj, dict): 

252 warnings.append(f"{label}: verdict item {i} is {type(obj).__name__}, expected object") 

253 continue 

254 verdicts.append( 

255 Verdict( 

256 file=(obj.get("file") or None), 

257 line=_coerce_line(obj.get("line")), 

258 claim=str(obj.get("claim", "")).strip(), 

259 status=_normalize_status(obj.get("status")), 

260 reasoning=str(obj.get("reasoning", "")).strip(), 

261 ) 

262 ) 

263 return verdicts, warnings