Coverage for plugin/scripts/issue_status.py: 100%

127 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-10-26 04:21 +0000

1#!/usr/bin/env python3 

2"""List a repository's open issues and say which could be fixed without a judgement call. 

3 

4`/rhiza:quality` files findings as issues, those issues become branches, and the 

5branches become requests. The plugin has had the first step and the last for some time — 

6`quality` writes the issues, `remote` looks after the requests once they exist — and 

7nothing in between. This is what `/rhiza:fix` reads in order to close that gap. 

8 

9**It reads; it never branches, commits, pushes or comments.** Deciding what an issue is 

10worth and writing the fix is `/rhiza:fix`'s job, and needs judgement this cannot have. 

11The split is the same one `pr_status.py` draws against `/rhiza:remote`, for the same 

12reason. 

13 

14Reading issues does not belong in ``platform_cli.py``, which maps a *write* onto `gh` or 

15`glab` and says so in its own docstring. The two CLIs disagree about the **shape** of an 

16issue, not merely about flag names; ``_issue_forge`` owns that disagreement — the argv, 

17the subprocess and the per-platform keys — and this module sees issues only in the 

18shared vocabulary it normalises them into. 

19 

20**One asymmetry is reported rather than smoothed over.** GitHub shares a number space 

21between issues and pull requests, so `#86` in an issue body may be either and both are 

22reachable from one lookup. GitLab does not: `#86` is an issue and `!86` a merge request. 

23So the dependency states this resolves are exact on GitHub, and on GitLab a reference 

24that is not an issue comes back ``unresolved`` rather than guessed at. An unresolved 

25reference pushes an issue toward *reported* rather than *fixed*, so the weaker path 

26degrades safe. 

27 

28**It also reads the tracker's history, for `/rhiza:quality`.** A finding the user closed 

29last run — fixed, or declined as not worth doing — is not new because this run found it 

30again, and re-filing it is how a scoring command turns into noise. ``--state closed`` or 

31``--state all`` lists those too, each with its ``state`` and, on GitHub, the 

32``state_reason`` it was closed with (``completed``, ``not_planned``, ``duplicate``). 

33GitLab records no reason, so there it comes back empty rather than guessed. Only open 

34issues get the triage signals: they describe whether an issue *can be fixed*, which is a 

35question a closed one no longer asks. 

36 

37Usage: 

38 uv run --python 3.12 --no-project python \ 

39 scripts/issue_status.py [--target-dir DIR] [--issue N] [--limit N] 

40 [--label L] [--state open|closed|all] 

41 [--json] [--dry-run] 

42 

43Exit codes: 

44 0 the forge answered (including "no open issues"), or --dry-run rendered 

45 1 the platform CLI failed, is absent, or is not authenticated 

46 2 the platform could not be determined 

47""" 

48 

49from __future__ import annotations 

50 

51import argparse 

52import json 

53import sys 

54from pathlib import Path 

55from typing import Any 

56 

57sys.path.insert(0, str(Path(__file__).resolve().parent)) 

58from _issue_forge import ( # noqa: E402 

59 STATES, 

60 ForgeQueryError, 

61 build_issue_command, 

62 build_reference_commands, 

63 build_request_command, 

64 normalize, 

65 run_json, 

66) 

67from _issue_signals import signals # noqa: E402 

68from _rhiza_forge import PlatformError, detect_platform # noqa: E402 

69from _rhiza_lock import lock_path, previously_tracked # noqa: E402 

70 

71EXIT_OK = 0 

72EXIT_CLI_FAILED = 1 

73EXIT_NO_PLATFORM = 2 

74 

75# States that mean a referenced issue or request is no longer in play. 

76_SETTLED = frozenset({"CLOSED", "MERGED", "closed", "merged"}) 

77 

78 

79def managed_paths(target_dir: Path) -> set[str]: 

80 """The template-owned paths recorded in this repo's lock, or an empty set. 

81 

82 A repo that is not rhiza-managed has no lock and therefore no template-owned 

83 paths — which is the right answer rather than an error, because an unmanaged repo 

84 owns everything in it. ``previously_tracked`` already returns an empty set for a 

85 lock that is absent or unreadable, so there is no second check here. 

86 """ 

87 return {str(path) for path in previously_tracked(lock_path(target_dir, None))} 

