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

86 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-17 21:31 +0000

1"""Suggested-patch output for verified findings (issue #10). 

2 

3The jury identifies issues; this renders a *separate*, opt-in "suggested 

4patches" section that turns verified findings into concrete, inspectable fix 

5suggestions. It is deliberately conservative: 

6 

7- only VERIFIED findings (a consensus group the verifier confirmed) produce a 

8 suggestion — unverified or rejected findings never do; 

9- suggestions are rendered as clearly-labelled blocks tied to one finding; 

10- nothing is ever applied automatically (read-only by design); the output is for 

11 a human to inspect, copy, or adapt. 

12 

13Pure and deterministic: given the same groups it renders the same markdown. 

14""" 

15 

16from __future__ import annotations 

17 

18from dataclasses import dataclass 

19from pathlib import Path 

20 

21from .consensus import BUCKET_REJECTED, FindingGroup 

22from .findings import fence_safe, flatten_inline 

23from .redaction import redact 

24 

25 

26@dataclass 

27class PatchSuggestion: 

28 file: str 

29 line: int | None 

30 severity: str 

31 claim: str 

32 suggested_fix: str 

33 

34 def location(self) -> str: 

35 loc = self.file or "?" 

36 if self.line is not None: 

37 loc = f"{loc}:{self.line}" 

38 return loc 

39 

40 

41def patch_suggestions(groups: list[FindingGroup]) -> list[PatchSuggestion]: 

42 """Return one suggestion per VERIFIED group that carries a suggested fix. 

43 

44 A group qualifies only when the verifier marked it ``verified`` (not 

45 unsupported/disputed and not merely unverified) AND its representative 

46 finding has a non-empty ``suggested_fix``. Order follows the input group 

47 order (already severity-sorted by the consensus pass). 

48 """ 

49 out: list[PatchSuggestion] = [] 

50 for g in groups: 

51 if getattr(g, "status", "") != "verified" or g.bucket == BUCKET_REJECTED: 

52 continue 

53 rep = g.representative 

54 fix = (getattr(rep, "suggested_fix", "") or "").strip() 

55 if not rep or not fix: 

56 continue 

57 out.append( 

58 PatchSuggestion( 

59 file=rep.file or "", 

60 line=rep.line, 

61 severity=g.severity, 

62 claim=(rep.claim or "").strip(), 

63 suggested_fix=fix, 

64 ) 

65 ) 

66 return out 

67 

68 

69def render_patch_suggestions(groups: list[FindingGroup]) -> str: 

70 """Render the "Suggested patches" markdown section, or "" when there are none. 

71 

72 Kept separate from the default report so the standard review flow stays 

73 read-only; the CLI emits this only under ``--suggest-patches``. 

74 """ 

75 suggestions = patch_suggestions(groups) 

76 if not suggestions: 

77 return "" 

78 lines = [ 

79 "## Suggested patches", 

80 "", 

81 "_Opt-in, read-only suggestions for **verified** findings only. Inspect " 

82 "before applying — nothing here is applied automatically._", 

83 "", 

84 ] 

85 for s in suggestions: 

86 # Flatten the heading text and break any fence-closer inside the 

87 # suggestion body so attacker-influenced finding text can't inject a 

88 # forged verdict/heading into the posted comment (audit 2026-06-13 r3). 

89 lines.append( 

90 f"### {flatten_inline(s.location())} — [{s.severity}] {flatten_inline(s.claim)}" 

91 ) 

92 lines.append("") 

93 lines.append("> Verified by the jury.") 

94 lines.append("") 

95 lines.append("```suggestion") 

96 lines.append(fence_safe(s.suggested_fix)) 

97 lines.append("```") 

98 lines.append("") 

99 return "\n".join(lines).rstrip() + "\n" 

100 

101 

102def parse_patch_suggestions(text: str) -> list[PatchSuggestion]: 

103 """Parse PatchSuggestion objects from a markdown report or suggested-patches block.""" 

104 import re 

105 

106 out: list[PatchSuggestion] = [] 

107 # Pattern matches: ### file.py:123 — [severity] claim 

108 heading_re = re.compile(r"^###\s+([^—\n]+?)(?::(\d+))?\s+—\s+\[([^\]]+)\]\s+(.+)$", re.MULTILINE) 

109 suggestion_block_re = re.compile(r"```suggestion\n(.*?)\n```", re.DOTALL) 

110 

111 matches = list(heading_re.finditer(text)) 

112 for i, m in enumerate(matches): 

113 file_path = m.group(1).strip() 

114 line_num = int(m.group(2)) if m.group(2) else None 

115 severity = m.group(3).strip() 

116 claim = m.group(4).strip() 

117 

118 start = m.end() 

119 end = matches[i + 1].start() if i + 1 < len(matches) else len(text) 

120 sub_content = text[start:end] 

121 

122 fix_match = suggestion_block_re.search(sub_content) 

123 if fix_match: 123 ↛ 112line 123 didn't jump to line 112 because the condition on line 123 was always true

124 fix = fix_match.group(1).strip() 

125 out.append( 

126 PatchSuggestion( 

127 file=file_path, 

128 line=line_num, 

129 severity=severity, 

130 claim=claim, 

131 suggested_fix=fix, 

132 ) 

133 ) 

134 return out 

135 

136 

137def apply_patch_suggestion( 

138 suggestion: PatchSuggestion, root_dir: Path | None = None 

139) -> tuple[bool, str]: 

140 """Safely apply a patch suggestion to the targeted file.""" 

141 root = (root_dir or Path.cwd()).resolve() 

142 try: 

143 target = (root / suggestion.file).resolve() 

144 target.relative_to(root) 

145 except (ValueError, RuntimeError): 

146 return False, f"Path traversal rejected: {suggestion.file}" 

147 

148 if not target.exists() or not target.is_file(): 

149 return False, f"File not found: {suggestion.file}" 

150 

151 fix = suggestion.suggested_fix 

152 if fix.startswith("---") or "@@" in fix: 

153 import subprocess 

154 

155 proc = subprocess.run( 

156 ["git", "apply", "-"], input=fix, text=True, cwd=str(root), capture_output=True 

157 ) 

158 if proc.returncode == 0: 

159 return True, f"Applied git patch to {suggestion.file}" 

160 return False, f"Git apply failed: {redact(proc.stderr.strip())[0] or 'patch does not apply cleanly'}" 

161 

162 lines = target.read_text(encoding="utf-8").splitlines(keepends=True) 

163 if suggestion.line is not None and 1 <= suggestion.line <= len(lines): 

164 idx = suggestion.line - 1 

165 lines[idx] = fix + ("\n" if not fix.endswith("\n") else "") 

166 target.write_text("".join(lines), encoding="utf-8") 

167 return True, f"Applied line replacement at {suggestion.file}:{suggestion.line}" 

168 

169 return False, f"Cannot apply non-diff suggestion without line match in {suggestion.file}" 

170