diff --git a/scripts/ci/pr_review_merge_scheduler.py b/scripts/ci/pr_review_merge_scheduler.py index c13adfe9..ad0bd3bf 100644 --- a/scripts/ci/pr_review_merge_scheduler.py +++ b/scripts/ci/pr_review_merge_scheduler.py @@ -16,7 +16,7 @@ from dataclasses import dataclass from datetime import datetime, timezone from typing import Any -from urllib.parse import quote +from urllib.parse import quote, urlparse PULL_REQUEST_FIELDS_FRAGMENT = """\ @@ -127,6 +127,9 @@ GITHUB_REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") REVIEW_BODY_HEAD_SHA_RE = re.compile(r"Head SHA:\s*`([0-9a-fA-F]{40})`") ACTIONS_JOB_DETAILS_URL_RE = re.compile(r"/actions/runs/\d+/job/(\d+)(?:[/?#]|$)") +ACTIONS_RUN_DETAILS_URL_RE = re.compile(r"/actions/runs/(\d+)(?:[/?#]|$)") +EXTERNAL_HEAD_UPDATE_RE = re.compile(r"head repo ([^\s]+) is external and not writable") +EXTERNAL_HEAD_MERGE_RE = re.compile(r"head repo ([^\s]+) is external; fork or external PR heads are excluded") DIRECT_MERGE_AUTO_FALLBACK_MARKERS = ( "base branch policy prohibits the merge", "is not mergeable", @@ -155,6 +158,24 @@ "did not emit a usable current-head control block", ) LAST_PUSH_APPROVAL_RESTAMP_MESSAGE = "chore: refresh head for last-push approval" +TRUSTED_GITHUB_CLI = "/usr/bin/gh" + + +def github_cli_for_environment(github_actions: str | None) -> str: + """Pin gh inside Actions while retaining portable local inspection.""" + + return TRUSTED_GITHUB_CLI if (github_actions or "").lower() == "true" else "gh" + + +GITHUB_CLI = github_cli_for_environment(os.environ.get("GITHUB_ACTIONS")) +SCHEDULER_WORKFLOW_NAMES = { + "PR Review Merge Scheduler", + "Required PR Review Merge Scheduler", +} +SCHEDULER_CHECK_NAMES = { + "cancel-closed-pr-runs", + "scan-pr-queue", +} @dataclass @@ -639,7 +660,7 @@ def is_transient_github_api_error(exc: RuntimeError) -> bool: def gh_graphql(query: str, **fields: str | int) -> dict[str, Any]: """Run a GitHub GraphQL query through gh and decode the JSON response.""" - cmd = ["gh", "api", "graphql", "-F", "query=@-"] + cmd = [GITHUB_CLI, "api", "graphql", "-F", "query=@-"] for key, value in fields.items(): flag = "-F" if isinstance(value, int) else "-f" cmd.extend([flag, f"{key}={value}"]) @@ -658,6 +679,15 @@ def gh_graphql(query: str, **fields: str | int) -> dict[str, Any]: time.sleep(delay) +def gh_graphql_mutation(query: str, **fields: str | int) -> dict[str, Any]: + """Run a GitHub GraphQL mutation with the scheduler mutation credential.""" + cmd = [GITHUB_CLI, "api", "graphql", "-F", "query=@-"] + for key, value in fields.items(): + flag = "-F" if isinstance(value, int) else "-f" + cmd.extend([flag, f"{key}={value}"]) + return json.loads(run(cmd, stdin=query)) + + def github_resource_inaccessible(exc: RuntimeError) -> bool: """Return whether GitHub denied an API read for the current integration token.""" @@ -667,7 +697,7 @@ def github_resource_inaccessible(exc: RuntimeError) -> bool: def gh_api_json(path: str) -> Any: """Run a GitHub REST API request through gh and decode the JSON response.""" - return json.loads(run_github_read(["gh", "api", path])) + return json.loads(run_github_read([GITHUB_CLI, "api", path])) def rest_review_node(review: dict[str, Any]) -> dict[str, Any]: @@ -684,7 +714,7 @@ def rest_review_node(review: dict[str, Any]) -> dict[str, Any]: } -def rest_check_node(check: dict[str, Any]) -> dict[str, Any]: +def rest_check_node(check: dict[str, Any], *, workflow_name: str | None = None) -> dict[str, Any]: """Convert a REST check-run payload into the GraphQL status rollup shape.""" return { @@ -694,10 +724,56 @@ def rest_check_node(check: dict[str, Any]) -> dict[str, Any]: "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": {}}}, + "checkSuite": { + "workflowRun": { + "workflow": {"name": workflow_name} if workflow_name else None, + } + }, } +def rest_strix_workflow_name( + repo: str, + check: dict[str, Any], + *, + expected_head_sha: str, + actions_runs_by_id: dict[str, Any] | None = None, +) -> str | None: + """Prove REST-fallback Strix provenance from the Actions run API.""" + + if check.get("name") != "strix" or ((check.get("app") or {}).get("slug")) != "github-actions": + return None + details_url = check.get("details_url") + if not isinstance(details_url, str): + return None + parsed = urlparse(details_url) + expected_prefix = f"/{repo}/actions/runs/".lower() + if parsed.scheme != "https" or parsed.netloc.lower() != "github.com" or not parsed.path.lower().startswith(expected_prefix): + return None + run_id = actions_run_id_from_details_url(details_url) + if run_id is None: + return None + if actions_runs_by_id is not None and run_id in actions_runs_by_id: + run = actions_runs_by_id[run_id] + else: + try: + run = gh_api_json(f"repos/{repo}/actions/runs/{run_id}") + except (RuntimeError, TypeError, ValueError): + run = None + if actions_runs_by_id is not None: + actions_runs_by_id[run_id] = run + if not isinstance(run, dict): + return None + workflow_name = run.get("name") + if ( + str(run.get("id")) != run_id + or run.get("head_sha") != expected_head_sha + or workflow_name not in {"Strix Security Scan", "Strix"} + ): + return None + return workflow_name + + def rest_pr_node(repo: str, pr: dict[str, Any]) -> dict[str, Any]: """Convert a REST pull request payload into the GraphQL shape used by the scheduler.""" @@ -708,6 +784,7 @@ def rest_pr_node(repo: str, pr: dict[str, Any]) -> dict[str, Any]: 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") files = gh_api_json(f"repos/{repo}/pulls/{number}/files?per_page=20") + actions_runs_by_id: dict[str, Any] = {} rest_merge_state = REST_MERGEABLE_STATE_MAP.get( str(pr.get("mergeable_state") or "").lower(), str(pr.get("mergeable_state") or "").upper(), @@ -733,7 +810,15 @@ def rest_pr_node(repo: str, pr: dict[str, Any]) -> dict[str, Any]: "statusCheckRollup": { "contexts": { "nodes": [ - rest_check_node(check) + rest_check_node( + check, + workflow_name=rest_strix_workflow_name( + repo, + check, + expected_head_sha=head.get("sha"), + actions_runs_by_id=actions_runs_by_id, + ), + ) for check in (checks.get("check_runs") or []) ] } @@ -824,11 +909,24 @@ def fetch_pr(repo: str, number: int) -> list[dict[str, Any]]: return prs +def fetch_pr_for_merge_revalidation(repo: str, number: int) -> dict[str, Any]: + """Fetch full live GraphQL evidence without a fail-open REST fallback.""" + repo = validate_github_repository(repo) + if not isinstance(number, int) or number <= 0: + raise ValueError(f"invalid pull request number: {number!r}") + owner, name = split_repo(repo) + payload = gh_graphql(PR_BY_NUMBER_QUERY, owner=owner, name=name, number=number) + pr = ((payload.get("data") or {}).get("repository") or {}).get("pullRequest") + if not isinstance(pr, dict): + raise RuntimeError(f"PR #{number} disappeared before merge-state revalidation") + return pr + + def fetch_rest_mergeable_state(repo: str, number: int) -> str: """Fetch and normalize GitHub REST mergeable_state for one pull request.""" - raw_state = run( + raw_state = run_github_read( [ - "gh", + GITHUB_CLI, "api", f"repos/{repo}/pulls/{number}", "--jq", @@ -853,9 +951,9 @@ def fetch_compare_branch_freshness(repo: str, pr: dict[str, Any]) -> dict[str, A base = quote(pr.get("baseRefName") or "base", safe="") head = quote(compare_ref_for_pr_head(repo, pr), safe=":") return json.loads( - run( + run_github_read( [ - "gh", + GITHUB_CLI, "api", f"repos/{repo}/compare/{base}...{head}", ] @@ -929,6 +1027,45 @@ def context_nodes(pr: dict[str, Any]) -> list[dict[str, Any]]: return contexts.get("nodes") or [] +def latest_check_runs(pr: dict[str, Any]) -> list[dict[str, Any]]: + """Return the latest same-name CheckRun for each workflow provenance.""" + latest: dict[tuple[str, str], tuple[datetime | None, int, dict[str, Any]]] = {} + for index, node in enumerate(context_nodes(pr)): + if node.get("__typename") != "CheckRun": + continue + workflow_name = ( + (((node.get("checkSuite") or {}).get("workflowRun") or {}).get("workflow") or {}).get("name") + or "" + ) + key = (workflow_name, node.get("name") or "check-run") + candidate = (parse_github_datetime(node.get("startedAt")), index, node) + previous = latest.get(key) + if previous is None or ( + candidate[0] or datetime.min.replace(tzinfo=timezone.utc), + candidate[1], + ) >= ( + previous[0] or datetime.min.replace(tzinfo=timezone.utc), + previous[1], + ): + latest[key] = candidate + return [node for _, _, node in sorted(latest.values(), key=lambda item: item[1])] + + +def latest_matching_check_run(pr: dict[str, Any], predicate: Any) -> dict[str, Any] | None: + """Return the latest provenanced CheckRun satisfying a predicate.""" + matches = [node for node in latest_check_runs(pr) if predicate(node)] + if not matches: + return None + return max( + enumerate(matches), + key=lambda item: ( + parse_github_datetime(item[1].get("startedAt")) + or datetime.min.replace(tzinfo=timezone.utc), + item[0], + ), + )[1] + + def is_opencode_context(node: dict[str, Any]) -> bool: """Return whether a check or status context belongs to OpenCode Review.""" if node.get("__typename") == "CheckRun": @@ -941,17 +1078,14 @@ def is_opencode_context(node: dict[str, Any]) -> bool: 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_name = workflow.get("name") - return workflow_name in {"Strix Security Scan", "Strix"} or ( - node.get("name") == "strix" and workflow_name is None - ) - return (node.get("context") or "") in {"strix", "Strix Security Scan"} + """Return whether a CheckRun is exact trusted Strix workflow evidence.""" + if node.get("__typename") != "CheckRun" or node.get("name") != "strix": + return False + workflow = ( + ((node.get("checkSuite") or {}).get("workflowRun") or {}).get("workflow") + or {} + ) + return workflow.get("name") in {"Strix Security Scan", "Strix"} def actions_job_id_from_details_url(value: str | None) -> str | None: @@ -962,15 +1096,18 @@ def actions_job_id_from_details_url(value: str | None) -> str | None: return match.group(1) if match else None +def actions_run_id_from_details_url(value: str | None) -> str | None: + """Return a GitHub Actions run id from a check-run details URL.""" + if not value: + return None + match = ACTIONS_RUN_DETAILS_URL_RE.search(value) + return match.group(1) if match else None + + def matching_actions_job_id(pr: dict[str, Any], predicate: Any) -> str | None: """Return the latest matching check-run job id, if GitHub exposed one.""" - for node in reversed(context_nodes(pr)): - if node.get("__typename") != "CheckRun" or not predicate(node): - continue - job_id = actions_job_id_from_details_url(node.get("detailsUrl")) - if job_id: - return job_id - return None + node = latest_matching_check_run(pr, predicate) + return actions_job_id_from_details_url(node.get("detailsUrl")) if node else None def parse_github_datetime(value: str | None) -> datetime | None: @@ -1049,18 +1186,15 @@ def opencode_in_progress(pr: dict[str, Any], *, stale_after_minutes: int | None def strix_evidence_state(pr: dict[str, Any]) -> str: - """Return missing, running, or complete for current-head Strix evidence.""" - found = False - for node in context_nodes(pr): - if not is_strix_context(node): - continue - found = True - status = (node.get("status") or node.get("state") or "").upper() - if status in RUNNING_CHECK_STATES: - return "running" - if node.get("__typename") == "CheckRun" and status != "COMPLETED": - return "running" - return "complete" if found else "missing" + """Return missing, running, failed, or complete for trusted Strix evidence.""" + node = latest_matching_check_run(pr, is_strix_context) + if node is None: + return "missing" + status = (node.get("status") or "").upper() + if status in RUNNING_CHECK_STATES or status != "COMPLETED": + return "running" + conclusion = (node.get("conclusion") or "").upper() + return "complete" if conclusion == "SUCCESS" else "failed" def unresolved_thread_count(pr: dict[str, Any]) -> int: @@ -1081,7 +1215,7 @@ def outdated_thread_ids(pr: dict[str, Any]) -> list[str]: def resolve_review_thread(thread_id: str) -> None: """Resolve one GitHub review thread by GraphQL node ID.""" - gh_graphql(RESOLVE_REVIEW_THREAD_MUTATION, threadId=thread_id) + gh_graphql_mutation(RESOLVE_REVIEW_THREAD_MUTATION, threadId=thread_id) def resolve_outdated_review_threads(pr: dict[str, Any], *, dry_run: bool) -> int: @@ -1234,7 +1368,7 @@ def dismiss_pull_request_review( try: run( [ - "gh", + GITHUB_CLI, "api", "-X", "PUT", @@ -1245,7 +1379,7 @@ def dismiss_pull_request_review( ) live_state = run_github_read( [ - "gh", + GITHUB_CLI, "api", f"repos/{repo}/pulls/{number}/reviews/{review_id}", "--jq", @@ -1287,7 +1421,7 @@ def dismiss_stale_opencode_approvals( number = str(int(pr["number"])) expected_head = validate_git_sha(pr["headRefOid"]) live_head = run_github_read( - ["gh", "api", f"repos/{repo}/pulls/{number}", "--jq", ".head.sha"] + [GITHUB_CLI, "api", f"repos/{repo}/pulls/{number}", "--jq", ".head.sha"] ).strip() if live_head != expected_head: raise RuntimeError( @@ -1335,7 +1469,7 @@ def dismiss_stale_opencode_change_requests(repo: str, pr: dict[str, Any], *, dry number = str(int(pr["number"])) expected_head = validate_git_sha(pr["headRefOid"]) live_head = run_github_read( - ["gh", "api", f"repos/{repo}/pulls/{number}", "--jq", ".head.sha"] + [GITHUB_CLI, "api", f"repos/{repo}/pulls/{number}", "--jq", ".head.sha"] ).strip() if live_head != expected_head: raise RuntimeError( @@ -1350,7 +1484,7 @@ def dismiss_stale_opencode_change_requests(repo: str, pr: dict[str, Any], *, dry ) run( [ - "gh", + GITHUB_CLI, "api", "-X", "PUT", @@ -1365,63 +1499,55 @@ def dismiss_stale_opencode_change_requests(repo: str, pr: dict[str, Any], *, dry def failed_status_checks(pr: dict[str, Any]) -> list[str]: """Return failing check or status context names from the PR rollup.""" failed: list[str] = [] - latest_check_runs: dict[ - tuple[str, str], - tuple[datetime | None, int, dict[str, Any]], - ] = {} - status_contexts: list[dict[str, Any]] = [] - for index, node in enumerate(context_nodes(pr)): - if node.get("__typename") != "CheckRun": - status_contexts.append(node) - continue - workflow = ( - (((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) - if previous is None: - latest_check_runs[key] = (started_at, index, node) - continue - previous_started_at, previous_index, _ = previous - if started_at is None and previous_started_at is not None: - continue - if previous_started_at is None and started_at is not None: - latest_check_runs[key] = (started_at, index, node) - continue - if (started_at or datetime.min.replace(tzinfo=timezone.utc), index) >= ( - previous_started_at or datetime.min.replace(tzinfo=timezone.utc), - previous_index, - ): - latest_check_runs[key] = (started_at, index, node) - - successful_status_contexts = { - node.get("context") - for node in status_contexts - if (node.get("state") or "").upper() == "SUCCESS" - } - for _, _, node in sorted(latest_check_runs.values(), key=lambda item: item[1]): + for node in latest_check_runs(pr): conclusion = (node.get("conclusion") or "").upper() 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: - continue failed.append(node.get("name") or "check-run") - for node in status_contexts: + for node in context_nodes(pr): + if node.get("__typename") == "CheckRun": + continue state = (node.get("state") or "").upper() if state in {"FAILURE", "ERROR"}: failed.append(node.get("context") or "status-context") return failed +def running_status_checks(pr: dict[str, Any]) -> list[str]: + """Return active peer checks, excluding this scheduler's own live job.""" + running: list[str] = [] + seen: set[str] = set() + for node in latest_check_runs(pr): + workflow_name = ( + (((node.get("checkSuite") or {}).get("workflowRun") or {}).get("workflow") or {}).get("name") + or "" + ) + name = node.get("name") or "check-run" + if workflow_name in SCHEDULER_WORKFLOW_NAMES or name in SCHEDULER_CHECK_NAMES: + continue + if ( + (node.get("status") or "").upper() in RUNNING_CHECK_STATES + and name not in seen + ): + running.append(name) + seen.add(name) + for node in context_nodes(pr): + if node.get("__typename") == "CheckRun": + continue + name = node.get("context") or "status-context" + if ( + (node.get("state") or "").upper() in RUNNING_CHECK_STATES + and name not in SCHEDULER_CHECK_NAMES + and name not in seen + ): + running.append(name) + seen.add(name) + return running + + def action_required_checks(pr: dict[str, Any]) -> list[str]: """Return check-run names that need explicit GitHub Actions approval or unblocking.""" required: list[str] = [] - for node in context_nodes(pr): - if node.get("__typename") != "CheckRun": - continue + for node in latest_check_runs(pr): conclusion = (node.get("conclusion") or "").upper() if conclusion in ACTION_REQUIRED_CONCLUSIONS: required.append(node.get("name") or "check-run") @@ -1438,6 +1564,69 @@ def workflow_action_required_reason(checks: list[str]) -> str: ) +def revalidate_merge_mutation_state( + repo: str, + inspected_pr: dict[str, Any], + *, + action: str, +) -> dict[str, Any]: + """Re-fetch and fail closed if merge-relevant live evidence became unsafe.""" + repo = validate_github_repository(repo) + number = int(inspected_pr["number"]) + expected_head = validate_git_sha(inspected_pr["headRefOid"]) + expected_base = validate_git_sha(inspected_pr["baseRefOid"]) + live = fetch_pr_for_merge_revalidation(repo, number) + + blockers: list[str] = [] + if live.get("headRefOid") != expected_head: + blockers.append(f"head changed from {expected_head} to {live.get('headRefOid') or ''}") + if live.get("baseRefOid") != expected_base: + blockers.append(f"base changed from {expected_base} to {live.get('baseRefOid') or ''}") + if live.get("baseRefName") != inspected_pr.get("baseRefName"): + blockers.append("base branch changed") + if live.get("isDraft"): + blockers.append("PR became draft") + if not same_repository_head(repo, live): + blockers.append("PR head is no longer in the target repository") + unresolved = unresolved_thread_count(live) + if unresolved: + blockers.append(f"{unresolved} active review thread(s) are unresolved") + if has_current_head_changes_requested(live): + blockers.append("current-head OpenCode review requests changes") + if not has_current_head_approval(live): + blockers.append("current-head OpenCode approval is absent") + + strix_state = strix_evidence_state(live) + if strix_state != "complete": + blockers.append(f"trusted same-head Strix evidence is {strix_state}") + failed_checks = failed_status_checks(live) + if failed_checks: + blockers.append(f"failed check(s): {', '.join(failed_checks[:5])}") + action_required = action_required_checks(live) + if action_required: + blockers.append(f"action-required check(s): {', '.join(action_required[:5])}") + running_checks = running_status_checks(live) + if running_checks: + blockers.append(f"running check(s): {', '.join(running_checks[:5])}") + + inspected_state = (inspected_pr.get("mergeStateStatus") or "").upper() + live_state = (live.get("mergeStateStatus") or "").upper() + if live_state != inspected_state: + blockers.append( + f"merge policy state changed from {inspected_state or ''} " + f"to {live_state or ''}" + ) + if live_state in {"BEHIND", "DIRTY", "CONFLICTING", "DRAFT", "UNKNOWN"}: + blockers.append(f"live merge policy state is {live_state}") + + if blockers: + raise RuntimeError( + f"PR #{number} live state changed or became unsafe before {action}; " + + "; ".join(blockers) + ) + return live + + def run_head_guarded_merge( repo: str, number: str, @@ -1446,7 +1635,7 @@ def run_head_guarded_merge( auto: bool, ) -> None: """Run a head-guarded merge using an allowed repository merge method.""" - args = ["gh", "pr", "merge", number, "--repo", repo] + args = [GITHUB_CLI, "pr", "merge", number, "--repo", repo] if auto: args.append("--auto") args.extend(["--squash", "--match-head-commit", head]) @@ -1464,7 +1653,7 @@ def run_head_guarded_merge( f"PR #{number}: squash is disabled; retrying {mode} with a merge commit " f"at guarded head {head}. GitHub reason: {reason}" ) - merge_args = ["gh", "pr", "merge", number, "--repo", repo] + merge_args = [GITHUB_CLI, "pr", "merge", number, "--repo", repo] if auto: merge_args.append("--auto") merge_args.extend(["--merge", "--match-head-commit", head]) @@ -1477,7 +1666,8 @@ def enable_auto_merge(repo: str, pr: dict[str, Any], *, dry_run: bool) -> None: if dry_run: return require_github_actions_mutation_actor("enable-auto-merge") - head = validate_git_sha(pr["headRefOid"]) + live = revalidate_merge_mutation_state(repo, pr, action="enable-auto-merge") + head = validate_git_sha(live["headRefOid"]) run_head_guarded_merge(repo, number, head, auto=True) @@ -1487,7 +1677,8 @@ def merge_pr(repo: str, pr: dict[str, Any], *, dry_run: bool) -> None: if dry_run: return require_github_actions_mutation_actor("direct-merge") - head = validate_git_sha(pr["headRefOid"]) + live = revalidate_merge_mutation_state(repo, pr, action="direct merge") + head = validate_git_sha(live["headRefOid"]) run_head_guarded_merge(repo, number, head, auto=False) @@ -1520,7 +1711,7 @@ def disable_auto_merge(repo: str, pr: dict[str, Any], *, dry_run: bool) -> None: if dry_run: return require_github_actions_mutation_actor("disable-auto-merge") - run(["gh", "pr", "merge", number, "--repo", repo, "--disable-auto"]) + run([GITHUB_CLI, "pr", "merge", number, "--repo", repo, "--disable-auto"]) def disable_auto_merge_decision( @@ -1544,7 +1735,7 @@ def update_branch(repo: str, pr: dict[str, Any], *, dry_run: bool) -> None: head = validate_git_sha(pr["headRefOid"]) run( [ - "gh", + GITHUB_CLI, "api", "-X", "PUT", @@ -1613,19 +1804,19 @@ def restamp_pr_head_for_last_push_approval(repo: str, pr: dict[str, Any], *, dry 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([GITHUB_CLI, "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; " f"expected {head}, observed {live_head or ''}" ) - current_commit = json.loads(run(["gh", "api", f"repos/{repo}/git/commits/{head}"])) + current_commit = json.loads(run([GITHUB_CLI, "api", f"repos/{repo}/git/commits/{head}"])) tree = current_commit.get("tree") or {} tree_sha = validate_git_sha(str(tree.get("sha") or "")) created_commit = json.loads( run( - ["gh", "api", "-X", "POST", f"repos/{repo}/git/commits", "--input", "-"], + [GITHUB_CLI, "api", "-X", "POST", f"repos/{repo}/git/commits", "--input", "-"], stdin=json.dumps( { "message": LAST_PUSH_APPROVAL_RESTAMP_MESSAGE, @@ -1637,7 +1828,7 @@ 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", "-"], + [GITHUB_CLI, "api", "-X", "PATCH", f"repos/{repo}/git/refs/heads/{head_ref}", "--input", "-"], stdin=json.dumps({"sha": new_head, "force": False}), ) return new_head @@ -1819,7 +2010,7 @@ 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([GITHUB_CLI, "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]]: @@ -1829,7 +2020,7 @@ def active_workflow_runs(repo: str, statuses: Sequence[str] = ("queued", "in_pro payload = json.loads( run_github_actions( [ - "gh", + GITHUB_CLI, "api", "--method", "GET", @@ -1985,7 +2176,7 @@ def cancel_one(run_id: str) -> tuple[str, str | None]: try: run_github_actions( [ - "gh", + GITHUB_CLI, "api", "-X", "POST", @@ -2071,7 +2262,7 @@ def dispatch_opencode_review(repo: str, workflow: str, pr: dict[str, Any], *, dr dispatch_repo = repository_dispatch_target(target_repo) run_github_dispatch( [ - "gh", + GITHUB_CLI, "api", "-X", "POST", @@ -2126,7 +2317,7 @@ def dispatch_strix_evidence(repo: str, workflow: str, pr: dict[str, Any], *, dry dispatch_repo = repository_dispatch_target(target_repo) run_github_dispatch( [ - "gh", + GITHUB_CLI, "api", "-X", "POST", @@ -2444,6 +2635,47 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio ) return decide("wait", reason) + if current_head_approved: + approved_strix_state = strix_evidence_state(pr) + if approved_strix_state != "complete": + reason = f"trusted same-head Strix evidence is {approved_strix_state}" + if pr.get("autoMergeRequest"): + return finish( + disable_auto_merge_decision( + repo, + pr, + dry_run=dry_run, + reason=f"{reason}; require a successful provenanced Strix CheckRun before re-enabling auto-merge", + ) + ) + if approved_strix_state == "missing" and trigger_reviews: + if not review_dispatch_allowed: + return decide("wait", f"{reason}; review dispatch limit reached") + wait_reason = repository_dispatch_wait_reason(repo, security_workflow) + if wait_reason: + return decide("wait", f"{reason}; {wait_reason}") + dispatch_strix_evidence(repo, security_workflow, pr, dry_run=dry_run) + return decide( + "security_dispatch", + f"{reason}; same-head Strix dispatched before merge consideration", + ) + action = "block" if approved_strix_state == "failed" else "wait" + return decide(action, f"{reason}; direct merge and auto-merge are withheld") + + running_checks = running_status_checks(pr) + if running_checks: + reason = f"current-head check(s) still running: {', '.join(running_checks[:5])}" + if pr.get("autoMergeRequest"): + return finish( + disable_auto_merge_decision( + repo, + pr, + dry_run=dry_run, + reason=f"{reason}; wait for all peer checks before re-enabling auto-merge", + ) + ) + return decide("wait", reason) + merge_before_update = current_head_can_attempt_merge(pr, merge_state) and ( merge_state == "CLEAN" or merge_mode in {"direct", "direct_or_auto"} ) @@ -2680,6 +2912,11 @@ def request_branch_update(freshness_reason: str, *, suffix: str = "") -> Decisio ) if strix_state == "running": return decide("wait", "same-head Strix evidence is still running") + if strix_state == "failed": + return decide( + "block", + "trusted same-head Strix evidence failed; fix the reported security findings before OpenCode dispatch", + ) # Legacy trusted-base Strix self-test sentinel while this scheduler rollout lands: # same-head Strix and OpenCode dispatched if not review_dispatch_allowed: @@ -2970,7 +3207,7 @@ def last_push_approval_restamp_summary(decisions: list[Decision]) -> list[str]: def parse_external_head_update_reason(reason: str) -> str | None: """Extract the external head repository from non-mutable update guidance.""" - match = re.search(r"head repo ([^\s]+) is external and not writable", reason) + match = EXTERNAL_HEAD_UPDATE_RE.search(reason) if not match: return None return match.group(1) @@ -2978,7 +3215,7 @@ 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 = EXTERNAL_HEAD_MERGE_RE.search(reason) if not match: return None return match.group(1) @@ -3186,7 +3423,19 @@ def self_test() -> None: } ] }, - "statusCheckRollup": {"contexts": {"nodes": []}}, + "statusCheckRollup": { + "contexts": { + "nodes": [ + { + "__typename": "CheckRun", + "name": "strix", + "status": "COMPLETED", + "conclusion": "SUCCESS", + "checkSuite": {"workflowRun": {"workflow": {"name": "Strix Security Scan"}}}, + } + ] + } + }, } assert has_current_head_approval(sample) assert not has_current_head_changes_requested(sample) @@ -3566,6 +3815,7 @@ def self_test() -> None: "name": "strix", "status": "COMPLETED", "conclusion": "SUCCESS", + "checkSuite": {"workflowRun": {"workflow": {"name": "Strix Security Scan"}}}, } ] } @@ -3716,6 +3966,11 @@ def main(argv: list[str]) -> int: raise SystemExit("--base-branch is required") if not args.project_flow: raise SystemExit("--project-flow is required") + try: + args.repo = validate_github_repository(args.repo) + args.base_branch = validate_git_ref(args.base_branch) + except ValueError as exc: + raise SystemExit(str(exc)) from exc if args.pr_number < 0: raise SystemExit("--pr-number must not be negative") if args.review_dispatch_limit < -1: diff --git a/tests/test_pr_review_merge_scheduler.py b/tests/test_pr_review_merge_scheduler.py index 1fe10da7..aab80f2e 100644 --- a/tests/test_pr_review_merge_scheduler.py +++ b/tests/test_pr_review_merge_scheduler.py @@ -22,10 +22,23 @@ SHORT_FINE_GRAINED_TOKEN_BODY = ("A" * 7) + TOKEN_SEPARATOR + ("e" * 7) +@pytest.fixture(autouse=True) +def _use_test_github_cli(monkeypatch): + """Keep argv assertions portable while Actions pins the trusted CLI path.""" + monkeypatch.setattr(sched, "GITHUB_CLI", "gh") + + def fake_github_token(prefix, body): return f"{prefix}{TOKEN_SEPARATOR}{body}" +def test_github_cli_is_pinned_only_inside_actions(): + assert sched.github_cli_for_environment("true") == sched.TRUSTED_GITHUB_CLI + assert sched.github_cli_for_environment("TRUE") == sched.TRUSTED_GITHUB_CLI + assert sched.github_cli_for_environment(None) == "gh" + assert sched.github_cli_for_environment("false") == "gh" + + def fake_fine_grained_github_token(body): return "github" + TOKEN_SEPARATOR + "pat" + TOKEN_SEPARATOR + body @@ -94,6 +107,16 @@ def strix_check(status="COMPLETED", conclusion="SUCCESS", workflow="Strix Securi return value +def approved_pr(**overrides): + """Build an approved PR with successful provenanced same-head Strix evidence.""" + value = make_pr( + reviews={"nodes": [opencode_review("APPROVED", "head")]}, + statusCheckRollup={"contexts": {"nodes": [strix_check()]}}, + ) + value.update(overrides) + return value + + def opencode_check(status="IN_PROGRESS", started_at=None, details_url=None): value = { "__typename": "CheckRun", @@ -222,6 +245,41 @@ def fake_run_with_env(args, stdin=None, env=None): assert calls[1][2]["GH_TOKEN"] == "read-token" +def test_graphql_mutation_keeps_mutation_token_and_pinned_cli(monkeypatch): + calls = [] + monkeypatch.setenv("GH_TOKEN", "mutation-token") + monkeypatch.setenv("SCHEDULER_READ_TOKEN", "read-token") + monkeypatch.setattr(sched, "GITHUB_CLI", sched.TRUSTED_GITHUB_CLI) + monkeypatch.setattr( + sched, + "run", + lambda args, stdin=None: calls.append((args, stdin)) or '{"data": {}}', + ) + monkeypatch.setattr( + sched, + "run_with_env", + lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError("mutation must not use the read-token environment") + ), + ) + + assert sched.gh_graphql_mutation("mutation", threadId="thread-1") == {"data": {}} + assert calls == [ + ( + [ + "/usr/bin/gh", + "api", + "graphql", + "-F", + "query=@-", + "-f", + "threadId=thread-1", + ], + "mutation", + ) + ] + + def test_run_passes_shell_metacharacters_as_plain_arguments(tmp_path): sentinel = tmp_path / "pwned" payload = f"feature; touch {sentinel}; #" @@ -393,7 +451,12 @@ def fake_run(args, stdin=None): calls.append(args) return "dirty\n" - monkeypatch.setattr(sched, "run", fake_run) + monkeypatch.setattr(sched, "run_github_read", fake_run) + monkeypatch.setattr( + sched, + "run", + lambda *args, **kwargs: pytest.fail("read-only REST lookup used mutation credential"), + ) assert sched.fetch_rest_mergeable_state("owner/repo", 7) == "DIRTY" assert calls == [["gh", "api", "repos/owner/repo/pulls/7", "--jq", ".mergeable_state // \"\""]] @@ -403,7 +466,7 @@ def fake_compare_run(args, stdin=None): calls.append(args) return '{"status":"behind","behind_by":3}' - monkeypatch.setattr(sched, "run", fake_compare_run) + monkeypatch.setattr(sched, "run_github_read", fake_compare_run) compare = sched.fetch_compare_branch_freshness( "owner/repo", { @@ -505,9 +568,30 @@ def test_rest_pr_fallback_shapes_reviews_and_checks(monkeypatch): "conclusion": "success", "started_at": "2026-06-30T00:00:00Z", "details_url": "https://github.com/owner/repo/actions/runs/1/job/2", - } + }, + { + "name": "strix", + "status": "completed", + "conclusion": "success", + "started_at": "2026-06-30T00:00:01Z", + "details_url": "https://github.com/owner/repo/actions/runs/10/job/20", + "app": {"slug": "github-actions"}, + }, + { + "name": "strix", + "status": "completed", + "conclusion": "success", + "started_at": "2026-06-30T00:00:02Z", + "details_url": "https://github.com/owner/repo/actions/runs/10/job/21", + "app": {"slug": "github-actions"}, + }, ] }, + "repos/owner/repo/actions/runs/10": { + "id": 10, + "name": "Strix Security Scan", + "head_sha": "abc123", + }, "repos/owner/repo/pulls/42/files?per_page=20": [ {"filename": "scripts/ci/pr_review_merge_scheduler.py"}, ], @@ -541,6 +625,7 @@ def fake_api(path): "repos/owner/repo/pulls/42/reviews?per_page=100", "repos/owner/repo/commits/abc123/check-runs?per_page=100", "repos/owner/repo/pulls/42/files?per_page=20", + "repos/owner/repo/actions/runs/10", ] assert node["number"] == 42 assert node["mergeStateStatus"] == "CLEAN" @@ -552,6 +637,58 @@ def fake_api(path): assert node["reviews"]["nodes"][0]["commit"]["oid"] == "abc123" assert node["statusCheckRollup"]["contexts"]["nodes"][0]["status"] == "COMPLETED" assert node["statusCheckRollup"]["contexts"]["nodes"][0]["conclusion"] == "SUCCESS" + assert [ + context["checkSuite"]["workflowRun"]["workflow"]["name"] + for context in node["statusCheckRollup"]["contexts"]["nodes"][1:] + ] == ["Strix Security Scan", "Strix Security Scan"] + assert sched.strix_evidence_state(node) == "complete" + + +def test_rest_strix_workflow_provenance_fails_closed(monkeypatch): + calls = [] + + def fake_api(path): + calls.append(path) + return { + "id": 10, + "name": "Strix Security Scan", + "head_sha": "other-head", + } + + monkeypatch.setattr(sched, "gh_api_json", fake_api) + check = { + "name": "strix", + "details_url": "https://github.com/owner/repo/actions/runs/10/job/20", + "app": {"slug": "github-actions"}, + } + assert sched.rest_strix_workflow_name("owner/repo", check, expected_head_sha="current-head") is None + assert calls == ["repos/owner/repo/actions/runs/10"] + + check["app"] = {"slug": "untrusted-app"} + assert sched.rest_strix_workflow_name("owner/repo", check, expected_head_sha="current-head") is None + assert calls == ["repos/owner/repo/actions/runs/10"] + + check["app"] = {"slug": "github-actions"} + check["details_url"] = "https://example.invalid/owner/repo/actions/runs/10/job/20" + assert sched.rest_strix_workflow_name("owner/repo", check, expected_head_sha="current-head") is None + assert calls == ["repos/owner/repo/actions/runs/10"] + + check["details_url"] = None + assert sched.rest_strix_workflow_name("owner/repo", check, expected_head_sha="current-head") is None + + check["details_url"] = "https://github.com/owner/repo/actions/runs/not-a-run/job/20" + assert sched.rest_strix_workflow_name("owner/repo", check, expected_head_sha="current-head") is None + + check["details_url"] = "https://github.com/owner/repo/actions/runs/11/job/20" + monkeypatch.setattr( + sched, + "gh_api_json", + lambda path: (_ for _ in ()).throw(RuntimeError("Actions run lookup failed")), + ) + assert sched.rest_strix_workflow_name("owner/repo", check, expected_head_sha="current-head") is None + + monkeypatch.setattr(sched, "gh_api_json", lambda path: []) + assert sched.rest_strix_workflow_name("owner/repo", check, expected_head_sha="current-head") is None def test_fetch_pr_falls_back_to_rest_when_graphql_denied(monkeypatch): @@ -565,6 +702,35 @@ def deny_graphql(*args, **kwargs): assert sched.fetch_pr("owner/repo", 77) == [{"number": 77, "repo": "owner/repo"}] +def test_fetch_pr_for_merge_revalidation_requires_live_graphql_pr(monkeypatch): + calls = [] + + def live_graphql(query, **fields): + calls.append((query, fields)) + return {"data": {"repository": {"pullRequest": {"number": 77}}}} + + monkeypatch.setattr(sched, "gh_graphql", live_graphql) + + assert sched.fetch_pr_for_merge_revalidation("owner/repo", 77) == {"number": 77} + assert calls == [ + ( + sched.PR_BY_NUMBER_QUERY, + {"owner": "owner", "name": "repo", "number": 77}, + ) + ] + + with pytest.raises(ValueError, match="invalid pull request number"): + sched.fetch_pr_for_merge_revalidation("owner/repo", 0) + + monkeypatch.setattr( + sched, + "gh_graphql", + lambda *args, **kwargs: {"data": {"repository": {"pullRequest": None}}}, + ) + with pytest.raises(RuntimeError, match="disappeared before merge-state revalidation"): + sched.fetch_pr_for_merge_revalidation("owner/repo", 77) + + def test_rest_api_wrapper_and_fetch_pr_rest(monkeypatch): run_calls = [] @@ -880,12 +1046,25 @@ def test_context_review_and_check_helpers(): assert not sched.is_opencode_context({"context": "strix"}) assert sched.is_strix_context(strix_check()) assert sched.is_strix_context(strix_check(workflow="Strix")) - assert sched.is_strix_context({"context": "Strix Security Scan"}) - assert sched.is_strix_context({"__typename": "CheckRun", "name": "strix", "checkSuite": {"workflowRun": {"workflow": None}}}) + assert not sched.is_strix_context({"context": "Strix Security Scan"}) + assert not sched.is_strix_context( + {"__typename": "CheckRun", "name": "strix", "status": "COMPLETED"} + ) + assert sched.TRUSTED_GITHUB_CLI == "/usr/bin/gh" + assert not sched.is_strix_context( + { + "__typename": "CheckRun", + "name": "strix", + "checkSuite": {"workflowRun": {"workflow": None}}, + } + ) assert not sched.is_strix_context({"context": "lint"}) assert sched.actions_job_id_from_details_url(None) is None assert sched.actions_job_id_from_details_url("https://github.com/owner/repo/actions/runs/123/job/456?pr=1") == "456" assert sched.actions_job_id_from_details_url("https://github.com/owner/repo/actions/runs/123") is None + assert sched.actions_run_id_from_details_url(None) is None + assert sched.actions_run_id_from_details_url("https://github.com/owner/repo/actions/runs/123/job/456?pr=1") == "123" + assert sched.actions_run_id_from_details_url("https://github.com/owner/repo/checks/123") is None check_jobs = make_pr( statusCheckRollup={ "contexts": { @@ -972,7 +1151,7 @@ def test_context_review_and_check_helpers(): assert not sched.opencode_in_progress(unrelated) assert sched.opencode_progress_state(unrelated, stale_after_minutes=45) == "absent" assert sched.strix_evidence_state(make_pr()) == "missing" - assert sched.strix_evidence_state(unrelated) == "running" + assert sched.strix_evidence_state(unrelated) == "missing" mixed_contexts = make_pr( statusCheckRollup={"contexts": {"nodes": [{"context": "lint", "state": "SUCCESS"}, strix_check()]}} ) @@ -984,8 +1163,70 @@ def test_context_review_and_check_helpers(): assert sched.strix_evidence_state(make_pr(statusCheckRollup={"contexts": {"nodes": [strix_check()]}})) == "complete" assert ( sched.strix_evidence_state(make_pr(statusCheckRollup={"contexts": {"nodes": [strix_check(conclusion="FAILURE")]}})) - == "complete" + == "failed" + ) + + unrelated_workflow = make_pr( + statusCheckRollup={ + "contexts": {"nodes": [strix_check(workflow="Unrelated Security Workflow")]} + } + ) + assert sched.strix_evidence_state(unrelated_workflow) == "missing" + + newest_strix_alias = make_pr( + statusCheckRollup={ + "contexts": { + "nodes": [ + { + **strix_check(conclusion="FAILURE"), + "startedAt": "2026-06-25T08:00:00Z", + }, + { + **strix_check(workflow="Strix"), + "startedAt": "2026-06-25T07:00:00Z", + }, + ] + } + } + ) + assert sched.strix_evidence_state(newest_strix_alias) == "failed" + + running_peers = make_pr( + statusCheckRollup={ + "contexts": { + "nodes": [ + { + "__typename": "CheckRun", + "name": "scan-pr-queue", + "status": "IN_PROGRESS", + "checkSuite": {"workflowRun": {"workflow": None}}, + }, + {"__typename": "CheckRun", "name": "cancel-closed-pr-runs", "status": "QUEUED"}, + { + "__typename": "CheckRun", + "name": "tests", + "status": "IN_PROGRESS", + "startedAt": "2026-06-25T07:05:00Z", + "checkSuite": { + "workflowRun": {"workflow": {"name": "Repository CI"}} + }, + }, + { + "__typename": "CheckRun", + "name": "tests", + "status": "QUEUED", + "startedAt": "2026-06-25T07:06:00Z", + "checkSuite": { + "workflowRun": {"workflow": {"name": "Alternative CI"}} + }, + }, + {"context": "tests", "state": "PENDING"}, + {"context": "legacy-security", "state": "PENDING"}, + ] + } + } ) + assert sched.running_status_checks(running_peers) == ["tests", "legacy-security"] threaded = make_pr( reviewThreads={ @@ -1268,7 +1509,7 @@ def test_review_state_and_failed_checks(): } } ) - assert sched.failed_status_checks(manual_strix_supersedes_pr_target_failure) == ["lint"] + assert sched.failed_status_checks(manual_strix_supersedes_pr_target_failure) == ["strix", "lint"] opencode_pr_target_failure_without_status = make_pr( statusCheckRollup={ "contexts": { @@ -1290,7 +1531,7 @@ def test_review_state_and_failed_checks(): } } ) - assert sched.failed_status_checks(manual_opencode_supersedes_pr_target_failure) == ["lint"] + assert sched.failed_status_checks(manual_opencode_supersedes_pr_target_failure) == ["opencode-review", "lint"] def test_workflow_run_followup_defers_deterministic_fallback_retry(monkeypatch): @@ -1398,7 +1639,7 @@ def test_deterministic_fallback_detection_ignores_unrelated_reviews(): def test_current_head_approval_cleans_previous_head_change_gate_before_merge(): - pr = make_pr( + pr = approved_pr( reviews={ "nodes": [ { @@ -1568,6 +1809,11 @@ def fake_run(args, stdin=None): return "" monkeypatch.setattr(sched, "run", fake_run) + monkeypatch.setattr( + sched, + "revalidate_merge_mutation_state", + lambda repo, inspected_pr, action: inspected_pr, + ) pr = make_pr(baseRefOid=base_sha, headRefOid=head_sha) sched.enable_auto_merge("owner/repo", pr, dry_run=True) sched.merge_pr("owner/repo", pr, dry_run=True) @@ -1651,6 +1897,230 @@ def fake_run(args, stdin=None): ] +def test_merge_mutations_revalidate_all_live_security_evidence(monkeypatch): + head_sha = "a" * 40 + base_sha = "b" * 40 + safe = approved_pr( + headRefOid=head_sha, + baseRefOid=base_sha, + reviews={"nodes": [opencode_review("APPROVED", head_sha)]}, + ) + calls = [] + + monkeypatch.setattr(sched, "require_github_actions_mutation_actor", lambda action: None) + monkeypatch.setattr(sched, "fetch_pr_for_merge_revalidation", lambda repo, number: safe) + monkeypatch.setattr(sched, "run", lambda args, stdin=None: calls.append(args) or "") + + assert sched.revalidate_merge_mutation_state("owner/repo", safe, action="test merge") is safe + sched.merge_pr("owner/repo", safe, dry_run=False) + sched.enable_auto_merge("owner/repo", safe, dry_run=False) + assert len(calls) == 2 + assert calls[0][:4] == ["gh", "pr", "merge", "1"] + assert "--auto" not in calls[0] + assert "--auto" in calls[1] + + unsafe_variants = [] + for label, mutation in ( + ("head changed", lambda pr: pr.update(headRefOid="c" * 40)), + ("base changed", lambda pr: pr.update(baseRefOid="d" * 40)), + ("base branch changed", lambda pr: pr.update(baseRefName="develop")), + ("draft changed", lambda pr: pr.update(isDraft=True)), + ( + "head repository changed", + lambda pr: pr.update( + isCrossRepository=True, + headRepository={"nameWithOwner": "fork/repo"}, + ), + ), + ( + "thread changed", + lambda pr: pr.update( + reviewThreads={ + "nodes": [{"id": "new-thread", "isResolved": False, "isOutdated": False}] + } + ), + ), + ( + "review changed", + lambda pr: pr.update( + reviews={"nodes": [opencode_review("CHANGES_REQUESTED", head_sha)]} + ), + ), + ("Strix changed", lambda pr: pr.update(statusCheckRollup={"contexts": {"nodes": []}})), + ( + "checks changed", + lambda pr: pr.update( + statusCheckRollup={ + "contexts": { + "nodes": [ + strix_check(), + { + "__typename": "CheckRun", + "name": "tests", + "status": "IN_PROGRESS", + "checkSuite": { + "workflowRun": {"workflow": {"name": "Repository CI"}} + }, + }, + ] + } + } + ), + ), + ( + "checks failed", + lambda pr: pr.update( + statusCheckRollup={ + "contexts": { + "nodes": [ + strix_check(), + { + "__typename": "CheckRun", + "name": "tests", + "status": "COMPLETED", + "conclusion": "FAILURE", + "checkSuite": { + "workflowRun": {"workflow": {"name": "Repository CI"}} + }, + }, + ] + } + } + ), + ), + ( + "checks require action", + lambda pr: pr.update( + statusCheckRollup={ + "contexts": { + "nodes": [ + strix_check(), + { + "__typename": "CheckRun", + "name": "manual-gate", + "status": "COMPLETED", + "conclusion": "ACTION_REQUIRED", + "checkSuite": { + "workflowRun": {"workflow": {"name": "Repository CI"}} + }, + }, + ] + } + } + ), + ), + ("policy changed", lambda pr: pr.update(mergeStateStatus="BLOCKED")), + ("unsafe policy state", lambda pr: pr.update(mergeStateStatus="BEHIND")), + ): + candidate = json.loads(json.dumps(safe)) + mutation(candidate) + unsafe_variants.append((label, candidate)) + + for label, unsafe in unsafe_variants: + monkeypatch.setattr( + sched, + "fetch_pr_for_merge_revalidation", + lambda repo, number, value=unsafe: value, + ) + with pytest.raises(RuntimeError, match="live state changed or became unsafe"): + sched.revalidate_merge_mutation_state("owner/repo", safe, action=label) + + +def test_approved_pr_never_merges_without_provenanced_successful_strix(): + approved = make_pr(reviews={"nodes": [opencode_review("APPROVED", "head")]}) + missing = inspect(approved, merge_mode="direct_or_auto") + assert missing.action == "security_dispatch" + assert "Strix evidence is missing" in missing.reason + + spoofed = make_pr( + reviews={"nodes": [opencode_review("APPROVED", "head")]}, + statusCheckRollup={ + "contexts": { + "nodes": [ + { + "__typename": "CheckRun", + "name": "strix", + "status": "COMPLETED", + "conclusion": "SUCCESS", + }, + {"context": "strix", "state": "SUCCESS"}, + ] + } + }, + ) + spoofed_decision = inspect(spoofed, merge_mode="direct_or_auto") + assert spoofed_decision.action == "security_dispatch" + assert sched.strix_evidence_state(spoofed) == "missing" + + running = approved_pr( + statusCheckRollup={ + "contexts": {"nodes": [strix_check(status="IN_PROGRESS", conclusion="")]} + } + ) + running_decision = inspect(running, merge_mode="direct_or_auto") + assert running_decision.action == "wait" + assert "Strix evidence is running" in running_decision.reason + + +def test_approved_pr_strix_and_peer_check_wait_boundaries(monkeypatch): + approved_without_strix = make_pr( + reviews={"nodes": [opencode_review("APPROVED", "head")]}, + ) + + auto_merge_missing = inspect( + {**approved_without_strix, "autoMergeRequest": {"enabledAt": "now"}}, + merge_mode="direct_or_auto", + ) + assert auto_merge_missing.action == "disable_auto_merge" + assert "Strix evidence is missing" in auto_merge_missing.reason + + dispatch_limited = inspect( + approved_without_strix, + merge_mode="direct_or_auto", + review_dispatch_allowed=False, + ) + assert dispatch_limited.action == "wait" + assert "review dispatch limit reached" in dispatch_limited.reason + + with monkeypatch.context() as env: + env.setenv("SCHEDULER_REQUIRED_WORKFLOW_REPOSITORY", "central/review") + env.delenv("SCHEDULER_ALLOW_CROSS_REPO_REPOSITORY_DISPATCH", raising=False) + env.delenv("SCHEDULER_DISPATCH_TOKEN", raising=False) + env.delenv("GITHUB_REPOSITORY", raising=False) + cross_repo_wait = inspect(approved_without_strix, merge_mode="direct_or_auto") + assert cross_repo_wait.action == "wait" + assert "cross-repository repository-dispatch credential" in cross_repo_wait.reason + + peer_check = { + "__typename": "CheckRun", + "name": "tests", + "status": "IN_PROGRESS", + "checkSuite": {"workflowRun": {"workflow": {"name": "Repository CI"}}}, + } + approved_with_running_peer = approved_pr( + statusCheckRollup={"contexts": {"nodes": [strix_check(), peer_check]}}, + ) + running_wait = inspect(approved_with_running_peer, merge_mode="direct_or_auto") + assert running_wait.action == "wait" + assert running_wait.reason == "current-head check(s) still running: tests" + + running_auto_merge = inspect( + {**approved_with_running_peer, "autoMergeRequest": {"enabledAt": "now"}}, + merge_mode="direct_or_auto", + ) + assert running_auto_merge.action == "disable_auto_merge" + assert "wait for all peer checks before re-enabling auto-merge" in running_auto_merge.reason + + unapproved_failed_strix = make_pr( + statusCheckRollup={ + "contexts": {"nodes": [strix_check(conclusion="FAILURE")]} + }, + ) + failed_before_review = inspect(unapproved_failed_strix) + assert failed_before_review.action == "block" + assert "fix the reported security findings before OpenCode dispatch" in failed_before_review.reason + + def test_last_push_approval_restamp_creates_same_tree_child(monkeypatch): calls = [] head_sha = "a" * 40 @@ -2389,7 +2859,7 @@ def fake_graphql(query, **fields): calls.append((query, fields)) return {"data": {"resolveReviewThread": {"thread": {"id": fields["threadId"], "isResolved": True}}}} - monkeypatch.setattr(sched, "gh_graphql", fake_graphql) + monkeypatch.setattr(sched, "gh_graphql_mutation", fake_graphql) assert sched.resolve_outdated_review_threads(pr, dry_run=True) == 2 assert calls == [] @@ -2819,7 +3289,7 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): assert unknown_auto_merge.action == "disable_auto_merge" assert "mergeability is still being calculated" in unknown_auto_merge.reason rest_clean = inspect( - make_pr( + approved_pr( mergeStateStatus="BEHIND", restMergeableState="CLEAN", reviews={"nodes": [opencode_review("APPROVED", "head")]}, @@ -2828,7 +3298,7 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): assert rest_clean.action == "auto_merge" assert inspect(make_pr(reviewThreads={"nodes": [{"isResolved": False}]})).reason == "1 unresolved review thread(s)" outdated_only = inspect( - make_pr( + approved_pr( reviewThreads={"nodes": [{"id": "outdated-thread", "isResolved": False, "isOutdated": True}]}, reviews={"nodes": [opencode_review("APPROVED", "head")]}, ) @@ -2874,7 +3344,7 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): ) assert action_required_auto.action == "disable_auto_merge" assert "workflow action required: opencode-review" in action_required_auto.reason - same_head_auto = make_pr( + same_head_auto = approved_pr( autoMergeRequest={"enabledAt": "now"}, reviews={"nodes": [opencode_review("APPROVED", "head", submitted_at="2026-06-25T06:59:59Z")]}, ) @@ -2887,7 +3357,7 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): dirty_auto_reason = sched.auto_merge_wait_reason("DIRTY") assert "auto-merge is already enabled" in dirty_auto_reason assert "conflict repair is required before GitHub can merge it" in dirty_auto_reason - blocked_auto = make_pr( + blocked_auto = approved_pr( restMergeableState="blocked", autoMergeRequest={"enabledAt": "now"}, reviews={"nodes": [opencode_review("APPROVED", "head")]}, @@ -2979,7 +3449,7 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): assert "branch is outdated before review dispatch" in stale_behind_decision.reason assert dispatched == [] - behind = make_pr(mergeStateStatus="BEHIND", reviews={"nodes": [opencode_review("APPROVED", "head")]}) + behind = approved_pr(mergeStateStatus="BEHIND") assert inspect(behind, update_branches=False).reason == "current-head OpenCode review approved; branch update disabled" called = [] monkeypatch.setattr(sched, "update_branch", lambda repo, pr, dry_run: called.append((repo, pr["number"], dry_run))) @@ -2989,7 +3459,7 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): assert "github-actions[bot]" in decision.reason assert called == [("owner/repo", 1, True)] called.clear() - blocked_behind = make_pr( + blocked_behind = approved_pr( mergeStateStatus="BLOCKED", compareBehindBy=2, reviews={"nodes": [opencode_review("APPROVED", "head")]}, @@ -3000,7 +3470,7 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): assert "GitHub mergeability is BLOCKED" in blocked_behind_decision.reason assert called == [("owner/repo", 1, True)] called.clear() - external_behind = make_pr( + external_behind = approved_pr( mergeStateStatus="BEHIND", isCrossRepository=True, maintainerCanModify=False, @@ -3012,7 +3482,7 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): assert "fork/repo is external and not writable" in external_decision.reason assert sched.decision_guidance(external_decision)["type"] == "external_head_update_required" assert called == [] - external_mutable = make_pr( + external_mutable = approved_pr( mergeStateStatus="BEHIND", isCrossRepository=True, maintainerCanModify=True, @@ -3065,7 +3535,7 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): assert "workflow action required: opencode-review" in action_required_decision.reason assert called == [] called.clear() - behind_auto_merge_enabled = make_pr( + behind_auto_merge_enabled = approved_pr( mergeStateStatus="BEHIND", reviews={"nodes": [opencode_review("APPROVED", "head")]}, autoMergeRequest={"enabledAt": "now"}, @@ -3077,7 +3547,7 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): assert called == [("owner/repo", 1, True)] assert disabled == [] called.clear() - rest_behind = make_pr( + rest_behind = approved_pr( mergeStateStatus="CLEAN", restMergeableState="BEHIND", reviews={"nodes": [opencode_review("APPROVED", "head")]}, @@ -3383,7 +3853,7 @@ def test_wait_for_updated_branch_head_returns_none_when_still_outdated(monkeypat def test_inspect_pr_dispatches_strix_after_update_branch_observes_new_head(monkeypatch): updated = [] dispatched = [] - old_head_pr = make_pr( + old_head_pr = approved_pr( mergeStateStatus="BEHIND", reviews={"nodes": [opencode_review("APPROVED", "head")]}, autoMergeRequest={"enabledAt": "now"}, @@ -3411,7 +3881,7 @@ def test_inspect_pr_dispatches_strix_after_update_branch_observes_new_head(monke def test_inspect_pr_notes_when_update_branch_head_is_not_observed(monkeypatch): updated = [] - pr = make_pr( + pr = approved_pr( mergeStateStatus="BEHIND", reviews={"nodes": [opencode_review("APPROVED", "head")]}, ) @@ -3691,16 +4161,16 @@ def test_update_branch_summary_includes_followup_notes(): def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): - approved = make_pr(reviews={"nodes": [opencode_review("APPROVED", "head")]}) + approved = approved_pr() failed = make_pr( reviews={"nodes": [opencode_review("APPROVED", "head")]}, statusCheckRollup={"contexts": {"nodes": [{"__typename": "CheckRun", "name": "strix", "conclusion": "FAILURE"}]}}, ) assert inspect(failed).reason == "failed check(s): strix" - assert inspect(make_pr(reviews={"nodes": [opencode_review("APPROVED", "head")]}, autoMergeRequest={"enabledAt": "now"})).reason == ( + assert inspect(approved_pr(autoMergeRequest={"enabledAt": "now"})).reason == ( "current head is approved; auto-merge already enabled" ) - approved_with_auto_merge = make_pr( + approved_with_auto_merge = approved_pr( reviews={"nodes": [opencode_review("APPROVED", "head")]}, autoMergeRequest={"enabledAt": "now"}, ) @@ -3719,7 +4189,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): assert inspect(approved, merge_mode="unknown").reason == ( "current head is approved; unsupported merge mode: unknown" ) - blocked_approved = make_pr( + blocked_approved = approved_pr( mergeStateStatus="BLOCKED", reviews={"nodes": [opencode_review("APPROVED", "head")]}, ) @@ -3732,7 +4202,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): assert inspect(blocked_approved, merge_mode="unknown").reason == ( "current head is approved; unsupported merge mode: unknown" ) - blocked_unmergeable = make_pr( + blocked_unmergeable = approved_pr( mergeable="UNKNOWN", mergeStateStatus="BLOCKED", reviews={"nodes": [opencode_review("APPROVED", "head")]}, @@ -3753,7 +4223,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): "current head is approved; direct merge waits for CLEAN mergeability; GitHub mergeability is BLOCKED" ) external_unmergeable = inspect( - make_pr( + approved_pr( mergeable="UNKNOWN", mergeStateStatus="BLOCKED", isCrossRepository=True, @@ -3772,7 +4242,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): lambda repo, pr, dry_run: direct_merges.append((repo, pr["number"], dry_run)), ) blocked_direct = inspect( - make_pr( + approved_pr( mergeStateStatus="BLOCKED", reviews={"nodes": [opencode_review("APPROVED", "head")]}, ), @@ -3793,7 +4263,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): assert direct_merges == [("owner/repo", 1, True), ("owner/repo", 1, True)] already_auto_direct_or_auto = inspect( - make_pr( + approved_pr( autoMergeRequest={"enabledAt": "now"}, reviews={"nodes": [opencode_review("APPROVED", "head")]}, ), @@ -3808,7 +4278,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): ] clean_but_compare_behind = inspect( - make_pr( + approved_pr( mergeStateStatus="CLEAN", compareBehindBy=20, reviews={"nodes": [opencode_review("APPROVED", "head")]}, @@ -3825,7 +4295,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): ] blocked_but_mergeable_and_compare_behind = inspect( - make_pr( + approved_pr( mergeStateStatus="BLOCKED", compareBehindBy=20, reviews={"nodes": [opencode_review("APPROVED", "head")]}, @@ -3847,7 +4317,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): assert inspect(approved).action == "auto_merge" assert auto_merges == [("owner/repo", 1, True)] blocked_direct_or_auto = inspect( - make_pr( + approved_pr( mergeStateStatus="BLOCKED", reviews={"nodes": [opencode_review("APPROVED", "head")]}, ), @@ -3865,7 +4335,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): ] assert auto_merges == [("owner/repo", 1, True)] blocked_already_auto = inspect( - make_pr( + approved_pr( mergeStateStatus="BLOCKED", autoMergeRequest={"enabledAt": "now"}, reviews={"nodes": [opencode_review("APPROVED", "head")]}, @@ -3893,7 +4363,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): ] external_approved = inspect( - make_pr( + approved_pr( isCrossRepository=True, headRepository={"nameWithOwner": "fork/repo"}, reviews={"nodes": [opencode_review("APPROVED", "head")]}, @@ -3904,7 +4374,7 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): assert "fork or external PR heads are excluded from scheduler direct merge and auto-merge" in external_approved.reason assert sched.decision_guidance(external_approved)["type"] == "external_head_merge_excluded" external_blocked = inspect( - make_pr( + approved_pr( mergeStateStatus="BLOCKED", isCrossRepository=True, headRepository={"nameWithOwner": "fork/repo"}, @@ -4020,7 +4490,7 @@ def test_inspect_pr_waits_when_same_head_dispatch_is_already_running(monkeypatch def test_direct_or_auto_falls_back_to_auto_merge_when_branch_policy_blocks_direct_merge(monkeypatch): - approved = make_pr(reviews={"nodes": [opencode_review("APPROVED", "head")]}) + approved = approved_pr() auto_merges = [] def policy_blocked_merge(repo, pr, dry_run): @@ -4046,7 +4516,7 @@ def policy_blocked_merge(repo, pr, dry_run): assert auto_merges == [("owner/repo", 1, True)] already_queued = inspect( - make_pr( + approved_pr( autoMergeRequest={"enabledAt": "now"}, reviews={"nodes": [opencode_review("APPROVED", "head")]}, ), @@ -4057,7 +4527,7 @@ def policy_blocked_merge(repo, pr, dry_run): assert "existing auto-merge request remains queued" in already_queued.reason assert auto_merges == [("owner/repo", 1, True)] - blocked = make_pr( + blocked = approved_pr( mergeStateStatus="BLOCKED", reviews={"nodes": [opencode_review("APPROVED", "head")]}, ) @@ -4082,7 +4552,7 @@ def non_policy_merge_failure(repo, pr, dry_run): def test_direct_or_auto_attempts_direct_merge_when_mergeability_is_blocked(monkeypatch): - approved = make_pr( + approved = approved_pr( mergeStateStatus="BLOCKED", reviews={"nodes": [opencode_review("APPROVED", "head")]}, ) @@ -4216,6 +4686,27 @@ def test_main_rejects_invalid_branch_update_limit(): ) +@pytest.mark.parametrize( + ("flag", "value", "message"), + [ + ("--repo", "owner/repo/extra", "invalid GitHub repository"), + ("--base-branch", "feature..escape", "invalid git ref"), + ], +) +def test_main_rejects_untrusted_repository_and_ref_inputs(flag, value, message): + argv = [ + "--repo", + "owner/repo", + "--base-branch", + "main", + "--project-flow", + "github-flow", + ] + argv[argv.index(flag) + 1] = value + with pytest.raises(SystemExit, match=message): + sched.main(argv) + + def test_print_summary_self_test_parse_args_and_main(monkeypatch, capsys): sched.print_summary( [sched.Decision(1, "wait", "ready"), sched.Decision(2, "wait", "queued")],