88 

89 

90def resolve_references(platform: str, target_dir: Path, numbers: list[int]) -> dict[int, str]: 

91 """Map each of *numbers* to its state, or ``unresolved`` when it cannot be read. 

92 

93 One lookup per number, and only dependency references reach here — an issue that 

94 merely cites a merged PR is not sequenced behind it, and resolving every mention 

95 would turn a well-annotated issue into a stale one. 

96 """ 

97 states: dict[int, str] = {} 

98 for number in numbers: 

99 states[number] = "unresolved" 

100 for command in build_reference_commands(platform, number): 

101 try: 

102 payload = run_json(command, target_dir) 

103 except ForgeQueryError: 

104 continue 

105 state = str(payload.get("state") or "").strip() 

106 if state: 

107 states[number] = state 

108 break 

109 return states 

110 

111 

112def _requests_by_issue(requests: list[dict[str, Any]], issue_id: int) -> list[int]: 

113 """The open requests whose title or body mentions ``#issue_id``.""" 

114 token = f"#{issue_id}" 

115 hits: list[Any] = [] 

116 for request in requests: 

117 text = f"{request.get('title') or ''} {request.get('body') or ''}" 

118 if token in text: 

119 hits.append(request.get("number") or request.get("iid")) 

120 return [number for number in hits if number is not None] 

121 

122 

123def _enrich( 

124 issue: dict[str, Any], 

125 platform: str, 

126 target_dir: Path, 

127 requests: list[dict[str, Any]], 

128 owned: set[str], 

129) -> dict[str, Any]: 

130 """Attach the derived signals to one normalised *issue*, in place, and return it. 

131 

132 Two passes over :func:`signals` deliberately: the first reads the paths out of the 

133 body so this can resolve them against the tree and the lock, and the second folds 

134 those answers back in. The alternative — a signals call that takes a repository — 

135 would put the filesystem inside the one module that has no business touching it. 

136 """ 

137 body = str(issue["body"]) 

138 named = [str(path) for path in signals(body)["paths"]] 

139 missing = tuple(path for path in named if not (target_dir / path).exists()) 

140 template_owned = tuple(path for path in named if path in owned) 

141 open_requests = tuple(_requests_by_issue(requests, int(issue["id"]))) 

142 states = resolve_references(platform, target_dir, list(signals(body)["dependencies"])) 

143 superseded = tuple(number for number, state in states.items() if state in _SETTLED) 

144 issue["reference_states"] = {str(k): v for k, v in states.items()} 

145 issue["signals"] = signals( 

146 body, 

147 template_owned=template_owned, 

148 missing=missing, 

149 open_requests=open_requests, 

150 superseded_by=superseded, 

151 ) 

152 return issue 

153 

154 

155def collect( 

156 target_dir: Path, 

157 *, 

158 limit: int, 

159 labels: list[str], 

160 only: list[int], 

161 dry_run: bool, 

162 state: str = "open", 

163) -> dict[str, Any]: 

164 """Gather the issues in *state*, the open ones with derived signals and a category. 

165 

166 Raises: 

167 PlatformError: the hosting platform could not be determined. 

168 ForgeQueryError: a platform CLI was absent, failed, or was unauthenticated. 

169 """ 

170 platform = detect_platform(target_dir) 

171 command = build_issue_command(platform, limit=limit, labels=labels, state=state) 

172 if dry_run: 

173 return {"platform": platform, "command": command, "dry_run": True, "issues": []} 

174 

175 raw_issues = run_json(command, target_dir) 

176 requests = run_json(build_request_command(platform, limit=limit), target_dir) 

177 owned = managed_paths(target_dir) 

178 notes: list[str] = [] 

179 if platform == "gitlab": 

180 notes.append( 

181 "gitlab: `#N` names an issue, never a merge request — a reference to one " 

182 "comes back unresolved rather than guessed at" 

183 ) 

184 

185 issues: list[dict[str, Any]] = [] 

186 for raw in raw_issues: 

187 issue = normalize(platform, raw) 

188 if only and issue["id"] not in only: 

189 continue 

190 if issue["state"] == "closed": 

191 issue["signals"] = None 

192 issues.append(issue) 

193 continue 

194 issues.append(_enrich(issue, platform, target_dir, requests, owned)) 

195 

