Coverage for core/langgraph/tools/github_pr_comment.py: 97%
139 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 00:49 +0000
1"""Per-tool module for the `github-pr-comment` deterministic builtin.
3Owns the tool's model-facing definition, its instruction surface (none), the
4PR-review-only helpers (findings lookup/validation, WIP title sync) and the
5review execution itself. The definition and the execution here are the single
6source; the legacy builtin site delegates to the pure `github_pr_comment`
7handler.
8"""
9from __future__ import annotations
11import json
12import logging
13import os
14from typing import Any
16import httpx
18from .contract import ToolContext, ToolResult
20try:
21 from tools._ticket_common import (
22 _hard_failure, _http_get, _http_patch, _http_post, _tool_content,
23 )
24except ImportError: # pragma: no cover - main.py run (qualified imports)
25 from langgraph.tools._ticket_common import (
26 _hard_failure, _http_get, _http_patch, _http_post, _tool_content,
27 )
29# Legacy logger name kept so caplog assertions in test_builtins_pr keep
30# capturing the WIP-sync info records after the handler moved here.
31log = logging.getLogger("_builtins_pr")
33GITHUB_PR_COMMENT_TOOL_NAME = "github-pr-comment"
34DEFAULT_STEP_ID = "github_pr_comment"
36GITHUB_PR_COMMENT_DEFINITION: dict[str, Any] = {
37 "type": "function",
38 "function": {
39 "name": GITHUB_PR_COMMENT_TOOL_NAME,
40 "description": (
41 "Post the pull-request review with `github-pr-comment` instead of "
42 "calling the forge API yourself. Set `findings_key` (default "
43 "`review_output`) to the state dot-path holding the findings and "
44 "`fail_on_severity` (default `Critical`) to the severity that "
45 "forces `REQUEST_CHANGES`; the result reports `findings_count` and the "
46 "WIP title sync status."
47 ),
48 "parameters": {
49 "type": "object",
50 "properties": {
51 "findings_key": {
52 "type": "string",
53 "description": (
54 "State dot-path holding the findings (default "
55 "`review_output`)."
56 ),
57 },
58 "fail_on_severity": {
59 "type": "string",
60 "description": (
61 "Severity that forces `REQUEST_CHANGES` (default "
62 "`Critical`)."
63 ),
64 },
65 },
66 "additionalProperties": False,
67 },
68 },
69}
71_PARAM_KEYS = ("findings_key", "fail_on_severity")
72_OUTPUT_KEYS = ("review_posted", "findings_count", "wip", "wip_error")
75def _hard_failure_type() -> type[Exception]:
76 """The runner's hard-failure type, resolved without an import cycle."""
77 try:
78 from runner import RunHardFailure
79 except ImportError: # pragma: no cover - main.py run (qualified imports)
80 from langgraph.runner import RunHardFailure
81 return RunHardFailure
84def _require_env(name: str) -> str:
85 value = os.environ.get(name)
86 if not value:
87 raise _hard_failure(
88 "VALIDATION_ERROR",
89 f"github-pr-comment: required env var {name!r} is not set",
90 )
91 return value
94def _validate_findings(findings: list) -> None:
95 for index, finding in enumerate(findings):
96 if not isinstance(finding, dict):
97 raise _hard_failure(
98 "VALIDATION_ERROR",
99 f"github-pr-comment: finding #{index} is not an object "
100 f"(got {type(finding).__name__})",
101 )
104def _is_success_status(status_code: int) -> bool:
105 return 200 <= status_code < 300
108def _raise_for_http_status(resp: httpx.Response, error_label: str) -> None:
109 """Convert a non-2xx external response into a RESOURCE_ERROR."""
110 if not _is_success_status(resp.status_code):
111 raise _hard_failure(
112 "RESOURCE_ERROR",
113 f"{error_label} failed ({resp.status_code}): {resp.text[:500]}",
114 )
117def _get_findings(state: dict, findings_key: str) -> list:
118 """Read findings from state by dot-path key (e.g. review.review_output).
120 A missing path segment or a non-list value yields an empty list (manifest
121 policy).
122 """
123 cur: Any = state
124 for part in str(findings_key).split("."):
125 if isinstance(cur, dict):
126 cur = cur.get(part)
127 else:
128 return []
129 return cur if isinstance(cur, list) else []
132_WIP_PREFIXES = ("WIP: ",)
135def _strip_wip_prefix(title: str) -> tuple[str, bool]:
136 lower = title.lower()
137 for prefix in _WIP_PREFIXES:
138 if lower.startswith(prefix.lower()):
139 return title[len(prefix):], True
140 return title, False
143def _wip_sync_failure_reason(status_code: int) -> str:
144 if status_code == 401:
145 return "authentication failed — check bot token"
146 if status_code == 403:
147 return "forbidden — bot lacks write access to the repository (read-only bot)"
148 if status_code == 404:
149 return "pull request not found (closed or deleted?)"
150 if 500 <= status_code <= 599:
151 return "Gitea server error"
152 return "unexpected status"
155def github_pr_comment(step: dict, state: dict) -> dict:
156 """Post PR review (APPROVED / REQUEST_CHANGES) via Gitea/GitHub REST API.
158 Findings at or above `fail_on_severity` -> event `REQUEST_CHANGES`,
159 otherwise `APPROVED`. Config/profile/malformed findings raise
160 RunHardFailure(VALIDATION_ERROR); external HTTP, transport, and
161 malformed-response failures raise RunHardFailure(RESOURCE_ERROR).
162 After a successful review the WIP marker in the PR title
163 is synced: REQUEST_CHANGES -> prepend "WIP: "; APPROVED -> strip the
164 "WIP: " prefix. The WIP sync is best-effort: a non-2xx title PATCH
165 emits an info-level log entry with a cause classified by status
166 code (401 authentication, 403 missing repo write access, 404 PR
167 gone, 5xx server error) and the run continues.
168 No commit status is set; requiring
169 human approval is a repo admin concern (branch protection).
171 Env vars used:
172 RUN_PROFILE - must be "PR" (guard; anything else raises)
173 GITEA_BASE_URL - Gitea instance URL
174 GIT_TOKEN - personal access token
175 PR_NUMBER - pull request number
176 REPO_FULL_NAME - repository full name (owner/repo)
177 """
178 run_profile = os.environ.get("RUN_PROFILE")
179 if run_profile != "PR":
180 raise _hard_failure(
181 "VALIDATION_ERROR",
182 f"github-pr-comment requires RUN_PROFILE=PR (got {run_profile!r})",
183 )
185 params = step.get("params", {}) or {}
186 findings_key = params.get("findings_key", "review_output")
187 fail_on_severity = params.get("fail_on_severity", "Critical")
189 base_url = _require_env("GITEA_BASE_URL").rstrip("/")
190 token = _require_env("GIT_TOKEN")
191 pr_number = _require_env("PR_NUMBER")
192 repo_full_name = _require_env("REPO_FULL_NAME")
194 findings = _get_findings(state, findings_key)
195 _validate_findings(findings)
196 headers = {"Authorization": f"token {token}", "Content-Type": "application/json"}
198 # PR review: standard Gitea/GitHub review, no commit status
199 has_blocking = any(f.get("severity") == fail_on_severity for f in findings)
200 event = "REQUEST_CHANGES" if has_blocking else "APPROVED"
202 body = _format_findings_as_markdown(findings)
203 review_label = "github-pr-comment: review POST"
204 review_resp = _http_post(
205 f"{base_url}/api/v1/repos/{repo_full_name}/pulls/{pr_number}/reviews",
206 headers=headers,
207 payload={"body": body, "event": event},
208 error_label=review_label,
209 )
210 _raise_for_http_status(review_resp, review_label)
212 pr_label = "github-pr-comment: PR GET"
213 pr_resp = _http_get(
214 f"{base_url}/api/v1/repos/{repo_full_name}/pulls/{pr_number}",
215 headers=headers,
216 error_label=pr_label,
217 )
218 _raise_for_http_status(pr_resp, pr_label)
219 try:
220 pr_data = pr_resp.json()
221 except json.JSONDecodeError:
222 raise _hard_failure(
223 "RESOURCE_ERROR",
224 f"{pr_label} returned invalid JSON: {pr_resp.text[:500]}",
225 ) from None
226 title = pr_data.get("title") if isinstance(pr_data, dict) else None
227 if not isinstance(title, str):
228 raise _hard_failure(
229 "RESOURCE_ERROR",
230 f"{pr_label} missing title: {pr_resp.text[:500]}",
231 )
233 stripped, has_wip_prefix = _strip_wip_prefix(title)
234 if event == "REQUEST_CHANGES":
235 new_title = None if has_wip_prefix else f"WIP: {title}"
236 else:
237 new_title = stripped if has_wip_prefix else None
239 wip = "unchanged"
240 wip_error = None
241 if new_title is not None:
242 patch_resp = _http_patch(
243 f"{base_url}/api/v1/repos/{repo_full_name}/pulls/{pr_number}",
244 headers=headers,
245 payload={"title": new_title},
246 error_label="github-pr-comment: WIP title PATCH",
247 )
248 if not _is_success_status(patch_resp.status_code):
249 log.info(
250 "github-pr-comment: WIP title sync failed (%s): %s — %s",
251 patch_resp.status_code,
252 patch_resp.text[:500],
253 _wip_sync_failure_reason(patch_resp.status_code),
254 )
255 wip = "sync_failed"
256 wip_error = f"{patch_resp.status_code}: {patch_resp.text[:500]}"
257 else:
258 wip = "set" if event == "REQUEST_CHANGES" else "cleared"
260 result = {**state, "review_posted": True, "findings_count": len(findings), "wip": wip}
261 if wip_error is not None:
262 result["wip_error"] = wip_error
263 else:
264 result.pop("wip_error", None)
265 return result
268def _format_findings_as_markdown(findings: list) -> str:
269 if not findings:
270 return "## PR Review\n\nNo findings."
271 _validate_findings(findings)
272 lines = ["## PR Review\n"]
273 for f in findings:
274 severity = f.get("severity", "Info")
275 finding = f.get("finding", f.get("message", ""))
276 evidence = f.get("evidence_ref", f.get("evidence", ""))
277 lines.append(f"- **{severity}**: {finding}")
278 if evidence:
279 lines.append(f" - Evidence: {evidence}")
280 lines.append(f"\n_{len(findings)} finding(s)_")
281 return "\n".join(lines)
284class _GitHubPrCommentTool:
285 name = GITHUB_PR_COMMENT_TOOL_NAME
287 def definition(self) -> dict[str, Any]:
288 return GITHUB_PR_COMMENT_DEFINITION
290 def instruction(self) -> str | None:
291 return None
293 def params(self) -> list[dict[str, Any]] | None:
294 return None
296 def execute(self, context: ToolContext) -> ToolResult:
297 step_id = (context.step or {}).get("id") or DEFAULT_STEP_ID
298 args = context.args or {}
299 params = {key: args[key] for key in _PARAM_KEYS if key in args}
300 synthetic_step = {"id": step_id, "params": params}
301 try:
302 new_state = github_pr_comment(synthetic_step, context.state or {})
303 except _hard_failure_type() as exc:
304 return ToolResult(
305 ok=False,
306 content=_tool_content(GITHUB_PR_COMMENT_TOOL_NAME, ok=False, error=exc.message),
307 error=exc.message,
308 )
309 output = {key: new_state[key] for key in _OUTPUT_KEYS if key in new_state}
310 return ToolResult(
311 ok=True,
312 content=_tool_content(GITHUB_PR_COMMENT_TOOL_NAME, ok=True, output=output),
313 state_patch=output,
314 )
317TOOL = _GitHubPrCommentTool()