9dabc57d9a
Some checks failed
CI / bootstrap (push) Successful in 9s
CI / lint (push) Successful in 21s
CI / typecheck (push) Successful in 25s
CI / complexity (push) Successful in 30s
CI / test (3.14) (push) Successful in 1m47s
CI / test (3.13) (push) Successful in 1m48s
CI / test (3.12) (push) Failing after 13m11s
789 lines
29 KiB
Python
789 lines
29 KiB
Python
#!/usr/bin/env python3
|
||
"""Pipeline-status oracle: determine PR phase in 6-phase PR pipeline.
|
||
|
||
Reads facts from GitHub (gh CLI), git, and memory files to deterministically
|
||
derive the current pipeline phase of a PR — no state file, like ``git status``
|
||
for the PR pipeline. CI gate via ``gh pr view --json statusCheckRollup`` (read-only):
|
||
CI ❌ blocks MERGE (transitive guard — first ❌ phase blocks all subsequent phases).
|
||
|
||
Usage:
|
||
python3 config/scripts/pipeline-status.py <PR_NUMBER> # single PR status
|
||
python3 config/scripts/pipeline-status.py # table of open PRs
|
||
|
||
Six phases:
|
||
1. ISSUE — GitHub issue exists and linked via Closes/Fixes #N
|
||
2. IMPLEMENT — PR exists + PR body has 4 required headings
|
||
3. CI — all checks on PR head SHA completed & success (statusCheckRollup)
|
||
4. REVIEW — APPROVE found in PR comments
|
||
5. MERGE — PR state is MERGED
|
||
6. MEMORY — PR#N distilled into repos/{host}/{org}/{repo}.md
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import functools
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from dataclasses import dataclass
|
||
from enum import StrEnum
|
||
from pathlib import Path
|
||
|
||
|
||
def _resolve_repo_root() -> Path:
|
||
"""Resolve repo root via git (cwd-aware), fallback to script location."""
|
||
result = subprocess.run(
|
||
["git", "rev-parse", "--show-toplevel"], capture_output=True, text=True, check=False
|
||
)
|
||
if result.returncode == 0 and result.stdout.strip():
|
||
return Path(result.stdout.strip()).resolve()
|
||
return Path(__file__).resolve().parent.parent.parent
|
||
|
||
|
||
REPO_ROOT = _resolve_repo_root()
|
||
_MEMORY_BASE = os.environ.get(
|
||
"OPENCODE_MEMORY_DIR",
|
||
str(REPO_ROOT / "app_data" / "opencode-memory"),
|
||
)
|
||
MEMORY_DIR = Path(_MEMORY_BASE) / "repos"
|
||
|
||
PHASE_NAMES = ["ISSUE", "IMPLEMENT", "CI", "REVIEW", "MERGE", "MEMORY"]
|
||
|
||
CI_WAIT_TIMEOUT = 300
|
||
CI_POLL_INTERVAL = 10
|
||
CI_NO_RUNS_RETRY = 3
|
||
CI_NO_RUNS_INTERVAL = 5
|
||
|
||
CLOSURE_RE = re.compile(r"(?:Closes|Fixes|Resolves)\s+#(\d+)", re.IGNORECASE)
|
||
REVIEW_APPROVE_RE = re.compile(
|
||
r"## Code Review Summary.*?###\s*Verdict:\s*APPROVE\b",
|
||
re.IGNORECASE | re.DOTALL,
|
||
)
|
||
REVIEW_VERDICT_RE = re.compile(
|
||
r"## Code Review Summary.*?###\s*Verdict:\s*(\w+)",
|
||
re.IGNORECASE | re.DOTALL,
|
||
)
|
||
JSON_FIELD_RE = re.compile(r'"(\w+)"\s*:\s*"?([^",}]*)"?', re.IGNORECASE)
|
||
|
||
|
||
class PhaseStatus(StrEnum):
|
||
"""Phase check result."""
|
||
|
||
DONE = "DONE"
|
||
NOT_DONE = "NOT_DONE"
|
||
AMBIGUOUS = "AMBIGUOUS"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PhaseResult:
|
||
"""Result of a single phase check."""
|
||
|
||
status: PhaseStatus
|
||
detail: str
|
||
|
||
|
||
def _forgejo_request(method: str, path: str, body: dict | None = None) -> tuple[int, str, str]:
|
||
"""Forgejo REST API call. Returns (status_code, body_text, error).
|
||
|
||
On non-2xx returns (status_code, body_text, ""). ``urlopen`` raises
|
||
HTTPError for non-2xx which carries the body; we surface it so callers
|
||
can branch on the HTTP status. Network/parse errors return (0, "", err).
|
||
"""
|
||
base = os.environ.get("FORGEJO_URL")
|
||
token = os.environ.get("FORGEJO_TOKEN")
|
||
headers = {"Authorization": f"token {token}", "Accept": "application/json"}
|
||
data = None
|
||
if body is not None:
|
||
headers["Content-Type"] = "application/json"
|
||
data = json.dumps(body).encode()
|
||
req = urllib.request.Request( # noqa: S310 - base URL is operator-configured
|
||
f"{base}/api/v1{path}", method=method, headers=headers, data=data
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req) as r: # noqa: S310 - base URL is operator-configured
|
||
return r.status, r.read().decode("utf-8", "replace"), ""
|
||
except urllib.error.HTTPError as e:
|
||
return e.code, e.read().decode("utf-8", "replace"), ""
|
||
except OSError as e:
|
||
return 0, "", str(e)
|
||
|
||
|
||
def _forgejo_ci_rollup(repo: str, sha: str) -> tuple[int, str, str]:
|
||
"""Build a GitHub-style statusCheckRollup from the Forgejo commit status."""
|
||
sc, text, err = _forgejo_request("GET", f"/repos/{repo}/commits/{sha}/status")
|
||
if sc != 200:
|
||
return 1, "", err or f"commit status HTTP {sc}"
|
||
combined = json.loads(text)
|
||
rollup: list[dict] = []
|
||
for s in combined.get("statuses", []):
|
||
st = s.get("status", "").lower()
|
||
if st == "success":
|
||
rollup.append({"status": "COMPLETED", "conclusion": "SUCCESS"})
|
||
elif st == "pending":
|
||
rollup.append({"status": "IN_PROGRESS", "conclusion": ""})
|
||
elif st in ("failure", "error"):
|
||
rollup.append({"status": "COMPLETED", "conclusion": "FAILURE"})
|
||
else:
|
||
rollup.append({"status": "QUEUED", "conclusion": ""})
|
||
return 0, json.dumps({"statusCheckRollup": rollup}), ""
|
||
|
||
|
||
def _forgejo_pr_view(repo: str, n: str, fields: str) -> tuple[int, str, str]: # noqa: PLR0911
|
||
"""Translate ``gh pr view N --json <fields>`` to Forgejo API calls."""
|
||
rc, pr_text, err = _forgejo_request("GET", f"/repos/{repo}/pulls/{n}")
|
||
if rc != 200:
|
||
return 1, "", err or f"PR API HTTP {rc}"
|
||
pr = json.loads(pr_text)
|
||
if fields == "statusCheckRollup":
|
||
return _forgejo_ci_rollup(repo, pr["head"]["sha"])
|
||
if fields == "comments":
|
||
sc, comments_text, ce = _forgejo_request("GET", f"/repos/{repo}/issues/{n}/comments")
|
||
if sc != 200:
|
||
return 1, "", ce or f"comments HTTP {sc}"
|
||
comments = json.loads(comments_text)
|
||
return 0, json.dumps({"comments": [{"body": c.get("body", "")} for c in comments]}), ""
|
||
if fields == "state":
|
||
state = "MERGED" if pr.get("merged") else pr.get("state", "").upper()
|
||
return 0, json.dumps({"state": state}), ""
|
||
if fields:
|
||
return 0, json.dumps({fields: pr.get(fields, "")}), ""
|
||
return 0, pr_text, ""
|
||
|
||
|
||
def _forgejo_pr_dispatch(repo: str, args: list[str]) -> tuple[int, str, str]:
|
||
"""Translate ``gh pr ...`` argv to Forgejo API calls."""
|
||
action = args[1] if len(args) > 1 else ""
|
||
num = args[2] if len(args) > 2 and args[2].lstrip("-").isdigit() else None
|
||
if action == "view" and num is not None:
|
||
fields_idx = args.index("--json") if "--json" in args else -1
|
||
fields = args[fields_idx + 1] if fields_idx >= 0 else ""
|
||
return _forgejo_pr_view(repo, num, fields)
|
||
if action == "list":
|
||
sc, out_text, e = _forgejo_request("GET", f"/repos/{repo}/pulls?state=open")
|
||
if sc != 200:
|
||
return 1, "", e or f"pr list HTTP {sc}"
|
||
pulls = json.loads(out_text)
|
||
return 0, json.dumps([{"number": p["number"]} for p in pulls]), ""
|
||
return 1, "", f"gh pr argv {args!r} not supported in Forgejo mode"
|
||
|
||
|
||
def _forgejo_gh_dispatch(args: list[str]) -> tuple[int, str, str] | None: # noqa: PLR0911
|
||
"""Translate a ``gh`` argv to a Forgejo API call. Returns None to defer.
|
||
|
||
Returns ``(rc, stdout, stderr)`` shaped like ``run_cmd`` so callers stay
|
||
unchanged. Returns ``None`` if ``FORGEJO_URL`` is unset (defer to gh) or
|
||
the argv is not a supported gh subcommand.
|
||
"""
|
||
if not os.environ.get("FORGEJO_URL"):
|
||
return None
|
||
if not os.environ.get("FORGEJO_TOKEN"):
|
||
return 1, "", "Forgejo mode requires FORGEJO_TOKEN"
|
||
if not args or args[0] != "gh":
|
||
return None
|
||
repo_idx = args.index("--repo") if "--repo" in args else -1
|
||
repo = args[repo_idx + 1] if repo_idx >= 0 else None
|
||
if repo is None:
|
||
return 1, "", "Forgejo mode requires --repo owner/name"
|
||
sub = args[1] if len(args) > 1 else ""
|
||
if sub == "auth" and "status" in args:
|
||
return 0, "", ""
|
||
if sub == "pr":
|
||
return _forgejo_pr_dispatch(repo, args)
|
||
if sub == "issue" and len(args) > 2 and args[2].lstrip("-").isdigit():
|
||
sc, out_text, e = _forgejo_request("GET", f"/repos/{repo}/issues/{args[2]}")
|
||
if sc == 200:
|
||
return 0, out_text, ""
|
||
return 1, "", e or f"issue HTTP {sc}"
|
||
return 1, "", f"gh argv {args!r} not supported in Forgejo mode"
|
||
|
||
|
||
def run_cmd(args: list[str]) -> tuple[int, str, str]:
|
||
"""Run a command, return (returncode, stdout, stderr).
|
||
|
||
Forgejo dispatch (ADR-forgejo): when ``FORGEJO_URL`` is set, ``gh`` argv
|
||
is translated to a Forgejo REST API call instead of spawning ``gh``.
|
||
GitHub users (no ``FORGEJO_URL``) see byte-identical behaviour — the
|
||
``gh`` / ``git`` subprocess path is untouched.
|
||
"""
|
||
if args and args[0] == "gh":
|
||
result = _forgejo_gh_dispatch(args)
|
||
if result is not None:
|
||
return result
|
||
proc = subprocess.run(args, capture_output=True, text=True, check=False)
|
||
return proc.returncode, proc.stdout, proc.stderr
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CiPollConfig:
|
||
"""CI polling parameters.
|
||
|
||
Priority: CLI flag > env var > constant in code.
|
||
"""
|
||
|
||
wait_timeout: int
|
||
poll_interval: int
|
||
|
||
|
||
def _env_int(name: str, default: int) -> int:
|
||
"""Read int from env var, fallback to default on missing/invalid."""
|
||
raw = os.environ.get(name)
|
||
if raw is None:
|
||
return default
|
||
try:
|
||
return int(raw)
|
||
except ValueError:
|
||
return default
|
||
|
||
|
||
def _load_ci_config(argv: list[str] | None = None) -> CiPollConfig:
|
||
"""Build CiPollConfig from CLI flags + env vars + constants.
|
||
|
||
Priority: CLI flag > env var > constant. CLI flags ``--ci-wait-timeout N``
|
||
and ``--ci-poll-interval N`` parsed manually from ``argv`` (last wins).
|
||
"""
|
||
args = argv if argv is not None else sys.argv[1:]
|
||
cli_timeout: int | None = None
|
||
cli_interval: int | None = None
|
||
i = 0
|
||
while i < len(args):
|
||
if args[i] == "--ci-wait-timeout" and i + 1 < len(args):
|
||
try:
|
||
cli_timeout = int(args[i + 1])
|
||
except ValueError:
|
||
cli_timeout = None
|
||
i += 2
|
||
continue
|
||
if args[i] == "--ci-poll-interval" and i + 1 < len(args):
|
||
try:
|
||
cli_interval = int(args[i + 1])
|
||
except ValueError:
|
||
cli_interval = None
|
||
i += 2
|
||
continue
|
||
i += 1
|
||
|
||
timeout = (
|
||
cli_timeout
|
||
if cli_timeout is not None
|
||
else _env_int("OPENCODE_CI_WAIT_TIMEOUT", CI_WAIT_TIMEOUT)
|
||
)
|
||
interval = (
|
||
cli_interval
|
||
if cli_interval is not None
|
||
else _env_int("OPENCODE_CI_POLL_INTERVAL", CI_POLL_INTERVAL)
|
||
)
|
||
return CiPollConfig(wait_timeout=timeout, poll_interval=interval)
|
||
|
||
|
||
def parse_remote_url(url: str) -> tuple[str, str, str]:
|
||
"""Parse git remote URL into (host, org, repo).
|
||
|
||
Supports both HTTPS and SSH formats:
|
||
https://github.com/org/repo.git -> (github.com, org, repo)
|
||
git@github.com:org/repo.git -> (github.com, org, repo)
|
||
"""
|
||
ssh_match = re.match(r"git@([^:]+):([^/]+)/(.+?)(?:\.git)?$", url)
|
||
if ssh_match:
|
||
return ssh_match.group(1), ssh_match.group(2), ssh_match.group(3)
|
||
# `(?:[^/@]*@)?` optionally skips `user:password@` userinfo before host.
|
||
# Needed because `git config url.insteadOf` rewrites `https://github.com/`
|
||
# to `https://x-access-token:TOKEN@github.com/`, and `git remote get-url
|
||
# origin` returns the rewritten URL (see ADR-022 / ADR-023).
|
||
https_match = re.match(r"https?://(?:[^/@]*@)?([^/]+)/([^/]+)/(.+?)(?:\.git)?$", url)
|
||
if https_match:
|
||
return https_match.group(1), https_match.group(2), https_match.group(3)
|
||
raise ValueError(f"Cannot parse remote URL: {url}")
|
||
|
||
|
||
def _resolve_memory_base() -> tuple[Path, str]:
|
||
"""Derive (memory_dir, repo_name) from ``git remote get-url origin``."""
|
||
rc, out, err = run_cmd(["git", "remote", "get-url", "origin"])
|
||
if rc != 0:
|
||
raise RuntimeError(f"Cannot get git remote URL: {err.strip()}")
|
||
host, org, repo = parse_remote_url(out.strip())
|
||
return MEMORY_DIR / host / org, repo
|
||
|
||
|
||
def get_memory_files() -> list[Path]:
|
||
"""All ``repo*.md`` sorted by mtime descending (newest first)."""
|
||
base_dir, repo = _resolve_memory_base()
|
||
if not base_dir.exists():
|
||
return []
|
||
rot_pattern = re.compile(rf"^{re.escape(repo)}(-\d+)?$")
|
||
files = [f for f in base_dir.glob("*.md") if rot_pattern.fullmatch(f.stem)]
|
||
return sorted(files, key=lambda p: p.stat().st_mtime, reverse=True)
|
||
|
||
|
||
@functools.cache
|
||
def get_repo_full_name() -> str:
|
||
"""Return ``org/repo`` from git remote (cached, one call per run).
|
||
|
||
Used for GitHub Actions API URL: ``repos/{org}/{repo}/actions/runs``.
|
||
Cached via ``functools.cache`` (one git call per process); tests reset
|
||
via ``get_repo_full_name.cache_clear()``.
|
||
"""
|
||
rc, out, err = run_cmd(["git", "remote", "get-url", "origin"])
|
||
if rc != 0:
|
||
raise RuntimeError(f"Cannot get git remote URL: {err.strip()}")
|
||
_host, org, repo = parse_remote_url(out.strip())
|
||
return f"{org}/{repo}"
|
||
|
||
|
||
def extract_json_field(json_str: str, field: str) -> str | None:
|
||
"""Extract a string field value from JSON (simple regex, no json import)."""
|
||
match = re.search(rf'"{field}"\s*:\s*"([^"]*)"', json_str)
|
||
return match.group(1) if match else None
|
||
|
||
|
||
def check_gh_auth() -> str | None:
|
||
"""Check if gh CLI is authenticated. Returns error message or None."""
|
||
rc, _, err = run_cmd(["gh", "auth", "status"])
|
||
if rc != 0:
|
||
return f"gh CLI не авторизован: {err.strip()}"
|
||
return None
|
||
|
||
|
||
def check_issue(pr_number: int) -> PhaseResult:
|
||
"""Phase 1: ISSUE — issue exists and linked via Closes/Fixes #N."""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "body", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"PR #{pr_number} не существует")
|
||
|
||
matches = CLOSURE_RE.findall(out)
|
||
if not matches:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, "Closes/Fixes #N не найден в body")
|
||
|
||
issue_numbers = list({int(m) for m in matches})
|
||
if len(issue_numbers) > 1:
|
||
return PhaseResult(
|
||
PhaseStatus.AMBIGUOUS,
|
||
f"несколько issue в body: {', '.join(f'#{n}' for n in issue_numbers)}",
|
||
)
|
||
|
||
issue_num = issue_numbers[0]
|
||
rc2, _, err2 = run_cmd(["gh", "issue", "view", str(issue_num), "--repo", get_repo_full_name()])
|
||
if rc2 != 0:
|
||
return PhaseResult(
|
||
PhaseStatus.NOT_DONE,
|
||
f"issue #{issue_num} не существует: {err2.strip()}",
|
||
)
|
||
|
||
return PhaseResult(PhaseStatus.DONE, f"#{issue_num} связан через Closes #{issue_num}")
|
||
|
||
|
||
def check_implement(pr_number: int) -> PhaseResult:
|
||
"""Phase 2: IMPLEMENT — PR exists + PR body has 4 required headings."""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "body", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"PR #{pr_number} не существует")
|
||
try:
|
||
pr_body = json.loads(out).get("body", "") or ""
|
||
except (json.JSONDecodeError, ValueError):
|
||
return PhaseResult(
|
||
PhaseStatus.NOT_DONE, f"PR #{pr_number}: не удалось распарсить JSON body"
|
||
)
|
||
required = ["## Что сделано", "## Почему", "## Watch out", "## Pending"]
|
||
missing = [h for h in required if h not in pr_body]
|
||
if missing:
|
||
return PhaseResult(
|
||
PhaseStatus.NOT_DONE, f"PR body не содержит heading'и: {', '.join(missing)}"
|
||
)
|
||
return PhaseResult(PhaseStatus.DONE, "PR body: 4 heading'а валидны")
|
||
|
||
|
||
def check_ci(pr_number: int) -> PhaseResult:
|
||
"""Phase 3: CI — all checks on PR head SHA completed & success.
|
||
|
||
Uses ``gh pr view --json statusCheckRollup`` which aggregates ALL
|
||
workflows for the PR head SHA (CI, CI (always), ADR check, etc.).
|
||
This handles docs-only PRs where ``ci.yml`` has ``paths-ignore`` and
|
||
only ``always-ci.yml`` runs.
|
||
"""
|
||
config = _load_ci_config()
|
||
return _run_ci_loop(pr_number, config)
|
||
|
||
|
||
def _run_ci_loop(pr_number: int, config: CiPollConfig) -> PhaseResult:
|
||
"""Initial CI query + edge-case dispatch + delegate to poll/no-checks helpers."""
|
||
kind, json_str, err = _query_ci_rollup(pr_number)
|
||
if kind == "error":
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, err)
|
||
if kind == "no_checks":
|
||
return _retry_no_checks(pr_number, config)
|
||
return _classify_rollup_with_poll(pr_number, config, json_str)
|
||
|
||
|
||
def _retry_no_checks(pr_number: int, config: CiPollConfig) -> PhaseResult:
|
||
"""Retry CI query when no checks registered yet (CI may lag after push).
|
||
|
||
Up to ``CI_NO_RUNS_RETRY`` total attempts, sleeping ``CI_NO_RUNS_INTERVAL``
|
||
between attempts. On success → classify/poll; exhausted → AMBIGUOUS.
|
||
"""
|
||
for attempt in range(CI_NO_RUNS_RETRY):
|
||
if attempt > 0:
|
||
time.sleep(CI_NO_RUNS_INTERVAL)
|
||
kind, json_str, err = _query_ci_rollup(pr_number)
|
||
if kind == "error":
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, err)
|
||
if kind == "rollup":
|
||
return _classify_rollup_with_poll(pr_number, config, json_str)
|
||
return PhaseResult(
|
||
PhaseStatus.AMBIGUOUS,
|
||
f"нет CI checks на PR #{pr_number} — возможна проблема триггера",
|
||
)
|
||
|
||
|
||
def _classify_rollup_with_poll(pr_number: int, config: CiPollConfig, json_str: str) -> PhaseResult:
|
||
"""Classify rollup; if in_progress → poll until completed or timeout."""
|
||
result = _classify_rollup(json_str)
|
||
if result.status != PhaseStatus.AMBIGUOUS or "in progress" not in result.detail.lower():
|
||
return result
|
||
elapsed = 0
|
||
while elapsed < config.wait_timeout:
|
||
if elapsed + config.poll_interval > config.wait_timeout:
|
||
break
|
||
time.sleep(config.poll_interval)
|
||
elapsed += config.poll_interval
|
||
kind, json_str_new, err = _query_ci_rollup(pr_number)
|
||
if kind == "error":
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, err)
|
||
if kind == "no_checks":
|
||
return PhaseResult(
|
||
PhaseStatus.AMBIGUOUS,
|
||
f"нет CI checks на PR #{pr_number} — возможна проблема триггера",
|
||
)
|
||
result = _classify_rollup(json_str_new)
|
||
if result.status != PhaseStatus.AMBIGUOUS or "in progress" not in result.detail.lower():
|
||
return result
|
||
return PhaseResult(
|
||
PhaseStatus.AMBIGUOUS,
|
||
f"CI ещё идёт после {config.wait_timeout}s — проверь вручную: gh pr checks {pr_number}",
|
||
)
|
||
|
||
|
||
def _query_ci_rollup(pr_number: int) -> tuple[str, str, str]:
|
||
"""Query PR statusCheckRollup via gh CLI.
|
||
|
||
GitHub aggregates ALL checks (CI, CI (always), ADR check) for PR head SHA.
|
||
Return ``(kind, json_str, error)`` where ``kind`` is:
|
||
- ``"error"`` — API call failed (rc != 0).
|
||
- ``"no_checks"`` — rollup array is empty (no checks registered yet).
|
||
- ``"rollup"`` — JSON with statusCheckRollup array, ``json_str`` set.
|
||
"""
|
||
rc, out, err = run_cmd(
|
||
[
|
||
"gh",
|
||
"pr",
|
||
"view",
|
||
str(pr_number),
|
||
"--json",
|
||
"statusCheckRollup",
|
||
"--repo",
|
||
get_repo_full_name(),
|
||
]
|
||
)
|
||
if rc != 0:
|
||
return "error", "", f"PR API error: {err.strip()}"
|
||
json_str = out.strip()
|
||
if not json_str:
|
||
return "no_checks", "", ""
|
||
if re.search(r'"statusCheckRollup"\s*:\s*\[\s*\]', json_str):
|
||
return "no_checks", "", ""
|
||
return "rollup", json_str, ""
|
||
|
||
|
||
def _classify_rollup(json_str: str) -> PhaseResult:
|
||
"""Classify CI status from statusCheckRollup JSON.
|
||
|
||
All checks COMPLETED + SUCCESS/SKIPPED/NEUTRAL → DONE.
|
||
Any check COMPLETED + non-success conclusion → NOT_DONE.
|
||
Any check IN_PROGRESS/QUEUED/PENDING → AMBIGUOUS (poll).
|
||
"""
|
||
statuses = re.findall(r'"status"\s*:\s*"([^"]*)"', json_str, re.IGNORECASE)
|
||
if not statuses:
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, "no checks found in rollup")
|
||
|
||
in_progress = [s for s in statuses if s.upper() in ("IN_PROGRESS", "QUEUED", "PENDING")]
|
||
if in_progress:
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, "CI in progress")
|
||
|
||
conclusions = re.findall(r'"conclusion"\s*:\s*"([^"]*)"', json_str, re.IGNORECASE)
|
||
null_conclusions = re.findall(r'"conclusion"\s*:\s*null', json_str, re.IGNORECASE)
|
||
|
||
for c in conclusions:
|
||
if c.upper() not in ("SUCCESS", "SKIPPED", "NEUTRAL"):
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"CI {c.lower()} — fix needed")
|
||
|
||
if null_conclusions:
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, "CI completed but conclusion missing")
|
||
|
||
return PhaseResult(PhaseStatus.DONE, "CI green")
|
||
|
||
|
||
def _extract_comment_bodies(json_str: str) -> list[str]:
|
||
"""Extract 'body' fields from gh pr view --json comments output.
|
||
|
||
Handles JSON string escaping (\\n, \\", \\\\).
|
||
"""
|
||
bodies = []
|
||
for match in re.finditer(r'"body"\s*:\s*"((?:[^"\\]|\\.)*)"', json_str):
|
||
raw = match.group(1)
|
||
body = raw.encode().decode("unicode_escape")
|
||
bodies.append(body)
|
||
return bodies
|
||
|
||
|
||
def check_review(pr_number: int) -> PhaseResult:
|
||
"""Phase 5: REVIEW — APPROVE found in PR comments from code reviewer.
|
||
|
||
Looks for '## Code Review Summary' heading
|
||
with '### Verdict: APPROVE'. Only the latest reviewer comment counts —
|
||
if reviewer changed from APPROVE to REQUEST_CHANGES, NOT_DONE.
|
||
"""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "comments", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, "не удалось получить комментарии PR")
|
||
|
||
comment_bodies = _extract_comment_bodies(out)
|
||
if not comment_bodies:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, "нет комментариев PR")
|
||
|
||
reviewer_verdict = None
|
||
for body in comment_bodies:
|
||
if REVIEW_VERDICT_RE.search(body):
|
||
match = REVIEW_VERDICT_RE.search(body)
|
||
reviewer_verdict = match.group(1).upper() if match else None
|
||
|
||
if reviewer_verdict is None:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, "Code Review Summary не найден в комментариях")
|
||
|
||
if reviewer_verdict == "APPROVE":
|
||
return PhaseResult(PhaseStatus.DONE, "APPROVE найден в Code Review Summary")
|
||
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"последний verdict reviewer'а: {reviewer_verdict}")
|
||
|
||
|
||
def check_merge(pr_number: int) -> PhaseResult:
|
||
"""Phase 6: MERGE — PR state is MERGED."""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "state", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"PR #{pr_number} не существует")
|
||
|
||
state = extract_json_field(out, "state")
|
||
if state is None:
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, "не удалось распарсить state PR")
|
||
|
||
if state == "MERGED":
|
||
return PhaseResult(PhaseStatus.DONE, "merged")
|
||
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"state={state}")
|
||
|
||
|
||
def check_memory(pr_number: int) -> PhaseResult:
|
||
"""Phase 7: MEMORY — PR#N distilled into memory file."""
|
||
try:
|
||
files = get_memory_files()
|
||
except (RuntimeError, ValueError) as exc:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, str(exc))
|
||
|
||
if not files:
|
||
return PhaseResult(
|
||
PhaseStatus.NOT_DONE,
|
||
"memory files не найдены",
|
||
)
|
||
|
||
pattern = f"PR#{pr_number}"
|
||
for f in files:
|
||
if pattern in f.read_text():
|
||
return PhaseResult(PhaseStatus.DONE, f"{pattern} в {f.name}")
|
||
|
||
return PhaseResult(
|
||
PhaseStatus.NOT_DONE,
|
||
f"{pattern} не найден в {len(files)} файл(ах)",
|
||
)
|
||
|
||
|
||
def get_pr_title(pr_number: int) -> str:
|
||
"""Get PR title via gh CLI."""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "title", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return f"PR #{pr_number}"
|
||
title = extract_json_field(out, "title")
|
||
return title if title else f"PR #{pr_number}"
|
||
|
||
|
||
NEXT_ACTIONS: dict[str, str] = {
|
||
"ISSUE": "dispatch subagent (subagent_type=general, template=A) for PR #N",
|
||
"IMPLEMENT": "dispatch subagent (subagent_type=general, template=A) for PR #N",
|
||
"CI": "проверь статус CI вручную (gh run view)",
|
||
"REVIEW": "dispatch subagent (subagent_type=reviewer, template=C) for PR #N",
|
||
"MERGE": "call merge_pr tool with pr_number=N",
|
||
"MEMORY": "dispatch subagent (subagent_type=memory-syncer, template=E) for PR #N",
|
||
}
|
||
|
||
REVIEW_NEXT_REQUEST_CHANGES = (
|
||
"dispatch subagent (subagent_type=general) with prompt 'fix reviewer comments: <list>', "
|
||
"commit, push → re-loop (pipeline_status проверит CI автоматически)"
|
||
)
|
||
REVIEW_NEXT_NEEDS_DISCUSSION = "уточни вопросы с автором PR (verdict: NEEDS_DISCUSSION)"
|
||
REVIEW_NEXT_DEFAULT = NEXT_ACTIONS["REVIEW"]
|
||
|
||
STATUS_ICONS: dict[PhaseStatus, str] = {
|
||
PhaseStatus.DONE: "✅",
|
||
PhaseStatus.NOT_DONE: "❌",
|
||
PhaseStatus.AMBIGUOUS: "⚠️",
|
||
}
|
||
|
||
|
||
def get_next_action(phase_name: str, pr_number: int) -> str:
|
||
"""Get NEXT action description for a not-done phase."""
|
||
action = NEXT_ACTIONS.get(phase_name, "уточнить статус")
|
||
return action.replace("N", str(pr_number))
|
||
|
||
|
||
def get_next_action_review(result: PhaseResult) -> str:
|
||
"""REVIEW-фаза: NEXT зависит от вердикта reviewer'а в result.detail."""
|
||
detail = result.detail.upper()
|
||
if "REQUEST_CHANGES" in detail:
|
||
return REVIEW_NEXT_REQUEST_CHANGES
|
||
if "NEEDS_DISCUSSION" in detail:
|
||
return REVIEW_NEXT_NEEDS_DISCUSSION
|
||
return REVIEW_NEXT_DEFAULT
|
||
|
||
|
||
def run_all_checks(pr_number: int) -> list[PhaseResult]:
|
||
"""Run all 6 phase checks, return results in order."""
|
||
return [
|
||
check_issue(pr_number),
|
||
check_implement(pr_number),
|
||
check_ci(pr_number),
|
||
check_review(pr_number),
|
||
check_merge(pr_number),
|
||
check_memory(pr_number),
|
||
]
|
||
|
||
|
||
def find_current_phase(results: list[PhaseResult]) -> int | None:
|
||
"""Return index of first not-done phase, or None if all done."""
|
||
for i, result in enumerate(results):
|
||
if result.status != PhaseStatus.DONE:
|
||
return i
|
||
return None
|
||
|
||
|
||
def format_single_pr(pr_number: int, results: list[PhaseResult]) -> str:
|
||
"""Format detailed output for a single PR."""
|
||
title = get_pr_title(pr_number)
|
||
lines = [f"PR #{pr_number}: {title}", ""]
|
||
|
||
for i, (name, result) in enumerate(zip(PHASE_NAMES, results, strict=True)):
|
||
icon = STATUS_ICONS[result.status]
|
||
lines.append(f"{icon} {i + 1}. {name:<12} {result.detail}")
|
||
|
||
lines.append("")
|
||
|
||
current = find_current_phase(results)
|
||
if current is None:
|
||
lines.append("Status: COMPLETE")
|
||
else:
|
||
result = results[current]
|
||
if result.status == PhaseStatus.AMBIGUOUS:
|
||
lines.append(f"AMBIGUOUS: {result.detail}")
|
||
lines.append("NEXT: уточните статус вручную")
|
||
else:
|
||
if PHASE_NAMES[current] == "REVIEW":
|
||
action = get_next_action_review(result)
|
||
else:
|
||
action = get_next_action(PHASE_NAMES[current], pr_number)
|
||
lines.append(f"NEXT: {action}")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def list_open_pr_numbers() -> list[int]:
|
||
"""List numbers of all open PRs."""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "list", "--state", "open", "--json", "number", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return []
|
||
numbers = re.findall(r'"number"\s*:\s*(\d+)', out)
|
||
return sorted(int(n) for n in numbers)
|
||
|
||
|
||
def format_pr_row(pr_number: int) -> str:
|
||
"""Format a single row for the open-PRs table."""
|
||
results = run_all_checks(pr_number)
|
||
title = get_pr_title(pr_number)
|
||
icon_str = "".join(STATUS_ICONS[r.status] for r in results)
|
||
|
||
current = find_current_phase(results)
|
||
if current is None:
|
||
next_action = "COMPLETE"
|
||
elif PHASE_NAMES[current] == "REVIEW":
|
||
next_action = get_next_action_review(results[current])
|
||
else:
|
||
next_action = get_next_action(PHASE_NAMES[current], pr_number)
|
||
|
||
short_title = title[:40] + "..." if len(title) > 40 else title
|
||
return f"PR#{pr_number:<5} {short_title:<43} {icon_str} NEXT: {next_action}"
|
||
|
||
|
||
def format_table(pr_numbers: list[int]) -> str:
|
||
"""Format table of all open PRs."""
|
||
if not pr_numbers:
|
||
return "Нет открытых PR"
|
||
rows = [format_pr_row(n) for n in pr_numbers]
|
||
return "\n".join(rows)
|
||
|
||
|
||
def pr_exists(pr_number: int) -> bool:
|
||
"""Check if PR exists via gh CLI."""
|
||
rc, _, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "number", "--repo", get_repo_full_name()]
|
||
)
|
||
return rc == 0
|
||
|
||
|
||
def main() -> None:
|
||
"""Entry point: parse args and dispatch to single-PR or table mode."""
|
||
auth_error = check_gh_auth()
|
||
if auth_error:
|
||
print(auth_error, file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
if len(sys.argv) > 1:
|
||
try:
|
||
pr_number = int(sys.argv[1])
|
||
except ValueError:
|
||
print(f"Некорректный номер PR: {sys.argv[1]}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
if not pr_exists(pr_number):
|
||
print(f"PR #{pr_number} не существует", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
results = run_all_checks(pr_number)
|
||
print(format_single_pr(pr_number, results))
|
||
else:
|
||
pr_numbers = list_open_pr_numbers()
|
||
print(format_table(pr_numbers))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|