diff --git a/scripts/ci/pr_review_merge_scheduler.py b/scripts/ci/pr_review_merge_scheduler.py index 75e18c860..7371f44d2 100644 --- a/scripts/ci/pr_review_merge_scheduler.py +++ b/scripts/ci/pr_review_merge_scheduler.py @@ -18,7 +18,6 @@ from typing import Any from urllib.parse import quote - PULL_REQUEST_FIELDS_FRAGMENT = """\ fragment SchedulerPullRequestFields on PullRequest { number @@ -123,8 +122,21 @@ "Required OpenCode Review", "OpenCode Review Dispatch", } -RUNNING_CHECK_STATES = {"PENDING", "EXPECTED", "QUEUED", "IN_PROGRESS", "WAITING", "REQUESTED"} -FAILED_CHECK_CONCLUSIONS = {"FAILURE", "ERROR", "CANCELLED", "TIMED_OUT", "STARTUP_FAILURE"} +RUNNING_CHECK_STATES = { + "PENDING", + "EXPECTED", + "QUEUED", + "IN_PROGRESS", + "WAITING", + "REQUESTED", +} +FAILED_CHECK_CONCLUSIONS = { + "FAILURE", + "ERROR", + "CANCELLED", + "TIMED_OUT", + "STARTUP_FAILURE", +} ACTION_REQUIRED_CONCLUSIONS = {"ACTION_REQUIRED"} GIT_REF_RE = re.compile(r"^(?!-)[A-Za-z0-9._/-]+$") GIT_SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$") @@ -181,14 +193,27 @@ class Decision: SENSITIVE_DATA_SCRUB_PATTERNS = ( - (re.compile(r'(?i)(bearer\s+)[^\s"\'\\]+'), r'\1***'), - (re.compile(r'(?i)(token\s+)[^\s"\'\\]+'), r'\1***'), - (re.compile(r'(?i)\b(?:github_pat_[A-Za-z0-9_]+|gh[pousr]_[A-Za-z0-9_]+)\b'), '***'), - (re.compile(r'\b(sk-[A-Za-z0-9_-]+)'), '***'), - (re.compile(r'\b(xox[baprs]-[A-Za-z0-9-]+)'), '***'), - (re.compile(r'\b(AKIA[0-9A-Z]{16})'), '***'), - (re.compile(r'(?i)((?:api[_-]?key|access[_-]?token|refresh[_-]?token|id[_-]?token|client[_-]?secret|password|passwd|secret)\s*[:=]\s*)["\']?[^"\'\s]+["\']?'), r'\1***'), - (re.compile(r'(?i)((?:authorization|proxy-authorization)\s*:\s*(?:bearer|basic)\s+)[A-Za-z0-9._~+\/=-]+'), r'\1***'), + (re.compile(r'(?i)(bearer\s+)[^\s"\'\\]+'), r"\1***"), + (re.compile(r'(?i)(token\s+)[^\s"\'\\]+'), r"\1***"), + ( + re.compile(r"(?i)\b(?:github_pat_[A-Za-z0-9_]+|gh[pousr]_[A-Za-z0-9_]+)\b"), + "***", + ), + (re.compile(r"\b(sk-[A-Za-z0-9_-]+)"), "***"), + (re.compile(r"\b(xox[baprs]-[A-Za-z0-9-]+)"), "***"), + (re.compile(r"\b(AKIA[0-9A-Z]{16})"), "***"), + ( + re.compile( + r'(?i)((?:api[_-]?key|access[_-]?token|refresh[_-]?token|id[_-]?token|client[_-]?secret|password|passwd|secret)\s*[:=]\s*)["\']?[^"\'\s]+["\']?' + ), + r"\1***", + ), + ( + re.compile( + r"(?i)((?:authorization|proxy-authorization)\s*:\s*(?:bearer|basic)\s+)[A-Za-z0-9._~+\/=-]+" + ), + r"\1***", + ), ) @@ -203,7 +228,9 @@ def scrub_sensitive_data(text: str | None) -> str | None: def mutation_token_source() -> str: """Return the configured scheduler mutation credential source.""" - return (os.environ.get("SCHEDULER_MUTATION_TOKEN_SOURCE") or "github-token").strip() or "github-token" + return ( + os.environ.get("SCHEDULER_MUTATION_TOKEN_SOURCE") or "github-token" + ).strip() or "github-token" def mutation_token_label() -> str: @@ -232,11 +259,20 @@ def contract_decision(decision: Decision) -> str: """Map scheduler actions into the bounded PR decision contract.""" if decision.action in {"update_branch", "restamp_head"}: return "UPDATE_BRANCH" - if decision.action in {"wait", "security_dispatch", "review_dispatch", "disable_auto_merge", "action_error"}: + if decision.action in { + "wait", + "security_dispatch", + "review_dispatch", + "disable_auto_merge", + "action_error", + }: return "WAIT" if decision.action in {"skip", "auto_merge", "merge"}: return "NO_ACTION" - if decision.action == "block" and "current-head OpenCode review requested changes" in decision.reason: + if ( + decision.action == "block" + and "current-head OpenCode review requested changes" in decision.reason + ): return "REQUEST_CHANGES" return "WAIT" @@ -433,10 +469,14 @@ def run(args: Sequence[str], *, stdin: str | None = None) -> str: return run_with_env(args, stdin=stdin) -def run_with_env(args: Sequence[str], *, stdin: str | None = None, env: dict[str, str] | None = None) -> str: +def run_with_env( + args: Sequence[str], *, stdin: str | None = None, env: dict[str, str] | None = None +) -> str: """Run a command with an optional environment override and scrub failures.""" if isinstance(args, str) or not all(isinstance(arg, str) for arg in args): - raise TypeError("run() requires a sequence of argv strings; shell command strings are not allowed") + raise TypeError( + "run() requires a sequence of argv strings; shell command strings are not allowed" + ) argv = list(args) try: process = subprocess.run( @@ -449,7 +489,7 @@ def run_with_env(args: Sequence[str], *, stdin: str | None = None, env: dict[str env=env, ) except subprocess.CalledProcessError as exc: - scrubbed_args = scrub_sensitive_data(' '.join(argv)) + scrubbed_args = scrub_sensitive_data(" ".join(argv)) scrubbed_stderr = scrub_sensitive_data(exc.stderr or "") raise RuntimeError( f"Command failed ({exc.returncode}): {scrubbed_args}\n{scrubbed_stderr}" @@ -579,7 +619,9 @@ def repository_dispatch_target(repo: str) -> str: default branch, so callers cannot select a privileged workflow ref. """ target_repo = validate_github_repository(repo) - dispatch_repo = (os.environ.get("SCHEDULER_REQUIRED_WORKFLOW_REPOSITORY") or "").strip() + dispatch_repo = ( + os.environ.get("SCHEDULER_REQUIRED_WORKFLOW_REPOSITORY") or "" + ).strip() if not dispatch_repo: return target_repo return validate_github_repository(dispatch_repo) @@ -593,11 +635,15 @@ def env_flag_enabled(name: str) -> bool: def repository_dispatch_wait_reason(repo: str, workflow: str) -> str | None: """Explain why cross-repository required repository dispatch should wait.""" target_repo = validate_github_repository(repo) - dispatch_repo = (os.environ.get("SCHEDULER_REQUIRED_WORKFLOW_REPOSITORY") or "").strip() + dispatch_repo = ( + os.environ.get("SCHEDULER_REQUIRED_WORKFLOW_REPOSITORY") or "" + ).strip() if not dispatch_repo: return None dispatch_repo = validate_github_repository(dispatch_repo) - if dispatch_repo == target_repo or env_flag_enabled("SCHEDULER_ALLOW_CROSS_REPO_REPOSITORY_DISPATCH"): + if dispatch_repo == target_repo or env_flag_enabled( + "SCHEDULER_ALLOW_CROSS_REPO_REPOSITORY_DISPATCH" + ): return None execution_repo = (os.environ.get("GITHUB_REPOSITORY") or "").strip() if os.environ.get("SCHEDULER_DISPATCH_TOKEN") and execution_repo == dispatch_repo: @@ -642,7 +688,10 @@ def is_transient_github_api_error(exc: Exception) -> bool: return True message = str(exc) folded = message.lower() - return any(marker in message or marker.lower() in folded for marker in TRANSIENT_GITHUB_API_ERRORS) + return any( + marker in message or marker.lower() in folded + for marker in TRANSIENT_GITHUB_API_ERRORS + ) def gh_graphql(query: str, **fields: str | int) -> dict[str, Any]: @@ -652,7 +701,9 @@ def gh_graphql(query: str, **fields: str | int) -> dict[str, Any]: flag = "-F" if isinstance(value, int) else "-f" cmd.extend([flag, f"{key}={value}"]) max_attempts = 4 - for attempt in range(1, max_attempts + 1): # pragma: no branch - last failed attempt always raises + for attempt in range( + 1, max_attempts + 1 + ): # pragma: no branch - last failed attempt always raises try: return json.loads(run_github_read(cmd, stdin=query)) except (RuntimeError, json.JSONDecodeError) as exc: @@ -699,7 +750,9 @@ def rest_check_node(check: dict[str, Any]) -> dict[str, Any]: "__typename": "CheckRun", "name": check.get("name"), "status": (check.get("status") or "").upper(), - "conclusion": (check.get("conclusion") or "").upper() if check.get("conclusion") else None, + "conclusion": ( + (check.get("conclusion") or "").upper() if check.get("conclusion") else None + ), "startedAt": check.get("started_at"), "detailsUrl": check.get("details_url"), "checkSuite": {"workflowRun": {"workflow": {}}}, @@ -714,7 +767,9 @@ def rest_pr_node(repo: str, pr: dict[str, Any]) -> dict[str, Any]: base = pr.get("base") or {} head_repo = head.get("repo") or {} reviews = gh_api_json(f"repos/{repo}/pulls/{number}/reviews?per_page=100") - checks = gh_api_json(f"repos/{repo}/commits/{head.get('sha')}/check-runs?per_page=100") + checks = gh_api_json( + f"repos/{repo}/commits/{head.get('sha')}/check-runs?per_page=100" + ) files = gh_api_json(f"repos/{repo}/pulls/{number}/files?per_page=20") rest_merge_state = REST_MERGEABLE_STATE_MAP.get( str(pr.get("mergeable_state") or "").lower(), @@ -731,18 +786,22 @@ def rest_pr_node(repo: str, pr: dict[str, Any]) -> dict[str, Any]: "baseRefOid": base.get("sha"), "headRefName": head.get("ref"), "headRefOid": head.get("sha"), - "isCrossRepository": (head_repo.get("full_name") or repo).lower() != repo.lower(), + "isCrossRepository": (head_repo.get("full_name") or repo).lower() + != repo.lower(), "maintainerCanModify": bool(pr.get("maintainer_can_modify")), "headRepository": {"nameWithOwner": head_repo.get("full_name") or repo}, "autoMergeRequest": pr.get("auto_merge"), "reviewThreads": {"nodes": []}, - "files": {"nodes": [{"path": file.get("filename")} for file in files if file.get("filename")]}, + "files": { + "nodes": [ + {"path": file.get("filename")} for file in files if file.get("filename") + ] + }, "reviews": {"nodes": [rest_review_node(review) for review in reviews]}, "statusCheckRollup": { "contexts": { "nodes": [ - rest_check_node(check) - for check in (checks.get("check_runs") or []) + rest_check_node(check) for check in (checks.get("check_runs") or []) ] } }, @@ -750,7 +809,9 @@ def rest_pr_node(repo: str, pr: dict[str, Any]) -> dict[str, Any]: } -def fetch_open_prs_rest(repo: str, max_prs: int, base_branch: str | None = None) -> list[dict[str, Any]]: +def fetch_open_prs_rest( + repo: str, max_prs: int, base_branch: str | None = None +) -> list[dict[str, Any]]: """Fetch open pull requests through REST when GraphQL is unavailable.""" prs: list[dict[str, Any]] = [] @@ -770,9 +831,13 @@ def fetch_open_prs_rest(repo: str, max_prs: int, base_branch: str | None = None) prs.extend(rest_pr_node(repo, pr) for pr in payload) # pragma: no cover else: max_workers = min(REST_MERGEABLE_STATE_WORKERS, len(payload)) - with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + with concurrent.futures.ThreadPoolExecutor( + max_workers=max_workers + ) as executor: # Keep original API sort order - prs.extend(list(executor.map(lambda pr: rest_pr_node(repo, pr), payload))) + prs.extend( + list(executor.map(lambda pr: rest_pr_node(repo, pr), payload)) + ) if len(payload) < page_size: break page += 1 @@ -840,7 +905,7 @@ def fetch_rest_mergeable_state(repo: str, number: int) -> str: "api", f"repos/{repo}/pulls/{number}", "--jq", - ".mergeable_state // \"\"", + '.mergeable_state // ""', ] ).strip() return REST_MERGEABLE_STATE_MAP.get(raw_state.lower(), raw_state.upper()) @@ -877,7 +942,9 @@ def enrich_rest_mergeable_states(repo: str, prs: list[dict[str, Any]]) -> None: def enrich(pr: dict[str, Any]) -> None: """Attach REST mergeability evidence to one pull request payload.""" try: - pr["restMergeableState"] = fetch_rest_mergeable_state(repo, int(pr["number"])) + pr["restMergeableState"] = fetch_rest_mergeable_state( + repo, int(pr["number"]) + ) except RuntimeError as exc: pr["restMergeableStateError"] = bounded_error_summary(str(exc)) try: @@ -945,21 +1012,22 @@ def is_opencode_context(node: dict[str, Any]) -> bool: # status. Organization required-workflow CheckRuns are deliberately # non-authoritative placeholders and must not suppress that dispatch. return False - workflow = ( - ((node.get("checkSuite") or {}).get("workflowRun") or {}).get("workflow") - or {} + workflow = ((node.get("checkSuite") or {}).get("workflowRun") or {}).get( + "workflow" + ) or {} + return ( + node.get("name") == "opencode-review" + or workflow.get("name") in OPENCODE_WORKFLOW_NAMES ) - return node.get("name") == "opencode-review" or workflow.get("name") in OPENCODE_WORKFLOW_NAMES return node.get("context") == "opencode-review" def is_strix_context(node: dict[str, Any]) -> bool: """Return whether a check or status context belongs to Strix evidence.""" if node.get("__typename") == "CheckRun": - workflow = ( - ((node.get("checkSuite") or {}).get("workflowRun") or {}).get("workflow") - or {} - ) + workflow = ((node.get("checkSuite") or {}).get("workflowRun") or {}).get( + "workflow" + ) or {} workflow_name = workflow.get("name") return workflow_name in {"Strix Security Scan", "Strix"} or ( node.get("name") == "strix" and workflow_name is None @@ -1055,9 +1123,15 @@ def opencode_progress_state( return "complete" if saw_complete else "absent" -def opencode_in_progress(pr: dict[str, Any], *, stale_after_minutes: int | None = None) -> bool: +def opencode_in_progress( + pr: dict[str, Any], *, stale_after_minutes: int | None = None +) -> bool: """Return whether any OpenCode review status for the PR is still actively running.""" - stale_after = DEFAULT_STALE_OPENCODE_MINUTES if stale_after_minutes is None else stale_after_minutes + stale_after = ( + DEFAULT_STALE_OPENCODE_MINUTES + if stale_after_minutes is None + else stale_after_minutes + ) return opencode_progress_state(pr, stale_after_minutes=stale_after) == "running" @@ -1078,17 +1152,23 @@ def strix_evidence_state(pr: dict[str, Any]) -> str: def unresolved_thread_count(pr: dict[str, Any]) -> int: """Count active, non-outdated unresolved review threads on a PR.""" - threads = ((pr.get("reviewThreads") or {}).get("nodes") or []) - return sum(1 for thread in threads if not thread.get("isResolved") and not thread.get("isOutdated")) + threads = (pr.get("reviewThreads") or {}).get("nodes") or [] + return sum( + 1 + for thread in threads + if not thread.get("isResolved") and not thread.get("isOutdated") + ) def outdated_thread_ids(pr: dict[str, Any]) -> list[str]: """Return unresolved review-thread IDs GitHub already marks outdated.""" - threads = ((pr.get("reviewThreads") or {}).get("nodes") or []) + threads = (pr.get("reviewThreads") or {}).get("nodes") or [] return [ thread["id"] for thread in threads - if thread.get("id") and not thread.get("isResolved") and thread.get("isOutdated") + if thread.get("id") + and not thread.get("isResolved") + and thread.get("isOutdated") ] @@ -1115,7 +1195,9 @@ def resolve_outdated_review_threads(pr: dict[str, Any], *, dry_run: bool) -> int return len(thread_ids) -def with_outdated_thread_cleanup_note(decision: Decision, count: int, *, dry_run: bool) -> Decision: +def with_outdated_thread_cleanup_note( + decision: Decision, count: int, *, dry_run: bool +) -> Decision: """Annotate a decision with the outdated-thread cleanup side effect.""" if count <= 0: return decision @@ -1124,7 +1206,9 @@ def with_outdated_thread_cleanup_note(decision: Decision, count: int, *, dry_run f"{verb} {count} outdated review thread(s) before active unresolved-thread checks; " "outdated diff comments are not current-head review blockers." ) - return Decision(decision.pr, decision.action, decision.reason, (*decision.notes, note)) + return Decision( + decision.pr, decision.action, decision.reason, (*decision.notes, note) + ) def review_author_login(review: dict[str, Any]) -> str: @@ -1140,9 +1224,10 @@ def is_opencode_review(review: dict[str, Any]) -> bool: def is_legacy_actions_opencode_review(review: dict[str, Any]) -> bool: """Return whether a legacy Actions-authored review contains OpenCode evidence.""" login = review_author_login(review) - return login in {"github-actions", "github-actions[bot]"} and "opencode" in ( - review.get("body") or "" - ).lower() + return ( + login in {"github-actions", "github-actions[bot]"} + and "opencode" in (review.get("body") or "").lower() + ) def is_automated_opencode_review(review: dict[str, Any]) -> bool: @@ -1256,15 +1341,19 @@ def dismiss_pull_request_review( f"message={message}", ] ) - live_state = run_github_read( - [ - "gh", - "api", - f"repos/{repo}/pulls/{number}/reviews/{review_id}", - "--jq", - ".state", - ] - ).strip().upper() + live_state = ( + run_github_read( + [ + "gh", + "api", + f"repos/{repo}/pulls/{number}/reviews/{review_id}", + "--jq", + ".state", + ] + ) + .strip() + .upper() + ) except RuntimeError as exc: print( "::warning::Stale OpenCode review dismissal failed for " @@ -1319,12 +1408,16 @@ def dismiss_stale_opencode_approvals( return dismissed, len(review_ids) - dismissed -def stale_approval_cleanup_note(dismissed: int, retained: int, *, dry_run: bool) -> str | None: +def stale_approval_cleanup_note( + dismissed: int, retained: int, *, dry_run: bool +) -> str | None: """Render exact stale-approval cleanup evidence for scheduler logs.""" notes: list[str] = [] if dismissed: verb = "would dismiss" if dry_run else "dismissed" - notes.append(f"{verb} {dismissed} latest previous-head automated OpenCode approval(s)") + notes.append( + f"{verb} {dismissed} latest previous-head automated OpenCode approval(s)" + ) if retained: notes.append( f"GitHub retained {retained} stale automated approval(s) after dismissal attempts; " @@ -1333,7 +1426,9 @@ def stale_approval_cleanup_note(dismissed: int, retained: int, *, dry_run: bool) return "; ".join(notes) if notes else None -def dismiss_stale_opencode_change_requests(repo: str, pr: dict[str, Any], *, dry_run: bool) -> int: +def dismiss_stale_opencode_change_requests( + repo: str, pr: dict[str, Any], *, dry_run: bool +) -> int: """Dismiss previous-head automated gates only after exact-head approval.""" if not has_current_head_approval(pr): return 0 @@ -1388,9 +1483,9 @@ def failed_status_checks(pr: dict[str, Any]) -> list[str]: status_contexts.append(node) continue workflow = ( - (((node.get("checkSuite") or {}).get("workflowRun") or {}).get("workflow") or {}).get("name") - or "" - ) + ((node.get("checkSuite") or {}).get("workflowRun") or {}).get("workflow") + or {} + ).get("name") or "" key = (workflow, node.get("name") or "check-run") started_at = parse_github_datetime(node.get("startedAt")) previous = latest_check_runs.get(key) @@ -1419,7 +1514,10 @@ def failed_status_checks(pr: dict[str, Any]) -> list[str]: if conclusion in FAILED_CHECK_CONCLUSIONS: if is_strix_context(node) and "strix" in successful_status_contexts: continue - if is_opencode_context(node) and "opencode-review" in successful_status_contexts: + if ( + is_opencode_context(node) + and "opencode-review" in successful_status_contexts + ): continue failed.append(node.get("name") or "check-run") for node in status_contexts: @@ -1444,7 +1542,9 @@ def action_required_checks(pr: dict[str, Any]) -> list[str]: def workflow_action_required_reason(checks: list[str]) -> str: """Return a scheduler reason for ACTION_REQUIRED check runs.""" visible = checks[:5] - suffix = f", +{len(checks) - len(visible)} more" if len(checks) > len(visible) else "" + suffix = ( + f", +{len(checks) - len(visible)} more" if len(checks) > len(visible) else "" + ) return ( f"workflow action required: {', '.join(visible)}{suffix}; " "approve or unblock the GitHub Actions run before treating checks as failed or passed" @@ -1545,7 +1645,9 @@ def disable_auto_merge_decision( ) -> Decision: """Disable auto-merge and return a WAIT decision with the concrete unsafe reason.""" disable_auto_merge(repo, pr, dry_run=dry_run) - return Decision(pr["number"], "disable_auto_merge", f"auto-merge disabled; {reason}") + return Decision( + pr["number"], "disable_auto_merge", f"auto-merge disabled; {reason}" + ) def update_branch(repo: str, pr: dict[str, Any], *, dry_run: bool) -> None: @@ -1614,19 +1716,25 @@ def last_push_approval_block_reason() -> str: ) -def restamp_pr_head_for_last_push_approval(repo: str, pr: dict[str, Any], *, dry_run: bool) -> str | None: +def restamp_pr_head_for_last_push_approval( + repo: str, pr: dict[str, Any], *, dry_run: bool +) -> str | None: """Create a same-tree child commit and move the PR head with a force=false ref update.""" if dry_run: return None require_github_actions_mutation_actor("last-push-approval-head-refresh") repo = validate_github_repository(repo) if not same_repository_head(repo, pr): - raise RuntimeError("last-push approval head refresh only supports same-repository PR heads") + raise RuntimeError( + "last-push approval head refresh only supports same-repository PR heads" + ) number = str(int(pr["number"])) head = validate_git_sha(pr["headRefOid"]) head_ref = validate_git_ref(pr["headRefName"]) - live_head = run(["gh", "api", f"repos/{repo}/pulls/{number}", "--jq", ".head.sha"]).strip() + live_head = run( + ["gh", "api", f"repos/{repo}/pulls/{number}", "--jq", ".head.sha"] + ).strip() if live_head != head: raise RuntimeError( "PR head changed before last-push approval head refresh; " @@ -1650,7 +1758,15 @@ def restamp_pr_head_for_last_push_approval(repo: str, pr: dict[str, Any], *, dry ) new_head = validate_git_sha(str(created_commit.get("sha") or "")) run( - ["gh", "api", "-X", "PATCH", f"repos/{repo}/git/refs/heads/{head_ref}", "--input", "-"], + [ + "gh", + "api", + "-X", + "PATCH", + f"repos/{repo}/git/refs/heads/{head_ref}", + "--input", + "-", + ], stdin=json.dumps({"sha": new_head, "force": False}), ) return new_head @@ -1750,14 +1866,18 @@ def post_update_branch_followup( if strix_state == "running": return f"{head_note}; same-head Strix evidence is already running" - opencode_state = opencode_progress_state(updated_pr, stale_after_minutes=stale_opencode_minutes) + opencode_state = opencode_progress_state( + updated_pr, stale_after_minutes=stale_opencode_minutes + ) if opencode_state == "running": return f"{head_note}; same-head OpenCode review is already running" wait_reason = repository_dispatch_wait_reason(repo, workflow) if wait_reason: return f"{head_note}; {wait_reason}" - dispatch_result = dispatch_opencode_review(repo, workflow, updated_pr, dry_run=dry_run) + dispatch_result = dispatch_opencode_review( + repo, workflow, updated_pr, dry_run=dry_run + ) if dispatch_result == "already_running": return f"{head_note}; same-head OpenCode workflow run is already active" return f"{head_note}; same-head Strix evidence is complete, so OpenCode review was dispatched" @@ -1832,10 +1952,14 @@ def rerun_actions_job(repo: str, job_id: str, *, dry_run: bool, action: str) -> if dry_run: return require_github_actions_control_actor(action) - run_github_actions(["gh", "api", "-X", "POST", f"repos/{repo}/actions/jobs/{job_id}/rerun"]) + run_github_actions( + ["gh", "api", "-X", "POST", f"repos/{repo}/actions/jobs/{job_id}/rerun"] + ) -def active_workflow_runs(repo: str, statuses: Sequence[str] = ("queued", "in_progress")) -> list[dict[str, Any]]: +def active_workflow_runs( + repo: str, statuses: Sequence[str] = ("queued", "in_progress") +) -> list[dict[str, Any]]: """Return active workflow runs for a repository.""" runs: list[dict[str, Any]] = [] for status in statuses: @@ -1860,7 +1984,9 @@ def active_workflow_runs(repo: str, statuses: Sequence[str] = ("queued", "in_pro def workflow_run_mentions_pr(run_data: dict[str, Any], pr_number: int) -> bool: """Return whether a workflow run is attached to the pull request number.""" - return any(pr.get("number") == pr_number for pr in run_data.get("pull_requests") or []) + return any( + pr.get("number") == pr_number for pr in run_data.get("pull_requests") or [] + ) def stale_pr_run_ids( @@ -1940,7 +2066,9 @@ def active_review_run_refs( None, ) if run_data.get("event") == "repository_dispatch" and dispatch_title_prefix: - dispatched_head = display_title.removeprefix(dispatch_title_prefix).lower() + dispatched_head = display_title.removeprefix( + dispatch_title_prefix + ).lower() if not GIT_SHA_RE.fullmatch(dispatched_head): continue (current if dispatched_head == head else stale).append(run_ref) @@ -2028,7 +2156,9 @@ def cancel_one(run_id: str) -> tuple[str, str | None]: else: max_workers = min(REST_MERGEABLE_STATE_WORKERS, len(run_ids)) with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: - results = list(executor.map(cancel_one, (str(run_id) for run_id in run_ids))) + results = list( + executor.map(cancel_one, (str(run_id) for run_id in run_ids)) + ) failures = {run_id: reason for run_id, reason in results if reason is not None} for run_id, reason in failures.items(): @@ -2059,7 +2189,9 @@ def cancel_stale_pr_runs(repo: str, pr: dict[str, Any], *, dry_run: bool) -> lis return run_ids -def cancel_stale_opencode_runs(repo: str, workflow: str, pr: dict[str, Any], *, dry_run: bool) -> list[str]: +def cancel_stale_opencode_runs( + repo: str, workflow: str, pr: dict[str, Any], *, dry_run: bool +) -> list[str]: """Force-cancel older OpenCode runs for the same PR before retrying current head.""" if dry_run: return [] @@ -2069,7 +2201,9 @@ def cancel_stale_opencode_runs(repo: str, workflow: str, pr: dict[str, Any], *, return [run_id for _, run_id in stale_refs] -def dispatch_opencode_review(repo: str, workflow: str, pr: dict[str, Any], *, dry_run: bool) -> str: +def dispatch_opencode_review( + repo: str, workflow: str, pr: dict[str, Any], *, dry_run: bool +) -> str: """Dispatch trusted OpenCode for the PR head, or report an active run. The review job is intentionally restricted to ``repository_dispatch``. A @@ -2123,7 +2257,9 @@ def dispatch_opencode_review(repo: str, workflow: str, pr: dict[str, Any], *, dr return "dispatched" -def dispatch_strix_evidence(repo: str, workflow: str, pr: dict[str, Any], *, dry_run: bool) -> str: +def dispatch_strix_evidence( + repo: str, workflow: str, pr: dict[str, Any], *, dry_run: bool +) -> str: """Dispatch same-head Strix workflow evidence before OpenCode reviews.""" job_id = matching_actions_job_id(pr, is_strix_context) if job_id: @@ -2143,9 +2279,7 @@ def dispatch_strix_evidence(repo: str, workflow: str, pr: dict[str, Any], *, dry if current_run_refs: print( "Strix evidence dispatch skipped: active same-head workflow run(s) " - + ", ".join( - f"{run_repo}@{run_id}" for run_repo, run_id in current_run_refs - ) + + ", ".join(f"{run_repo}@{run_id}" for run_repo, run_id in current_run_refs) ) return "already_running" base_ref, base_sha, head_sha = validated_pr_dispatch_fields(pr) @@ -2183,9 +2317,7 @@ def merge_conflict_guidance(pr: dict[str, Any], merge_state: str) -> str: head_ref = pr.get("headRefName") or "head" changed_files = conflict_changed_files_text(pr) changed_files_note = ( - f"changed files to inspect first: {changed_files}; " - if changed_files - else "" + f"changed files to inspect first: {changed_files}; " if changed_files else "" ) return ( f"merge conflict: {merge_state}; base={base_ref}, head={head_ref}; " @@ -2202,7 +2334,9 @@ def merge_conflict_guidance(pr: dict[str, Any], merge_state: str) -> str: def changed_file_paths(pr: dict[str, Any], *, limit: int = 10) -> list[str]: """Return changed file paths already present in the pull request payload.""" nodes = ((pr.get("files") or {}).get("nodes") or [])[:limit] - return [path for node in nodes if isinstance(path := node.get("path"), str) and path] + return [ + path for node in nodes if isinstance(path := node.get("path"), str) and path + ] def conflict_changed_files_text(pr: dict[str, Any], *, limit: int = 10) -> str: @@ -2277,12 +2411,22 @@ def inspect_pr( # PRs never receive an OpenCode review on their own — dispatch one here. # Merge automation stays default-branch-only; rulesets do not gate # feature-branch merges. - opencode_state = opencode_progress_state(pr, stale_after_minutes=stale_opencode_minutes) - if opencode_state in {"absent", "stale"} and trigger_reviews and review_dispatch_allowed: + opencode_state = opencode_progress_state( + pr, stale_after_minutes=stale_opencode_minutes + ) + if ( + opencode_state in {"absent", "stale"} + and trigger_reviews + and review_dispatch_allowed + ): wait_reason = repository_dispatch_wait_reason(repo, workflow) if wait_reason: - return Decision(number, "wait", f"stacked PR onto {base_ref}; {wait_reason}") - dispatch_result = dispatch_opencode_review(repo, workflow, pr, dry_run=dry_run) + return Decision( + number, "wait", f"stacked PR onto {base_ref}; {wait_reason}" + ) + dispatch_result = dispatch_opencode_review( + repo, workflow, pr, dry_run=dry_run + ) if dispatch_result == "already_running": return Decision( number, @@ -2302,10 +2446,12 @@ def inspect_pr( outdated_cleanup_count = resolve_outdated_review_threads(pr, dry_run=dry_run) stale_review_cleanup_count = 0 - stale_approval_cleanup_count, retained_stale_approval_count = dismiss_stale_opencode_approvals( - repo, - pr, - dry_run=dry_run, + stale_approval_cleanup_count, retained_stale_approval_count = ( + dismiss_stale_opencode_approvals( + repo, + pr, + dry_run=dry_run, + ) ) def finish(decision: Decision) -> Decision: @@ -2423,7 +2569,9 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio ) ) if not same_repository_head(repo, pr): - return decide("wait", f"{external_head_merge_reason(repo, pr)}; {conflict_reason}") + return decide( + "wait", f"{external_head_merge_reason(repo, pr)}; {conflict_reason}" + ) return decide( "block", "current head is approved, but auto-merge is not queued until merge conflict repair is pushed; " @@ -2480,16 +2628,25 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio if not enable_auto_merge_flag: if pr.get("autoMergeRequest"): return decide("wait", auto_merge_wait_reason(merge_state, pr)) - return decide("wait", "current head is approved; auto-merge disabled by scheduler inputs") + return decide( + "wait", + "current head is approved; auto-merge disabled by scheduler inputs", + ) if merge_mode == "disabled": if pr.get("autoMergeRequest"): return decide("wait", auto_merge_wait_reason(merge_state, pr)) - return decide("wait", "current head is approved; merge mode disabled by scheduler inputs") + return decide( + "wait", + "current head is approved; merge mode disabled by scheduler inputs", + ) if merge_mode in {"direct", "direct_or_auto"}: try: merge_pr(repo, pr, dry_run=dry_run) except RuntimeError as exc: - if merge_mode != "direct_or_auto" or not direct_merge_can_fallback_to_auto_merge(exc): + if ( + merge_mode != "direct_or_auto" + or not direct_merge_can_fallback_to_auto_merge(exc) + ): raise block_detail = direct_merge_block_detail(exc) if pr.get("autoMergeRequest"): @@ -2506,14 +2663,21 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio "so auto-merge was enabled with the same head guard evidence; " f"GitHub reported: {block_detail}", ) - state_note = "" if merge_state == "CLEAN" else f"; GitHub mergeability is {merge_state}" + state_note = ( + "" + if merge_state == "CLEAN" + else f"; GitHub mergeability is {merge_state}" + ) return decide( "merge", f"current head is approved; direct merge requested with {mutation_token_label()} " f"and --match-head-commit{state_note}", ) if merge_mode != "auto": - return decide("wait", f"current head is approved; unsupported merge mode: {merge_mode}") + return decide( + "wait", + f"current head is approved; unsupported merge mode: {merge_mode}", + ) if pr.get("autoMergeRequest"): return decide("wait", auto_merge_wait_reason(merge_state, pr)) enable_auto_merge(repo, pr, dry_run=dry_run) @@ -2523,11 +2687,16 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio if behind_by and (current_head_approved or auto_merge_enabled): if not update_branches: if current_head_approved: - return decide("wait", "current-head OpenCode review approved; branch update disabled") + return decide( + "wait", + "current-head OpenCode review approved; branch update disabled", + ) return decide("wait", "auto-merge already enabled; branch update disabled") if not can_update_pr_head(repo, pr): return decide("wait", non_mutable_head_reason(repo, pr)) - suffix = "; existing auto-merge request remains queued" if auto_merge_enabled else "" + suffix = ( + "; existing auto-merge request remains queued" if auto_merge_enabled else "" + ) if current_head_approved and merge_state == "BEHIND": freshness_reason = "current-head OpenCode review approved" elif current_head_approved: @@ -2573,7 +2742,9 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio new_head = restamp_pr_head_for_last_push_approval(repo, pr, dry_run=dry_run) notes = () if new_head: - notes = (f"last-push approval head refresh created same-tree head {short_sha(new_head)}",) + notes = ( + f"last-push approval head refresh created same-tree head {short_sha(new_head)}", + ) return finish( Decision( number, @@ -2584,14 +2755,15 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio ) ) - opencode_state = opencode_progress_state(pr, stale_after_minutes=stale_opencode_minutes) + opencode_state = opencode_progress_state( + pr, stale_after_minutes=stale_opencode_minutes + ) if opencode_state == "running": return decide("wait", "OpenCode review is already in progress") - if ( - os.environ.get("GITHUB_EVENT_NAME") == "workflow_run" - and has_current_head_deterministic_fallback_approval(pr) - ): + if os.environ.get( + "GITHUB_EVENT_NAME" + ) == "workflow_run" and has_current_head_deterministic_fallback_approval(pr): return decide( "wait", "current-head deterministic fallback is not merge evidence; defer real-model retry to the next scheduler heartbeat", @@ -2599,9 +2771,14 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio if behind_by and trigger_reviews: if not update_branches: - return decide("wait", "current head has no OpenCode approval; branch update disabled before review dispatch") + return decide( + "wait", + "current head has no OpenCode approval; branch update disabled before review dispatch", + ) if not can_update_pr_head(repo, pr): - head_repo = (pr.get("headRepository") or {}).get("nameWithOwner") or "" + head_repo = (pr.get("headRepository") or {}).get( + "nameWithOwner" + ) or "" return decide( "wait", f"current head has no OpenCode approval; branch is outdated before review dispatch, " @@ -2627,7 +2804,10 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio reason="mergeability is still being calculated and no branch freshness evidence is available; wait for GitHub mergeability evidence before re-enabling auto-merge", ) ) - return decide("wait", "mergeability is still being calculated and no branch freshness evidence is available") + return decide( + "wait", + "mergeability is still being calculated and no branch freshness evidence is available", + ) if current_head_approved: if pr.get("autoMergeRequest"): @@ -2635,9 +2815,15 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio if not same_repository_head(repo, pr): return decide("wait", external_head_merge_reason(repo, pr)) if not enable_auto_merge_flag: - return decide("wait", "current head is approved; auto-merge disabled by scheduler inputs") + return decide( + "wait", + "current head is approved; auto-merge disabled by scheduler inputs", + ) if merge_mode == "disabled": - return decide("wait", "current head is approved; merge mode disabled by scheduler inputs") + return decide( + "wait", + "current head is approved; merge mode disabled by scheduler inputs", + ) if merge_mode in {"direct", "direct_or_auto"}: if merge_mode == "direct_or_auto": try: @@ -2663,7 +2849,10 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio f"current head is approved; direct merge waits for CLEAN mergeability; GitHub mergeability is {merge_state}", ) if merge_mode != "auto": - return decide("wait", f"current head is approved; unsupported merge mode: {merge_mode}") + return decide( + "wait", + f"current head is approved; unsupported merge mode: {merge_mode}", + ) enable_auto_merge(repo, pr, dry_run=dry_run) return decide("auto_merge", "current head is approved; auto-merge enabled") @@ -2699,7 +2888,10 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio ) wait_reason = repository_dispatch_wait_reason(repo, security_workflow) if wait_reason: - return decide("wait", f"current head has no completed Strix evidence; {wait_reason}") + return decide( + "wait", + f"current head has no completed Strix evidence; {wait_reason}", + ) dispatch_strix_evidence(repo, security_workflow, pr, dry_run=dry_run) return decide( "security_dispatch", @@ -2716,7 +2908,9 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio ) wait_reason = repository_dispatch_wait_reason(repo, workflow) if wait_reason: - return decide("wait", f"current head has completed Strix evidence; {wait_reason}") + return decide( + "wait", f"current head has completed Strix evidence; {wait_reason}" + ) dispatch_result = dispatch_opencode_review(repo, workflow, pr, dry_run=dry_run) if dispatch_result == "already_running": return decide( @@ -2862,7 +3056,9 @@ def parse_conflict_changed_files(reason: str) -> list[str]: continue return [ file_path - for file_path in (part.strip() for part in segment[len(prefix) :].split("|")) + for file_path in ( + part.strip() for part in segment[len(prefix) :].split("|") + ) if file_path and not file_path.startswith("+") ] return [] @@ -2870,8 +3066,12 @@ def parse_conflict_changed_files(reason: str) -> list[str]: def conflict_repair_summary(decisions: list[Decision]) -> list[str]: """Return a GitHub Actions Summary section with concrete conflict repair steps.""" - conflicted = [(decision, parse_conflict_reason(decision.reason)) for decision in decisions] - conflicted = [(decision, parsed) for decision, parsed in conflicted if parsed is not None] + conflicted = [ + (decision, parse_conflict_reason(decision.reason)) for decision in decisions + ] + conflicted = [ + (decision, parsed) for decision, parsed in conflicted if parsed is not None + ] if not conflicted: return [] @@ -2959,7 +3159,12 @@ def update_branch_summary(decisions: list[Decision]) -> list[str]: "When repository permissions allow the mutation, GitHub records the resulting branch update under the selected workflow credential.", "The updated head is not merge evidence by itself. Wait for the new head to receive OpenCode approval, Strix evidence, required checks, and unresolved-thread checks before merge or auto-merge.", ] - followups = [(decision, note) for decision in updates for note in decision.notes if "update-branch" in note] + followups = [ + (decision, note) + for decision in updates + for note in decision.notes + if "update-branch" in note + ] if followups: lines.extend(["", "Follow-up evidence:"]) lines.extend(f"- PR #{decision.pr}: {note}" for decision, note in followups) @@ -2973,7 +3178,11 @@ def parse_last_push_approval_restamp_reason(reason: str) -> bool: def last_push_approval_restamp_summary(decisions: list[Decision]) -> list[str]: """Return a summary section explaining last-push approval restamps.""" - restamps = [decision for decision in decisions if parse_last_push_approval_restamp_reason(decision.reason)] + restamps = [ + decision + for decision in decisions + if parse_last_push_approval_restamp_reason(decision.reason) + ] if not restamps: return [] token_label = mutation_token_label() @@ -3005,7 +3214,10 @@ def parse_external_head_update_reason(reason: str) -> str | None: def parse_external_head_merge_reason(reason: str) -> str | None: """Extract the external head repository from merge-exclusion guidance.""" - match = re.search(r"head repo ([^\s]+) is external; fork or external PR heads are excluded", reason) + match = re.search( + r"head repo ([^\s]+) is external; fork or external PR heads are excluded", + reason, + ) if not match: return None return match.group(1) @@ -3123,20 +3335,30 @@ def summarize_action_error(exc: RuntimeError) -> str: return "scheduler action failed without stderr" summary = "; ".join(lines[:2]) lower_summary = summary.lower() - if "without `workflows` permission" in lower_summary or "without workflows permission" in lower_summary: + if ( + "without `workflows` permission" in lower_summary + or "without workflows permission" in lower_summary + ): summary = ( f"{summary}; workflow-file PRs need a scheduler mutation credential with GitHub `workflows` permission. " "Configure `PR_REVIEW_MERGE_TOKEN` or expand the selected GitHub App permission, then rerun the scheduler; " "do not leave this as a review comment for the PR author." ) - if "auto-merge is disabled" in lower_summary or "auto merge is disabled" in lower_summary: + if ( + "auto-merge is disabled" in lower_summary + or "auto merge is disabled" in lower_summary + ): summary = ( f"{summary}; native auto-merge is disabled for this repository. " "Use `--merge-mode direct_or_auto` so the scheduler attempts a guarded direct merge before queueing native auto-merge, " "or enable repository auto-merge when branch policy requires GitHub's queued merge path." ) if "resource not accessible by integration" in lower_summary: - if "mergepullrequest" in lower_summary or "enablepullrequestautomerge" in lower_summary or "gh pr merge" in lower_summary: + if ( + "mergepullrequest" in lower_summary + or "enablepullrequestautomerge" in lower_summary + or "gh pr merge" in lower_summary + ): summary = ( f"{summary}; scheduler GitHub token could not perform merge or auto-merge. " "Merging through GitHub Actions needs an explicit repo policy exception for scheduler-job `contents: write`; otherwise leave auto-merge disabled and keep update-branch on the lower-privilege PR-write path." @@ -3151,15 +3373,15 @@ def summarize_action_error(exc: RuntimeError) -> str: f"{summary}; scheduler GitHub token lacks a required repository mutation permission. " "Fix the scheduler job permissions instead of posting a code-review finding." ) - if "expected_head_sha" in lower_summary and ("422" in lower_summary or "head" in lower_summary): - summary = ( - f"{summary}; the PR head likely changed after inspection. Rerun the scheduler so it reads the new head before mutating." - ) + if "expected_head_sha" in lower_summary and ( + "422" in lower_summary or "head" in lower_summary + ): + summary = f"{summary}; the PR head likely changed after inspection. Rerun the scheduler so it reads the new head before mutating." return bounded_error_summary(summary) -def self_test() -> None: - """Exercise scheduler invariants without GitHub network access.""" +def _test_split_repo() -> None: + """Test split_repo logic.""" assert split_repo("owner/name") == ("owner", "name") assert split_repo("owner/name/extra") == ("owner", "name/extra") try: @@ -3177,6 +3399,10 @@ def self_test() -> None: raise AssertionError("expected ValueError") except ValueError: pass + + +def _test_inspect_pr() -> None: + """Test inspect_pr logic on various states.""" sample = { "number": 1, "headRefOid": "abc", @@ -3256,7 +3482,10 @@ def self_test() -> None: base_branch="main", ) assert decision.action == "disable_auto_merge" - assert "merge conflict repair is required before auto-merge can be queued" in decision.reason + assert ( + "merge conflict repair is required before auto-merge can be queued" + in decision.reason + ) assert "merge conflict: DIRTY" in decision.reason sample["restMergeableState"] = "UNKNOWN" sample["autoMergeRequest"] = None @@ -3276,7 +3505,12 @@ def self_test() -> None: sample["restMergeableState"] = "CLEAN" sample["autoMergeRequest"] = {"enabledAt": "2026-01-01T00:02:00Z"} sample["statusCheckRollup"]["contexts"]["nodes"] = [ - {"__typename": "CheckRun", "name": "strix", "status": "COMPLETED", "conclusion": "FAILURE"} + { + "__typename": "CheckRun", + "name": "strix", + "status": "COMPLETED", + "conclusion": "FAILURE", + } ] decision = inspect_pr( "owner/repo", @@ -3381,7 +3615,9 @@ def self_test() -> None: "name": "strix", "status": "COMPLETED", "conclusion": "SUCCESS", - "checkSuite": {"workflowRun": {"workflow": {"name": "Strix Security Scan"}}}, + "checkSuite": { + "workflowRun": {"workflow": {"name": "Strix Security Scan"}} + }, } ] decision = inspect_pr( @@ -3457,7 +3693,12 @@ def self_test() -> None: ) assert decision.action == "update_branch" sample["statusCheckRollup"]["contexts"]["nodes"] = [ - {"__typename": "CheckRun", "name": "strix", "status": "COMPLETED", "conclusion": "FAILURE"} + { + "__typename": "CheckRun", + "name": "strix", + "status": "COMPLETED", + "conclusion": "FAILURE", + } ] decision = inspect_pr( "owner/repo", @@ -3502,7 +3743,10 @@ def self_test() -> None: base_branch="main", ) assert decision.action == "disable_auto_merge" - assert "merge conflict repair is required before auto-merge can be queued" in decision.reason + assert ( + "merge conflict repair is required before auto-merge can be queued" + in decision.reason + ) assert "merge conflict: DIRTY" in decision.reason conflict_guidance = decision_guidance(decision) assert conflict_guidance @@ -3520,7 +3764,10 @@ def self_test() -> None: base_branch="main", ) assert decision.action == "block" - assert "auto-merge is not queued until merge conflict repair is pushed" in decision.reason + assert ( + "auto-merge is not queued until merge conflict repair is pushed" + in decision.reason + ) sample["reviews"]["nodes"][0]["commit"]["oid"] = "old" decision = inspect_pr( "owner/repo", @@ -3546,6 +3793,10 @@ def self_test() -> None: assert conflict_guidance["merge_state"] == "DIRTY" assert "update-branch cannot choose" in conflict_guidance["automation_limit"] assert "git status --short" in conflict_guidance["commands"] + + +def _test_inspect_blocked_pr() -> None: + """Test inspect_pr logic on a blocked PR.""" blocked_sample = { "number": 2, "headRefOid": "abc", @@ -3615,8 +3866,13 @@ def self_test() -> None: restamp_guidance = decision_guidance(decision) assert restamp_guidance assert restamp_guidance["type"] == "last_push_approval_restamp" - assert restamp_guidance["head_guard"] == "live PR head check plus force=false Git ref update" - blocked_sample["commits"]["nodes"][0]["commit"]["messageHeadline"] = LAST_PUSH_APPROVAL_RESTAMP_MESSAGE + assert ( + restamp_guidance["head_guard"] + == "live PR head check plus force=false Git ref update" + ) + blocked_sample["commits"]["nodes"][0]["commit"][ + "messageHeadline" + ] = LAST_PUSH_APPROVAL_RESTAMP_MESSAGE decision = inspect_pr( "owner/repo", blocked_sample, @@ -3630,6 +3886,10 @@ def self_test() -> None: ) assert decision.action == "wait" assert "head refresh already exists" in decision.reason + + +def _test_contract_decision() -> None: + """Test contract_decision mapping.""" assert contract_decision(Decision(1, "update_branch", "ok")) == "UPDATE_BRANCH" assert contract_decision(Decision(1, "restamp_head", "ok")) == "UPDATE_BRANCH" assert contract_decision(Decision(1, "wait", "ok")) == "WAIT" @@ -3639,10 +3899,16 @@ def self_test() -> None: assert contract_decision(Decision(1, "merge", "ok")) == "NO_ACTION" assert contract_decision(Decision(1, "skip", "ok")) == "NO_ACTION" assert ( - contract_decision(Decision(1, "block", "current-head OpenCode review requested changes")) + contract_decision( + Decision(1, "block", "current-head OpenCode review requested changes") + ) == "REQUEST_CHANGES" ) assert contract_decision(Decision(1, "block", "merge conflict: DIRTY")) == "WAIT" + + +def _test_decision_guidance() -> None: + """Test decision_guidance mapping.""" update_guidance = decision_guidance(Decision(1, "update_branch", "ok")) assert update_guidance assert update_guidance["actor"] == "github-actions[bot]" @@ -3656,10 +3922,18 @@ def self_test() -> None: assert merge_guidance["head_guard"] == "gh pr merge --match-head-commit" assert decision_guidance(Decision(1, "wait", "ok")) is None restamp_guidance = decision_guidance( - Decision(1, "restamp_head", f"{last_push_approval_block_reason()}; last-push approval head refresh requested") + Decision( + 1, + "restamp_head", + f"{last_push_approval_block_reason()}; last-push approval head refresh requested", + ) ) assert restamp_guidance assert restamp_guidance["type"] == "last_push_approval_restamp" + + +def _test_decision_payload() -> None: + """Test decision_payload mapping.""" payload = decision_payload( [Decision(1, "update_branch", "ok")], counts={"update_branch": 1}, @@ -3671,7 +3945,13 @@ def self_test() -> None: assert payload["decisions"][0]["contract_decision"] == "UPDATE_BRANCH" assert payload["decisions"][0]["guidance"]["actor"] == "github-actions[bot]" payload = decision_payload( - [Decision(1, "restamp_head", f"{last_push_approval_block_reason()}; last-push approval head refresh requested")], + [ + Decision( + 1, + "restamp_head", + f"{last_push_approval_block_reason()}; last-push approval head refresh requested", + ) + ], counts={"restamp_head": 1}, dry_run=True, base_branch="main", @@ -3688,6 +3968,16 @@ def self_test() -> None: ) assert payload["decisions"][0]["contract_decision"] == "NO_ACTION" assert payload["decisions"][0]["guidance"]["type"] == "github_actions_direct_merge" + + +def self_test() -> None: + """Exercise scheduler invariants without GitHub network access.""" + _test_split_repo() + _test_inspect_pr() + _test_inspect_blocked_pr() + _test_contract_decision() + _test_decision_guidance() + _test_decision_payload() print("self-test passed") @@ -3700,7 +3990,9 @@ def parse_args(argv: list[str]) -> argparse.Namespace: parser.add_argument("--max-prs", type=int, default=100) parser.add_argument("--pr-number", type=int, default=0) parser.add_argument("--dry-run", action="store_true") - parser.add_argument("--trigger-reviews", action=argparse.BooleanOptionalAction, default=True) + parser.add_argument( + "--trigger-reviews", action=argparse.BooleanOptionalAction, default=True + ) parser.add_argument( "--review-dispatch-limit", type=int, @@ -3713,19 +4005,27 @@ def parse_args(argv: list[str]) -> argparse.Namespace: default=int(os.environ.get("BRANCH_UPDATE_LIMIT", "1")), help="Maximum update-branch mutations per scheduler run; -1 means unlimited", ) - parser.add_argument("--enable-auto-merge", action=argparse.BooleanOptionalAction, default=True) + parser.add_argument( + "--enable-auto-merge", action=argparse.BooleanOptionalAction, default=True + ) parser.add_argument( "--merge-mode", choices=("auto", "direct", "direct_or_auto", "disabled"), default=os.environ.get("MERGE_MODE", "direct_or_auto"), ) - parser.add_argument("--update-branches", action=argparse.BooleanOptionalAction, default=True) + parser.add_argument( + "--update-branches", action=argparse.BooleanOptionalAction, default=True + ) parser.add_argument("--review-workflow", default="Required OpenCode Review") parser.add_argument("--security-workflow", default="Strix Security Scan") parser.add_argument( "--stale-opencode-minutes", type=int, - default=int(os.environ.get("STALE_OPENCODE_MINUTES", str(DEFAULT_STALE_OPENCODE_MINUTES))), + default=int( + os.environ.get( + "STALE_OPENCODE_MINUTES", str(DEFAULT_STALE_OPENCODE_MINUTES) + ) + ), ) parser.add_argument("--self-test", action="store_true") return parser.parse_args(argv) @@ -3749,15 +4049,23 @@ def main(argv: list[str]) -> int: raise SystemExit("--review-dispatch-limit must be -1 or greater") if args.branch_update_limit < -1: raise SystemExit("--branch-update-limit must be -1 or greater") - prs = fetch_pr(args.repo, args.pr_number) if args.pr_number else fetch_open_prs(args.repo, args.max_prs) + prs = ( + fetch_pr(args.repo, args.pr_number) + if args.pr_number + else fetch_open_prs(args.repo, args.max_prs) + ) decisions = [] review_dispatches_used = 0 branch_updates_used = 0 for pr in prs: review_dispatch_allowed = ( - args.review_dispatch_limit < 0 or review_dispatches_used < args.review_dispatch_limit + args.review_dispatch_limit < 0 + or review_dispatches_used < args.review_dispatch_limit + ) + branch_update_allowed = ( + args.branch_update_limit < 0 + or branch_updates_used < args.branch_update_limit ) - branch_update_allowed = args.branch_update_limit < 0 or branch_updates_used < args.branch_update_limit try: decision = inspect_pr( args.repo,