ossrules.md

babysit-pr Skill

Babysit a GitHub pull request after creation by continuously polling review comments, CI checks/workflow runs, and mergeability state until the PR is merged/closed or user help is required. Diagnose failures, retry likely flaky failures up to 3 times, auto-fix/push branch-related issues when appropriate, and keep watching open PRs so fresh review feedback is surfaced promptly. Use when the user asks Codex to monitor a PR, watch CI, handle review comments, or keep an eye on failures and feedback on an open PR.

Install
npx skills add 'https://github.com/openai/codex/tree/main/.codex/skills/babysit-pr'
Download bundle ↓

scripts/gh_pr_watch.py

scripts/gh_pr_watch.pyBrowse 6 files
View on GitHub
6,967 tokens

30,134 bytes
Token encoding: o200k_base
Snapshot 18d7ace

← Back to SKILL.md
#!/usr/bin/env python3"""Watch GitHub PR CI and review activity for Codex PR babysitting workflows.""" import argparseimport jsonimport osimport reimport subprocessimport sysimport tempfileimport timefrom pathlib import Pathfrom urllib.parse import urlparse FAILED_RUN_CONCLUSIONS = {    "failure",    "timed_out",    "cancelled",    "action_required",    "startup_failure",    "stale",}PENDING_CHECK_STATES = {    "QUEUED",    "IN_PROGRESS",    "PENDING",    "WAITING",    "REQUESTED",}REVIEW_BOT_LOGIN_KEYWORDS = {    "codex",}TRUSTED_AUTHOR_ASSOCIATIONS = {    "OWNER",    "MEMBER",    "COLLABORATOR",}MERGE_BLOCKING_REVIEW_DECISIONS = {    "REVIEW_REQUIRED",    "CHANGES_REQUESTED",}MERGE_CONFLICT_OR_BLOCKING_STATES = {    "BLOCKED",    "DIRTY",    "DRAFT",    "UNKNOWN",}  class GhCommandError(RuntimeError):    pass  def parse_args():    parser = argparse.ArgumentParser(        description=(            "Normalize PR/CI/review state for Codex PR babysitting and optionally "            "trigger flaky reruns."        )    )    parser.add_argument("--pr", default="auto", help="auto, PR number, or PR URL")    parser.add_argument("--repo", help="Optional OWNER/REPO override")    parser.add_argument(        "--poll-seconds", type=int, default=30, help="Watch poll interval"    )    parser.add_argument(        "--max-flaky-retries",        type=int,        default=3,        help="Max rerun cycles per head SHA before stop recommendation",    )    parser.add_argument("--state-file", help="Path to state JSON file")    parser.add_argument(        "--once", action="store_true", help="Emit one snapshot and exit"    )    parser.add_argument(        "--watch", action="store_true", help="Continuously emit JSONL snapshots"    )    parser.add_argument(        "--retry-failed-now",        action="store_true",        help="Rerun failed jobs for current failed workflow runs when policy allows",    )    parser.add_argument(        "--json",        action="store_true",        help="Emit machine-readable output (default behavior for --once and --retry-failed-now)",    )    args = parser.parse_args()     if args.poll_seconds <= 0:        parser.error("--poll-seconds must be > 0")    if args.max_flaky_retries < 0:        parser.error("--max-flaky-retries must be >= 0")    if args.watch and args.retry_failed_now:        parser.error("--watch cannot be combined with --retry-failed-now")    if not args.once and not args.watch and not args.retry_failed_now:        args.once = True    return args  def _format_gh_error(cmd, err):    stdout = (err.stdout or "").strip()    stderr = (err.stderr or "").strip()    parts = [f"GitHub CLI command failed: {' '.join(cmd)}"]    if stdout:        parts.append(f"stdout: {stdout}")    if stderr:        parts.append(f"stderr: {stderr}")    return "\n".join(parts)  def gh_text(args, repo=None):    cmd = ["gh"]    # `gh api` does not accept `-R/--repo` on all gh versions. The watcher's    # API calls use explicit endpoints (e.g. repos/{owner}/{repo}/...), so the    # repo flag is unnecessary there.    if repo and (not args or args[0] != "api"):        cmd.extend(["-R", repo])    cmd.extend(args)    try:        proc = subprocess.run(cmd, check=True, capture_output=True, text=True)    except FileNotFoundError as err:        raise GhCommandError("`gh` command not found") from err    except subprocess.CalledProcessError as err:        raise GhCommandError(_format_gh_error(cmd, err)) from err    return proc.stdout  def gh_json(args, repo=None):    raw = gh_text(args, repo=repo).strip()    if not raw:        return None    try:        return json.loads(raw)    except json.JSONDecodeError as err:        raise GhCommandError(            f"Failed to parse JSON from gh output for {' '.join(args)}"        ) from err  def parse_pr_spec(pr_spec):    if pr_spec == "auto":        return {"mode": "auto", "value": None}    if re.fullmatch(r"\d+", pr_spec):        return {"mode": "number", "value": pr_spec}    parsed = urlparse(pr_spec)    if parsed.scheme and parsed.netloc and "/pull/" in parsed.path:        return {"mode": "url", "value": pr_spec}    raise ValueError("--pr must be 'auto', a PR number, or a PR URL")  def pr_view_fields():    return (        "number,url,state,mergedAt,closedAt,headRefName,headRefOid,"        "headRepository,headRepositoryOwner,mergeable,mergeStateStatus,reviewDecision"    )  def checks_fields():    return "name,state,bucket,link,workflow,event,startedAt,completedAt"  def resolve_pr(pr_spec, repo_override=None):    parsed = parse_pr_spec(pr_spec)    cmd = ["pr", "view"]    if parsed["value"] is not None:        cmd.append(parsed["value"])    cmd.extend(["--json", pr_view_fields()])    data = gh_json(cmd, repo=repo_override)    if not isinstance(data, dict):        raise GhCommandError("Unexpected PR payload from `gh pr view`")     pr_url = str(data.get("url") or "")    repo = (        repo_override        or extract_repo_from_pr_url(pr_url)        or extract_repo_from_pr_view(data)    )    if not repo:        raise GhCommandError("Unable to determine OWNER/REPO for the PR")     state = str(data.get("state") or "")    merged = bool(data.get("mergedAt"))    closed = bool(data.get("closedAt")) or state.upper() == "CLOSED"     return {        "number": int(data["number"]),        "url": pr_url,        "repo": repo,        "head_sha": str(data.get("headRefOid") or ""),        "head_branch": str(data.get("headRefName") or ""),        "state": state,        "merged": merged,        "closed": closed,        "mergeable": str(data.get("mergeable") or ""),        "merge_state_status": str(data.get("mergeStateStatus") or ""),        "review_decision": str(data.get("reviewDecision") or ""),    }  def extract_repo_from_pr_view(data):    head_repo = data.get("headRepository")    head_owner = data.get("headRepositoryOwner")    owner = None    name = None    if isinstance(head_owner, dict):        owner = head_owner.get("login") or head_owner.get("name")    elif isinstance(head_owner, str):        owner = head_owner    if isinstance(head_repo, dict):        name = head_repo.get("name")        repo_owner = head_repo.get("owner")        if not owner and isinstance(repo_owner, dict):            owner = repo_owner.get("login") or repo_owner.get("name")    elif isinstance(head_repo, str):        name = head_repo    if owner and name:        return f"{owner}/{name}"    return None  def extract_repo_from_pr_url(pr_url):    parsed = urlparse(pr_url)    parts = [p for p in parsed.path.split("/") if p]    if len(parts) >= 4 and parts[2] == "pull":        return f"{parts[0]}/{parts[1]}"    return None  def load_state(path):    if path.exists():        try:            data = json.loads(path.read_text())        except json.JSONDecodeError as err:            raise RuntimeError(f"State file is not valid JSON: {path}") from err        if not isinstance(data, dict):            raise RuntimeError(f"State file must contain an object: {path}")        return data, False    return {        "pr": {},        "started_at": None,        "last_seen_head_sha": None,        "retries_by_sha": {},        "seen_issue_comment_ids": [],        "seen_review_comment_ids": [],        "seen_review_ids": [],        "last_snapshot_at": None,    }, True  def save_state(path, state):    path.parent.mkdir(parents=True, exist_ok=True)    payload = json.dumps(state, indent=2, sort_keys=True) + "\n"    fd, tmp_name = tempfile.mkstemp(        prefix=f"{path.name}.", suffix=".tmp", dir=path.parent    )    tmp_path = Path(tmp_name)    try:        with os.fdopen(fd, "w", encoding="utf-8") as tmp_file:            tmp_file.write(payload)        os.replace(tmp_path, path)    except Exception:        try:            tmp_path.unlink(missing_ok=True)        except OSError:            pass        raise  def default_state_file_for(pr):    repo_slug = pr["repo"].replace("/", "-")    return Path(f"/tmp/codex-babysit-pr-{repo_slug}-pr{pr['number']}.json")  def get_pr_checks(pr_spec, repo):    parsed = parse_pr_spec(pr_spec)    cmd = ["pr", "checks"]    if parsed["value"] is not None:        cmd.append(parsed["value"])    cmd.extend(["--json", checks_fields()])    data = gh_json(cmd, repo=repo)    if data is None:        return []    if not isinstance(data, list):        raise GhCommandError("Unexpected payload from `gh pr checks`")    return data  def is_pending_check(check):    bucket = str(check.get("bucket") or "").lower()    state = str(check.get("state") or "").upper()    return bucket == "pending" or state in PENDING_CHECK_STATES  def summarize_checks(checks):    pending_count = 0    failed_count = 0    passed_count = 0    for check in checks:        bucket = str(check.get("bucket") or "").lower()        if is_pending_check(check):            pending_count += 1        if bucket == "fail":            failed_count += 1        if bucket == "pass":            passed_count += 1    return {        "pending_count": pending_count,        "failed_count": failed_count,        "passed_count": passed_count,        "all_terminal": pending_count == 0,    }  def get_workflow_runs_for_sha(repo, head_sha):    endpoint = f"repos/{repo}/actions/runs"    data = gh_json(        [            "api",            endpoint,            "-X",            "GET",            "-f",            f"head_sha={head_sha}",            "-f",            "per_page=100",        ],        repo=repo,    )    if not isinstance(data, dict):        raise GhCommandError("Unexpected payload from actions runs API")    runs = data.get("workflow_runs") or []    if not isinstance(runs, list):        raise GhCommandError("Expected `workflow_runs` to be a list")    return runs  def failed_runs_from_workflow_runs(runs, head_sha):    failed_runs = []    for run in runs:        if not isinstance(run, dict):            continue        if str(run.get("head_sha") or "") != head_sha:            continue        conclusion = str(run.get("conclusion") or "")        if conclusion not in FAILED_RUN_CONCLUSIONS:            continue        failed_runs.append(            {                "run_id": run.get("id"),                "workflow_name": run.get("name") or run.get("display_title") or "",                "status": str(run.get("status") or ""),                "conclusion": conclusion,                "html_url": str(run.get("html_url") or ""),            }        )    failed_runs.sort(        key=lambda item: (            str(item.get("workflow_name") or ""),            str(item.get("run_id") or ""),        )    )    return failed_runs  def get_jobs_for_run(repo, run_id):    endpoint = f"repos/{repo}/actions/runs/{run_id}/jobs"    data = gh_json(["api", endpoint, "-X", "GET", "-f", "per_page=100"], repo=repo)    if not isinstance(data, dict):        raise GhCommandError("Unexpected payload from actions run jobs API")    jobs = data.get("jobs") or []    if not isinstance(jobs, list):        raise GhCommandError("Expected `jobs` to be a list")    return jobs  def failed_jobs_from_workflow_runs(repo, runs, head_sha):    failed_jobs = []    for run in runs:        if not isinstance(run, dict):            continue        if str(run.get("head_sha") or "") != head_sha:            continue        run_id = run.get("id")        if run_id in (None, ""):            continue        run_status = str(run.get("status") or "")        run_conclusion = str(run.get("conclusion") or "")        if (            run_status.lower() == "completed"            and run_conclusion not in FAILED_RUN_CONCLUSIONS        ):            continue        jobs = get_jobs_for_run(repo, run_id)        for job in jobs:            if not isinstance(job, dict):                continue            conclusion = str(job.get("conclusion") or "")            if conclusion not in FAILED_RUN_CONCLUSIONS:                continue            job_id = job.get("id")            logs_endpoint = None            if job_id not in (None, ""):                logs_endpoint = f"repos/{repo}/actions/jobs/{job_id}/logs"            failed_jobs.append(                {                    "run_id": run_id,                    "workflow_name": run.get("name") or run.get("display_title") or "",                    "run_status": run_status,                    "run_conclusion": run_conclusion,                    "job_id": job_id,                    "job_name": str(job.get("name") or ""),                    "status": str(job.get("status") or ""),                    "conclusion": conclusion,                    "html_url": str(job.get("html_url") or ""),                    "logs_endpoint": logs_endpoint,                }            )    failed_jobs.sort(        key=lambda item: (            str(item.get("workflow_name") or ""),            str(item.get("job_name") or ""),            str(item.get("job_id") or ""),        )    )    return failed_jobs  def get_authenticated_login():    data = gh_json(["api", "user"])    if not isinstance(data, dict) or not data.get("login"):        raise GhCommandError(            "Unable to determine authenticated GitHub login from `gh api user`"        )    return str(data["login"])  def comment_endpoints(repo, pr_number):    return {        "issue_comment": f"repos/{repo}/issues/{pr_number}/comments",        "review_comment": f"repos/{repo}/pulls/{pr_number}/comments",        "review": f"repos/{repo}/pulls/{pr_number}/reviews",    }  def gh_api_list_paginated(endpoint, repo=None, per_page=100):    items = []    page = 1    while True:        sep = "&" if "?" in endpoint else "?"        page_endpoint = f"{endpoint}{sep}per_page={per_page}&page={page}"        payload = gh_json(["api", page_endpoint], repo=repo)        if payload is None:            break        if not isinstance(payload, list):            raise GhCommandError(f"Unexpected paginated payload from gh api {endpoint}")        items.extend(payload)        if len(payload) < per_page:            break        page += 1    return items  def normalize_issue_comments(items):    out = []    for item in items:        if not isinstance(item, dict):            continue        out.append(            {                "kind": "issue_comment",                "id": str(item.get("id") or ""),                "author": extract_login(item.get("user")),                "author_association": str(item.get("author_association") or ""),                "created_at": str(item.get("created_at") or ""),                "body": str(item.get("body") or ""),                "path": None,                "line": None,                "url": str(item.get("html_url") or ""),            }        )    return out  def normalize_review_comments(items, review_states):    out = []    for item in items:        if not isinstance(item, dict):            continue        review_id = str(item.get("pull_request_review_id") or "")        if review_states.get(review_id) == "PENDING":            continue        line = item.get("line")        if line is None:            line = item.get("original_line")        out.append(            {                "kind": "review_comment",                "id": str(item.get("id") or ""),                "author": extract_login(item.get("user")),                "author_association": str(item.get("author_association") or ""),                "created_at": str(item.get("created_at") or ""),                "body": str(item.get("body") or ""),                "path": item.get("path"),                "line": line,                "url": str(item.get("html_url") or ""),            }        )    return out  def normalize_reviews(items):    out = []    for item in items:        if not isinstance(item, dict):            continue        if str(item.get("state") or "").upper() == "PENDING":            continue        out.append(            {                "kind": "review",                "id": str(item.get("id") or ""),                "author": extract_login(item.get("user")),                "author_association": str(item.get("author_association") or ""),                "created_at": str(                    item.get("submitted_at") or item.get("created_at") or ""                ),                "body": str(item.get("body") or ""),                "path": None,                "line": None,                "url": str(item.get("html_url") or ""),            }        )    return out  def extract_login(user_obj):    if isinstance(user_obj, dict):        return str(user_obj.get("login") or "")    return ""  def is_bot_login(login):    return bool(login) and login.endswith("[bot]")  def is_actionable_review_bot_login(login):    if not is_bot_login(login):        return False    lower_login = login.lower()    return any(keyword in lower_login for keyword in REVIEW_BOT_LOGIN_KEYWORDS)  def is_trusted_human_review_author(item, authenticated_login):    author = str(item.get("author") or "")    if not author:        return False    if authenticated_login and author == authenticated_login:        return True    association = str(item.get("author_association") or "").upper()    return association in TRUSTED_AUTHOR_ASSOCIATIONS  def fetch_new_review_items(pr, state, fresh_state, authenticated_login=None):    repo = pr["repo"]    pr_number = pr["number"]    endpoints = comment_endpoints(repo, pr_number)     issue_payload = gh_api_list_paginated(endpoints["issue_comment"], repo=repo)    review_comment_payload = gh_api_list_paginated(        endpoints["review_comment"], repo=repo    )    review_payload = gh_api_list_paginated(endpoints["review"], repo=repo)     issue_items = normalize_issue_comments(issue_payload)    review_states = {        str(item.get("id")): str(item.get("state") or "").upper()        for item in review_payload        if isinstance(item, dict) and item.get("id") not in (None, "")    }    pending_review_ids = {        review_id        for review_id, review_state in review_states.items()        if review_state == "PENDING"    }    pending_review_comment_ids = {        str(item.get("id"))        for item in review_comment_payload        if isinstance(item, dict)        and item.get("id") not in (None, "")        and str(item.get("pull_request_review_id") or "") in pending_review_ids    }    review_comment_items = normalize_review_comments(        review_comment_payload, review_states    )    review_items = normalize_reviews(review_payload)    all_items = issue_items + review_comment_items + review_items     seen_issue = {str(x) for x in state.get("seen_issue_comment_ids") or []}    seen_review_comment = {str(x) for x in state.get("seen_review_comment_ids") or []}    seen_review = {str(x) for x in state.get("seen_review_ids") or []}    seen_review_comment.difference_update(pending_review_comment_ids)    seen_review.difference_update(pending_review_ids)     # On a brand-new state file, surface existing review activity instead of    # silently treating it as seen. This avoids missing already-published review    # feedback when monitoring starts after comments were posted.     new_items = []    for item in all_items:        item_id = item.get("id")        if not item_id:            continue        author = item.get("author") or ""        if not author:            continue        if is_bot_login(author):            if not is_actionable_review_bot_login(author):                continue        elif not is_trusted_human_review_author(item, authenticated_login):            continue         kind = item["kind"]        if kind == "issue_comment" and item_id in seen_issue:            continue        if kind == "review_comment" and item_id in seen_review_comment:            continue        if kind == "review" and item_id in seen_review:            continue         new_items.append(item)        if kind == "issue_comment":            seen_issue.add(item_id)        elif kind == "review_comment":            seen_review_comment.add(item_id)        elif kind == "review":            seen_review.add(item_id)     new_items.sort(        key=lambda item: (            item.get("created_at") or "",            item.get("kind") or "",            item.get("id") or "",        )    )    state["seen_issue_comment_ids"] = sorted(seen_issue)    state["seen_review_comment_ids"] = sorted(seen_review_comment)    state["seen_review_ids"] = sorted(seen_review)    return new_items  def current_retry_count(state, head_sha):    retries = state.get("retries_by_sha") or {}    value = retries.get(head_sha, 0)    try:        return int(value)    except (TypeError, ValueError):        return 0  def set_retry_count(state, head_sha, count):    retries = state.get("retries_by_sha")    if not isinstance(retries, dict):        retries = {}    retries[head_sha] = int(count)    state["retries_by_sha"] = retries  def unique_actions(actions):    out = []    seen = set()    for action in actions:        if action not in seen:            out.append(action)            seen.add(action)    return out  def is_pr_ready_to_merge(pr, checks_summary, new_review_items):    if pr["closed"] or pr["merged"]:        return False    if not checks_summary["all_terminal"]:        return False    if checks_summary["failed_count"] > 0 or checks_summary["pending_count"] > 0:        return False    if new_review_items:        return False    if str(pr.get("mergeable") or "") != "MERGEABLE":        return False    if str(pr.get("merge_state_status") or "") in MERGE_CONFLICT_OR_BLOCKING_STATES:        return False    if str(pr.get("review_decision") or "") in MERGE_BLOCKING_REVIEW_DECISIONS:        return False    return True  def recommend_actions(    pr,    checks_summary,    failed_runs,    failed_jobs,    new_review_items,    retries_used,    max_retries,):    actions = []    if pr["closed"] or pr["merged"]:        if new_review_items:            actions.append("process_review_comment")        actions.append("stop_pr_closed")        return unique_actions(actions)     if is_pr_ready_to_merge(pr, checks_summary, new_review_items):        actions.append("ready_to_merge")        return unique_actions(actions)     if new_review_items:        actions.append("process_review_comment")     has_failed_pr_checks = checks_summary["failed_count"] > 0 or bool(failed_jobs)    if has_failed_pr_checks:        if checks_summary["all_terminal"] and retries_used >= max_retries:            actions.append("stop_exhausted_retries")        else:            actions.append("diagnose_ci_failure")            if (                checks_summary["all_terminal"]                and failed_runs                and retries_used < max_retries            ):                actions.append("retry_failed_checks")     if not actions:        actions.append("idle")    return unique_actions(actions)  def collect_snapshot(args):    pr = resolve_pr(args.pr, repo_override=args.repo)    state_path = (        Path(args.state_file) if args.state_file else default_state_file_for(pr)    )    state, fresh_state = load_state(state_path)     if not state.get("started_at"):        state["started_at"] = int(time.time())     authenticated_login = get_authenticated_login()    new_review_items = fetch_new_review_items(        pr,        state,        fresh_state=fresh_state,        authenticated_login=authenticated_login,    )    # Surface review feedback before drilling into CI and mergeability details.    # That keeps the babysitter responsive to new comments even when other    # actions are also available.    # `gh pr checks -R <repo>` requires an explicit PR/branch/url argument.    # After resolving `--pr auto`, reuse the concrete PR number.    checks = get_pr_checks(str(pr["number"]), repo=pr["repo"])    checks_summary = summarize_checks(checks)    workflow_runs = get_workflow_runs_for_sha(pr["repo"], pr["head_sha"])    failed_runs = failed_runs_from_workflow_runs(workflow_runs, pr["head_sha"])    failed_jobs = failed_jobs_from_workflow_runs(        pr["repo"], workflow_runs, pr["head_sha"]    )     retries_used = current_retry_count(state, pr["head_sha"])    actions = recommend_actions(        pr,        checks_summary,        failed_runs,        failed_jobs,        new_review_items,        retries_used,        args.max_flaky_retries,    )     state["pr"] = {"repo": pr["repo"], "number": pr["number"]}    state["last_seen_head_sha"] = pr["head_sha"]    state["last_snapshot_at"] = int(time.time())    save_state(state_path, state)     snapshot = {        "pr": pr,        "checks": checks_summary,        "failed_runs": failed_runs,        "failed_jobs": failed_jobs,        "new_review_items": new_review_items,        "actions": actions,        "retry_state": {            "current_sha_retries_used": retries_used,            "max_flaky_retries": args.max_flaky_retries,        },    }    return snapshot, state_path  def retry_failed_now(args):    snapshot, state_path = collect_snapshot(args)    pr = snapshot["pr"]    checks_summary = snapshot["checks"]    failed_runs = snapshot["failed_runs"]    retries_used = snapshot["retry_state"]["current_sha_retries_used"]    max_retries = snapshot["retry_state"]["max_flaky_retries"]     result = {        "snapshot": snapshot,        "state_file": str(state_path),        "rerun_attempted": False,        "rerun_count": 0,        "rerun_run_ids": [],        "reason": None,    }     if pr["closed"] or pr["merged"]:        result["reason"] = "pr_closed"        return result    if checks_summary["failed_count"] <= 0:        result["reason"] = "no_failed_pr_checks"        return result    if not failed_runs:        result["reason"] = "no_failed_runs"        return result    if not checks_summary["all_terminal"]:        result["reason"] = "checks_still_pending"        return result    if retries_used >= max_retries:        result["reason"] = "retry_budget_exhausted"        return result     for run in failed_runs:        run_id = run.get("run_id")        if run_id in (None, ""):            continue        gh_text(["run", "rerun", str(run_id), "--failed"], repo=pr["repo"])        result["rerun_run_ids"].append(run_id)     if result["rerun_run_ids"]:        state, _ = load_state(state_path)        new_count = current_retry_count(state, pr["head_sha"]) + 1        set_retry_count(state, pr["head_sha"], new_count)        state["last_snapshot_at"] = int(time.time())        save_state(state_path, state)        result["rerun_attempted"] = True        result["rerun_count"] = len(result["rerun_run_ids"])        result["reason"] = "rerun_triggered"    else:        result["reason"] = "failed_runs_missing_ids"     return result  def print_json(obj):    sys.stdout.write(json.dumps(obj, sort_keys=True) + "\n")    sys.stdout.flush()  def print_event(event, payload):    print_json({"event": event, "payload": payload})  def is_ci_green(snapshot):    checks = snapshot.get("checks") or {}    return (        bool(checks.get("all_terminal"))        and int(checks.get("failed_count") or 0) == 0        and int(checks.get("pending_count") or 0) == 0    )  def snapshot_change_key(snapshot):    pr = snapshot.get("pr") or {}    checks = snapshot.get("checks") or {}    review_items = snapshot.get("new_review_items") or []    return (        str(pr.get("head_sha") or ""),        str(pr.get("state") or ""),        str(pr.get("mergeable") or ""),        str(pr.get("merge_state_status") or ""),        str(pr.get("review_decision") or ""),        int(checks.get("passed_count") or 0),        int(checks.get("failed_count") or 0),        int(checks.get("pending_count") or 0),        tuple(            (str(item.get("kind") or ""), str(item.get("id") or ""))            for item in review_items            if isinstance(item, dict)        ),        tuple(snapshot.get("actions") or []),    )  def run_watch(args):    poll_seconds = args.poll_seconds    last_change_key = None    while True:        snapshot, state_path = collect_snapshot(args)        print_event(            "snapshot",            {                "snapshot": snapshot,                "state_file": str(state_path),                "next_poll_seconds": poll_seconds,            },        )        actions = set(snapshot.get("actions") or [])        if "stop_pr_closed" in actions or "stop_exhausted_retries" in actions:            print_event(                "stop", {"actions": snapshot.get("actions"), "pr": snapshot.get("pr")}            )            return 0         current_change_key = snapshot_change_key(snapshot)        changed = current_change_key != last_change_key        green = is_ci_green(snapshot)        pr = snapshot.get("pr") or {}        pr_open = not bool(pr.get("closed")) and not bool(pr.get("merged"))         if not green or pr_open:            poll_seconds = args.poll_seconds        elif changed or last_change_key is None:            poll_seconds = args.poll_seconds         last_change_key = current_change_key        time.sleep(poll_seconds)  def main():    args = parse_args()    try:        if args.retry_failed_now:            print_json(retry_failed_now(args))            return 0        if args.watch:            return run_watch(args)        snapshot, state_path = collect_snapshot(args)        snapshot["state_file"] = str(state_path)        print_json(snapshot)        return 0    except (GhCommandError, RuntimeError, ValueError) as err:        sys.stderr.write(f"gh_pr_watch.py error: {err}\n")        return 1    except KeyboardInterrupt:        sys.stderr.write("gh_pr_watch.py interrupted\n")        return 130  if __name__ == "__main__":    raise SystemExit(main())
← All Codex skillsProject instructions →