196 return { 

197 "platform": platform, 

198 "command": command, 

199 "dry_run": False, 

200 "state": state, 

201 "notes": notes, 

202 "issues": issues, 

203 } 

204 

205 

206def render(report: dict[str, Any]) -> str: 

207 """The report as lines for a human reading over the command's shoulder.""" 

208 if report.get("dry_run"): 

209 return " ".join(report["command"]) 

210 lines = [f"platform {report['platform']}"] 

211 for note in report.get("notes", []): 

212 lines.append(f"note {note}") 

213 if not report["issues"]: 

214 lines.append("no issues" if report.get("state", "open") != "open" else "no open issues") 

215 return "\n".join(lines) 

216 for issue in report["issues"]: 

217 facts = issue["signals"] 

218 if facts is None: 

219 reason = issue.get("state_reason") or "reason unrecorded" 

220 lines.append(f"{'closed':11} #{issue['id']} {issue['title']}") 

221 lines.append(f"{'':13}{reason}") 

222 continue 

223 lines.append(f"{facts['category']:11} #{issue['id']} {issue['title']}") 

224 lines.append(f"{'':13}{_why(facts)}") 

225 for caution in facts["cautions"]: 

226 lines.append(f"{'':13}caution: {caution}") 

227 return "\n".join(lines) 

228 

229 

230def _why(facts: dict[str, Any]) -> str: 

231 """One line naming the signal that put an issue in its category. 

232 

233 >>> _why({"category": "mechanical", "acceptance": "it works"}) 

234 'acceptance criterion, single-valued' 

235 >>> _why({"category": "decision", "acceptance": None, "decide_markers": []}) 

236 'no acceptance criterion — nobody has said what done means' 

237 """ 

238 category = facts["category"] 

239 if category == "blocked": 

240 return f"already has an open request: {facts.get('open_requests')}" 

241 if category == "stale": 

242 return f"sequenced behind {facts['superseded_by']}, which has settled" 

243 if category == "decision": 

244 if facts.get("acceptance") is None: 

245 return "no acceptance criterion — nobody has said what done means" 

246 return f"hands the reader a choice: {', '.join(facts['decide_markers'])}" 

247 if category == "optional": 

248 return "acceptance criterion admits more than one outcome" 

249 return "acceptance criterion, single-valued" 

250 

251 

252def main(argv: list[str] | None = None) -> int: 

253 """Parse arguments, gather the report, print it, and return an exit code.""" 

254 parser = argparse.ArgumentParser( 

255 description="List open issues and say which need no judgement call to fix." 

256 ) 

257 parser.add_argument("--target-dir", default=".", help="Repository root (default: cwd).") 

258 parser.add_argument( 

259 "--issue", 

260 action="append", 

261 type=int, 

262 default=[], 

263 help="Only this issue number; repeatable.", 

264 ) 

265 parser.add_argument("--limit", type=int, default=20, help="Maximum issues to fetch.") 

266 parser.add_argument( 

267 "--label", action="append", default=[], help="Only issues with this label; repeatable." 

268 ) 

269 parser.add_argument( 

270 "--state", 

271 choices=STATES, 

272 default="open", 

273 help="Which issues to list; closed ones carry no triage signals (default: open).", 

274 ) 

275 parser.add_argument( 

276 "--json", dest="json_output", action="store_true", help="Emit the report as JSON." 

277 ) 

278 parser.add_argument("--dry-run", action="store_true", help="Print the argv without running it.") 

279 args = parser.parse_args(argv) 

280 

281 target = Path(args.target_dir).resolve() 

282 try: 

283 report = collect( 

284 target, 

285 limit=args.limit, 

286 labels=args.label, 

287 only=args.issue, 

288 dry_run=args.dry_run, 

289 state=args.state, 

290 ) 

291 except PlatformError as exc: 

292 print(f"[ERROR] {exc}", file=sys.stderr) 

293 return EXIT_NO_PLATFORM 

294 except ForgeQueryError as exc: 

295 print(f"[ERROR] {exc}", file=sys.stderr) 

296 return EXIT_CLI_FAILED 

297 

298 print(json.dumps(report, indent=2) if args.json_output else render(report)) 

299 return EXIT_OK 

300 

301 

302if __name__ == "__main__": # pragma: no cover 

303 raise SystemExit(main())