From 99c874d4233ec2d35c5c12a709ba32fd2fd96f3d Mon Sep 17 00:00:00 2001 From: RAPHAEL LUGO Date: Mon, 15 Jun 2026 17:52:26 -0400 Subject: [PATCH] fix(codex): PR health label check state (#986) ## Description Fix the PR health label job so `status: ci failing` reflects the latest check attempt for each check, not historical failed or cancelled attempts that still appear in `statusCheckRollup`. This showed up on #984: the current checks were green, but the label job kept `status: ci failing` because older failed template runs were still present in the rollup payload. ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - Added a small `.github/scripts/pr-health-labels.py` helper that groups check-rollup entries by logical check name and evaluates only the newest entry for each check. - Updated the PR health workflow label job to call the helper instead of treating any historical failing rollup entry as current failure. - Added regression tests for historical failures followed by latest passing attempts, plus current latest failure behavior. ## Testing - [x] Unit tests pass (`pytest`) - [x] New tests added for new functionality - [x] Manual testing performed ### Test Output ```text PYTEST_ADDOPTS='-p no:cacheprovider' pytest scripts/tests -q 47 passed, 1 warning in 0.39s python .github/scripts/pr-health-labels.py --state-json '' passing data=$(gh pr view 984 --repo chopratejas/headroom --json statusCheckRollup) python .github/scripts/pr-health-labels.py --state-json "$data" passing ``` ## Real Behavior Proof - Environment: macOS local checkout, Python 3.11.7, live GitHub PR #984 check-rollup payload fetched with `gh pr view`. - Exact command / steps: Added regression coverage for historical failed/cancelled check runs followed by latest successful runs, ran the scripts test suite, and evaluated live PR #984's `statusCheckRollup` with the new helper. - Observed result: The helper returns `passing` for #984's live payload even though older failed/cancelled check runs are still present, while still returning `failing` when the latest attempt for a check failed. - Not tested: A full GitHub Actions run of the updated workflow on upstream before merge; this PR should exercise the workflow on itself. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review --- .github/scripts/pr-health-labels.py | 79 ++++++++++++++++++++++ .github/workflows/pr-health.yml | 9 +-- scripts/tests/test_pr_health_labels.py | 94 ++++++++++++++++++++++++++ 3 files changed, 174 insertions(+), 8 deletions(-) create mode 100644 .github/scripts/pr-health-labels.py create mode 100644 scripts/tests/test_pr_health_labels.py diff --git a/.github/scripts/pr-health-labels.py b/.github/scripts/pr-health-labels.py new file mode 100644 index 000000000..dfa9eaacc --- /dev/null +++ b/.github/scripts/pr-health-labels.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""Helpers for PR health maintenance labels.""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime, timezone +from typing import Any + +FAILING_STATES = {"FAILURE", "TIMED_OUT", "ACTION_REQUIRED", "CANCELLED", "ERROR"} + + +def _parse_timestamp(value: Any) -> datetime: + if not isinstance(value, str) or not value: + return datetime.min.replace(tzinfo=timezone.utc) + normalized = value.removesuffix("Z") + "+00:00" if value.endswith("Z") else value + try: + parsed = datetime.fromisoformat(normalized) + except ValueError: + return datetime.min.replace(tzinfo=timezone.utc) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed + + +def _check_key(check: dict[str, Any]) -> tuple[str, str]: + workflow = str(check.get("workflowName") or check.get("workflow") or "") + name = str(check.get("name") or check.get("context") or "") + return workflow, name + + +def _check_time(check: dict[str, Any]) -> datetime: + return max( + _parse_timestamp(check.get("startedAt")), + _parse_timestamp(check.get("completedAt")), + ) + + +def _state(check: dict[str, Any]) -> str: + return str(check.get("conclusion") or check.get("state") or "").upper() + + +def current_checks(payload: dict[str, Any]) -> list[dict[str, Any]]: + latest_by_key: dict[tuple[str, str], dict[str, Any]] = {} + for check in payload.get("statusCheckRollup") or []: + if not isinstance(check, dict): + continue + key = _check_key(check) + if not any(key): + continue + previous = latest_by_key.get(key) + if previous is None or _check_time(check) >= _check_time(previous): + latest_by_key[key] = check + return list(latest_by_key.values()) + + +def check_state(payload: dict[str, Any]) -> str: + for check in current_checks(payload): + if _state(check) in FAILING_STATES: + return "failing" + return "passing" + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--state-json", required=True, help="JSON from gh pr view") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv or sys.argv[1:]) + print(check_state(json.loads(args.state_json))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/.github/workflows/pr-health.yml b/.github/workflows/pr-health.yml index abfcae1c2..1b266e05c 100644 --- a/.github/workflows/pr-health.yml +++ b/.github/workflows/pr-health.yml @@ -189,14 +189,7 @@ jobs: --json isDraft,labels,mergeStateStatus,statusCheckRollup)" merge_state="$(jq -r '.mergeStateStatus // "UNKNOWN"' <<<"$data")" - check_state="$(jq -r ' - [ - (.statusCheckRollup // [])[] - | select((.conclusion // .state // "") as $s - | ["FAILURE", "TIMED_OUT", "ACTION_REQUIRED", "CANCELLED", "ERROR"] | index($s)) - ] - | if length > 0 then "failing" else "passing" end - ' <<<"$data")" + check_state="$(python3 .github/scripts/pr-health-labels.py --state-json "$data")" is_draft="$(jq -r '.isDraft' <<<"$data")" if [[ "$merge_state" == "BEHIND" ]]; then diff --git a/scripts/tests/test_pr_health_labels.py b/scripts/tests/test_pr_health_labels.py new file mode 100644 index 000000000..a7bb5ef18 --- /dev/null +++ b/scripts/tests/test_pr_health_labels.py @@ -0,0 +1,94 @@ +"""Tests for pr-health-labels.py.""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + + +def _load_module(): + script = Path(__file__).parents[2] / ".github" / "scripts" / "pr-health-labels.py" + spec = importlib.util.spec_from_file_location("pr_health_labels", script) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def test_check_state_ignores_historical_failures_when_latest_attempt_passed() -> None: + module = _load_module() + payload = { + "statusCheckRollup": [ + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "template", + "conclusion": "FAILURE", + "startedAt": "2026-06-14T13:38:35Z", + "completedAt": "2026-06-14T13:38:46Z", + }, + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "template", + "conclusion": "SUCCESS", + "startedAt": "2026-06-14T13:43:26Z", + "completedAt": "2026-06-14T13:43:35Z", + }, + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "label", + "conclusion": "CANCELLED", + "startedAt": "2026-06-14T13:43:18Z", + "completedAt": "2026-06-14T13:43:24Z", + }, + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "label", + "conclusion": "SUCCESS", + "startedAt": "2026-06-14T13:43:26Z", + "completedAt": "2026-06-14T13:43:35Z", + }, + { + "__typename": "CheckRun", + "workflowName": "", + "name": "GitGuardian Security Checks", + "conclusion": "SUCCESS", + "startedAt": "2026-06-14T13:38:32Z", + "completedAt": "2026-06-14T13:39:04Z", + }, + ] + } + + assert module.check_state(payload) == "passing" + + +def test_check_state_fails_when_latest_attempt_failed() -> None: + module = _load_module() + payload = { + "statusCheckRollup": [ + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "template", + "conclusion": "SUCCESS", + "startedAt": "2026-06-14T13:38:35Z", + "completedAt": "2026-06-14T13:38:46Z", + }, + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "template", + "conclusion": "FAILURE", + "startedAt": "2026-06-14T13:43:26Z", + "completedAt": "2026-06-14T13:43:35Z", + }, + ] + } + + assert module.check_state(payload) == "failing"