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
« 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.
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.
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.
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.
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.
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.
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]
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"""
49from __future__ import annotations
51import argparse
52import json
53import sys
54from pathlib import Path
55from typing import Any
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
71EXIT_OK = 0
72EXIT_CLI_FAILED = 1
73EXIT_NO_PLATFORM = 2
75# States that mean a referenced issue or request is no longer in play.
76_SETTLED = frozenset({"CLOSED", "MERGED", "closed", "merged"})
79def managed_paths(target_dir: Path) -> set[str]:
80 """The template-owned paths recorded in this repo's lock, or an empty set.
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))}
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.
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
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]
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.
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
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.
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": []}
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 )
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))
196 return {
197 "platform": platform,
198 "command": command,
199 "dry_run": False,
200 "state": state,
201 "notes": notes,
202 "issues": issues,
203 }
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)
230def _why(facts: dict[str, Any]) -> str:
231 """One line naming the signal that put an issue in its category.
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"
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)
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
298 print(json.dumps(report, indent=2) if args.json_output else render(report))
299 return EXIT_OK
302if __name__ == "__main__": # pragma: no cover
303 raise SystemExit(main())