b00b4bb7f2
On Windows, subprocess text=True without an explicit encoding decodes child output with the ANSI code page (e.g. 'gbk'); non-ASCII bytes then raise UnicodeDecodeError inside subprocess._readerthread, killing the Hermes backend before it becomes ready and surfacing as the desktop boot timeout. Sweep every hermes_cli text=True subprocess call to encoding='utf-8', errors='replace', and add an AST-based regression test that fails when a future text-mode call omits the encoding. Fixes #55658
118 lines
6.8 KiB
Python
118 lines
6.8 KiB
Python
"""Exact-head GitHub acceptance for explicitly declared PR tasks.
|
|
|
|
Network work happens outside SQLite transactions. The lifecycle owner persists
|
|
receipts only after rechecking the captured run/status/contract under its lock.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import subprocess
|
|
from urllib.parse import quote
|
|
|
|
_REPO = re.compile(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+")
|
|
_PR = re.compile(r"https://github\.com/([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)/pull/([1-9][0-9]*)")
|
|
|
|
|
|
def validate_contract(value: str | None) -> str:
|
|
if value is None or value == "local-only":
|
|
return "local-only"
|
|
if not isinstance(value, str) or not (_REPO.fullmatch(value) or _PR.fullmatch(value)):
|
|
raise ValueError("completion_contract must be local-only, OWNER/REPO, or an exact GitHub PR URL")
|
|
return value
|
|
|
|
|
|
def _api(endpoint: str, *, query: str | None = None, paginate: bool = False):
|
|
command = ["gh", "api", endpoint, "--hostname", "github.com"]
|
|
if query is not None:
|
|
command += ["-f", "query=" + query]
|
|
if paginate:
|
|
command += ["--paginate", "--slurp"]
|
|
result = subprocess.run(command, stdin=subprocess.DEVNULL, capture_output=True,
|
|
text=True, encoding="utf-8", errors="replace", timeout=30, check=True)
|
|
value = json.loads(result.stdout)
|
|
if isinstance(value, dict) and value.get("errors"):
|
|
raise ValueError("GitHub returned incomplete GraphQL evidence")
|
|
return value
|
|
|
|
|
|
def collect_acceptance(contract: str, published_pr: str | None) -> dict:
|
|
receipt = {"ok": False, "classification": "missing", "head_sha": None,
|
|
"pr_url": published_pr, "checks": [],
|
|
"recovery": "Fix required failures, rerun infrastructure checks or wait, then retry completion. "
|
|
"Use kanban_block if human input is needed; receipts remain on the task event log."}
|
|
try:
|
|
declared = _PR.fullmatch(contract)
|
|
url = contract if declared else published_pr
|
|
match = _PR.fullmatch(url or "")
|
|
if not match or (not declared and match[1] != contract) or (declared and published_pr and published_pr != contract):
|
|
receipt["detail"] = "Supply metadata.published_pr matching the persisted completion contract."
|
|
return receipt
|
|
repo, number = match[1], int(match[2])
|
|
receipt["pr_url"] = url
|
|
owner, name = repo.split("/")
|
|
query = '''{repository(owner:%s,name:%s){pullRequest(number:%d){headRefOid baseRefName state
|
|
baseRef{branchProtectionRule{requiredStatusChecks{context app{databaseId}}}}}}}''' % (
|
|
json.dumps(owner), json.dumps(name), number)
|
|
pr = _api("graphql", query=query)["data"]["repository"]["pullRequest"]
|
|
sha, branch = pr["headRefOid"], pr["baseRefName"]
|
|
receipt["head_sha"] = sha
|
|
if not re.fullmatch(r"[0-9a-f]{40}", sha) or pr["state"] not in {"OPEN", "MERGED"}:
|
|
raise ValueError("PR is closed or current head is unavailable")
|
|
protection = (pr.get("baseRef") or {}).get("branchProtectionRule") or {}
|
|
required = {(r["context"], (r.get("app") or {}).get("databaseId")) for r in protection.get("requiredStatusChecks", [])}
|
|
rules = _api(f"repos/{repo}/rules/branches/{quote(branch, safe='')}?per_page=100", paginate=True)
|
|
for page in rules:
|
|
for rule in page:
|
|
if rule["type"] == "required_status_checks":
|
|
required.update((r["context"], r.get("integration_id"))
|
|
for r in rule["parameters"]["required_status_checks"])
|
|
receipt["required"] = [{"context": c, "app_id": a} for c, a in sorted(required, key=str)]
|
|
if not required:
|
|
receipt["detail"] = "No repository-required checks are configured; explicitly use a local-only contract for non-CI tasks."
|
|
return receipt
|
|
pages = _api(f"repos/{repo}/commits/{sha}/check-runs?per_page=100&filter=latest", paginate=True)
|
|
runs = [run for page in pages for run in page["check_runs"]]
|
|
if len({r["id"] for r in runs}) != pages[0]["total_count"]:
|
|
raise ValueError("Incomplete check-run pagination")
|
|
statuses = [{**s, "sha": sha} for page in _api(f"repos/{repo}/commits/{sha}/statuses?per_page=100", paginate=True) for s in page]
|
|
outcomes = []
|
|
for context, app_id in sorted(required, key=str):
|
|
matching = [r for r in runs if r["name"] == context and
|
|
(app_id in (None, -1) or r["app"]["id"] == app_id)]
|
|
# A legacy status can satisfy an unpinned context, but never a check pinned to an app.
|
|
legacy = [s for s in statuses if s["context"] == context] if app_id in (None, -1) else []
|
|
selected = matching + ([max(legacy, key=lambda s: s["id"])] if legacy else [])
|
|
if not selected:
|
|
outcomes.append("missing")
|
|
receipt["checks"].append({"name": context, "classification": "missing", "head_sha": sha})
|
|
for check in selected:
|
|
is_run = "conclusion" in check
|
|
outcome = check.get("conclusion") if is_run else check["state"]
|
|
classification = _classify(check, sha, outcome, is_run)
|
|
outcomes.append(classification)
|
|
receipt["checks"].append({"name": context, "id": check["id"],
|
|
"url": check.get("html_url") or check.get("target_url"),
|
|
"head_sha": check.get("head_sha", check.get("sha")),
|
|
"classification": classification, "conclusion": outcome})
|
|
# Re-read after all pages: old-head successes are never transferable.
|
|
current = _api(f"repos/{repo}/pulls/{number}")
|
|
if current["head"]["sha"] != sha or current["base"]["ref"] != branch or (current["state"] == "closed" and not current.get("merged")):
|
|
receipt.update(classification="stale", detail="PR head/base changed while collecting evidence; retry.")
|
|
return receipt
|
|
receipt["classification"] = next((x for x in outcomes if x != "success"), "missing" if not outcomes else "success")
|
|
receipt["ok"] = receipt["classification"] == "success"
|
|
return receipt
|
|
except (OSError, subprocess.SubprocessError, ValueError, KeyError, TypeError, IndexError):
|
|
# Never persist gh stderr (credentials/host details); the failed phase is actionable.
|
|
receipt.update(classification="infra", detail="GitHub acceptance evidence unavailable or incomplete; check gh authentication/API access and retry.")
|
|
return receipt
|
|
|
|
|
|
def _classify(check: dict, sha: str, outcome: str | None, is_run: bool) -> str:
|
|
if check.get("head_sha", check.get("sha")) != sha:
|
|
return "stale"
|
|
if is_run and check.get("status") != "completed":
|
|
return "pending"
|
|
return {"success": "success", "failure": "failure", "error": "infra", "pending": "pending"}.get(outcome, "infra")
|