1144 lines
41 KiB
Python
1144 lines
41 KiB
Python
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
import requests
|
|
from github import Auth, Github
|
|
|
|
# Import scripts/ci/runner_configs.py (sibling-up dir) for runner_config -> runs_on lookup.
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
|
|
import runner_configs as _runner_configs # noqa: E402
|
|
|
|
# rerun-test workflow doesn't build sgl-kernel, so b200 stages always use the
|
|
# non-kernel pool when resolving the `$b200_runner` sentinel from runner_configs.yml.
|
|
_B200_DEFAULT_RUNNER = "4-gpu-b200"
|
|
|
|
# install_script values from runner_configs.yml are passed verbatim into a
|
|
# `bash ${{ inputs.install_script }}` step in rerun-test.yml. GHA expression
|
|
# substitution happens before bash parses, so shell metacharacters in the
|
|
# string would inject. Restrict the allowed shape to `scripts/ci/cuda/*.sh`
|
|
# (single path component under that dir, no whitespace/operators).
|
|
_ALLOWED_INSTALL_SCRIPT = re.compile(r"^scripts/ci/cuda/[\w.-]+\.sh$")
|
|
|
|
# Configuration
|
|
PERMISSIONS_FILE_PATH = ".github/CI_PERMISSIONS.json"
|
|
|
|
|
|
MAINTENANCE_ISSUE_NUMBER = 21065
|
|
|
|
|
|
def _check_rebase_gate(gh_repo, pr, token):
|
|
"""
|
|
Pre-dispatch gate mirroring `.github/actions/check-maintenance/action.yml`.
|
|
|
|
Without this, /rerun-test would dispatch a workflow_run
|
|
on a PR that's behind a required base, the action would catch it, and
|
|
every job in the run would fail at the gate — wasting runner time and
|
|
producing N error annotations instead of one comment. Pre-checking here
|
|
short-circuits the dispatch and posts a single explanatory comment.
|
|
|
|
Mirrors the action's two independent modes driven by issue #21065:
|
|
(1) Full-pause: maintenance issue is OPEN
|
|
(2) Rebase-required: issue body contains `MIN_BASE_SHA: <sha>`
|
|
Both bypassed by the `bypass-maintenance` PR label.
|
|
|
|
Returns (allowed: bool, message: Optional[str]). When allowed=False,
|
|
caller MUST post `message` to the PR and skip dispatch.
|
|
Fail-open on API errors (matches the action's behavior).
|
|
"""
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Accept": "application/vnd.github+json",
|
|
"X-GitHub-Api-Version": "2022-11-28",
|
|
}
|
|
repo_full_name = gh_repo.full_name
|
|
|
|
try:
|
|
issue_resp = requests.get(
|
|
f"https://api.github.com/repos/{repo_full_name}/issues/{MAINTENANCE_ISSUE_NUMBER}",
|
|
headers=headers,
|
|
timeout=15,
|
|
)
|
|
if issue_resp.status_code != 200:
|
|
print(
|
|
f"check_rebase_gate: issue fetch returned {issue_resp.status_code}; fail-open"
|
|
)
|
|
return True, None
|
|
issue_data = issue_resp.json()
|
|
except Exception as e:
|
|
print(f"check_rebase_gate: issue fetch failed ({e}); fail-open")
|
|
return True, None
|
|
|
|
issue_state = (issue_data.get("state") or "").lower()
|
|
issue_body = issue_data.get("body") or ""
|
|
|
|
min_base_sha = None
|
|
# First MIN_BASE_SHA: <sha> line wins. Match the action's parser:
|
|
# tolerate optional backticks and either ':' or '=' separator.
|
|
for line in issue_body.replace("\r", "").split("\n"):
|
|
m = re.match(
|
|
r"^\s*`?MIN_BASE_SHA`?\s*[:=]\s*`?([A-Fa-f0-9]+)`?",
|
|
line,
|
|
)
|
|
if m:
|
|
candidate = m.group(1)
|
|
if 7 <= len(candidate) <= 40:
|
|
min_base_sha = candidate
|
|
break
|
|
|
|
gate_active = (issue_state == "open") or bool(min_base_sha)
|
|
if not gate_active:
|
|
return True, None
|
|
|
|
bypass = any(
|
|
(lbl.name if hasattr(lbl, "name") else lbl.get("name")) == "bypass-maintenance"
|
|
for lbl in pr.get_labels()
|
|
)
|
|
if bypass:
|
|
print("check_rebase_gate: PR has bypass-maintenance label; allowing dispatch")
|
|
return True, None
|
|
|
|
if issue_state == "open":
|
|
msg = (
|
|
"## ⚠️ CI Maintenance Mode is Active\n"
|
|
"The CI infrastructure is currently under maintenance. "
|
|
"All PR CI runs are paused until maintenance is complete. "
|
|
"**Merging non-CI-fix PRs is prohibited during maintenance mode.**\n\n"
|
|
f"Follow [issue #{MAINTENANCE_ISSUE_NUMBER}]"
|
|
f"(https://github.com/{repo_full_name}/issues/{MAINTENANCE_ISSUE_NUMBER}) "
|
|
"for status updates. Re-run was not dispatched."
|
|
)
|
|
return False, msg
|
|
|
|
# MIN_BASE_SHA set, issue not OPEN — check rebase status.
|
|
pr_head_sha = pr.head.sha
|
|
try:
|
|
compare_resp = requests.get(
|
|
f"https://api.github.com/repos/{repo_full_name}/compare/{min_base_sha}...{pr_head_sha}",
|
|
headers=headers,
|
|
timeout=15,
|
|
)
|
|
if compare_resp.status_code != 200:
|
|
print(
|
|
f"check_rebase_gate: compare API returned {compare_resp.status_code}; fail-open"
|
|
)
|
|
return True, None
|
|
status = compare_resp.json().get("status", "unknown")
|
|
except Exception as e:
|
|
print(f"check_rebase_gate: compare API failed ({e}); fail-open")
|
|
return True, None
|
|
|
|
if status in ("ahead", "identical"):
|
|
return True, None
|
|
|
|
msg = (
|
|
"## ⚠️ Rebase Required Before Re-run\n"
|
|
f"A major update has landed on `main`. Your PR is `{status}` relative "
|
|
f"to required base commit `{min_base_sha[:12]}`.\n\n"
|
|
"**Re-run was not dispatched.** What to do:\n"
|
|
"- Rebase your branch onto the latest `main` and push again\n"
|
|
f"- Follow [issue #{MAINTENANCE_ISSUE_NUMBER}]"
|
|
f"(https://github.com/{repo_full_name}/issues/{MAINTENANCE_ISSUE_NUMBER}) for context\n"
|
|
"- CI-fix PRs may request the `bypass-maintenance` label to skip this check"
|
|
)
|
|
return False, msg
|
|
|
|
|
|
def find_workflow_run_url(
|
|
gh_repo,
|
|
workflow_id,
|
|
ref,
|
|
target_stage,
|
|
token,
|
|
dispatch_time,
|
|
pr_head_sha=None,
|
|
max_wait=30,
|
|
test_command=None,
|
|
):
|
|
"""
|
|
Poll for the workflow run URL after dispatch.
|
|
|
|
Uses the dynamic run-name feature to identify runs:
|
|
- Fork PRs: display_title = "[stage-name] sha"
|
|
- Non-fork PRs: display_title = "[stage-name]"
|
|
|
|
Args:
|
|
gh_repo: PyGithub repository object
|
|
workflow_id: ID of the workflow that was dispatched
|
|
ref: Branch/ref the workflow was dispatched on
|
|
target_stage: The stage name we're looking for
|
|
token: GitHub API token
|
|
dispatch_time: Unix timestamp when dispatch was triggered
|
|
pr_head_sha: PR head SHA (for fork PRs, used to match display_title)
|
|
max_wait: Maximum seconds to wait for the run to appear
|
|
|
|
Returns:
|
|
The workflow run URL if found, None otherwise.
|
|
"""
|
|
# Build expected display_title based on workflow's run-name.
|
|
# rerun-test includes test_command: "[rerun-test] <test_command> [<sha>]"
|
|
# Other workflows: "[stage-name] [<sha>]"
|
|
suffix = f" {test_command}" if test_command else ""
|
|
if pr_head_sha:
|
|
expected_title = f"[{target_stage}]{suffix} {pr_head_sha}"
|
|
else:
|
|
expected_title = f"[{target_stage}]{suffix}"
|
|
|
|
print(f"Looking for workflow run with display_title: {expected_title}")
|
|
|
|
for attempt in range(max_wait // 5):
|
|
time.sleep(5)
|
|
|
|
# Get recent workflow_dispatch runs for this workflow
|
|
runs_url = f"https://api.github.com/repos/{gh_repo.full_name}/actions/workflows/{workflow_id}/runs"
|
|
runs_resp = requests.get(
|
|
runs_url,
|
|
params={"event": "workflow_dispatch", "branch": ref, "per_page": 10},
|
|
headers={
|
|
"Authorization": f"Bearer {token}",
|
|
"Accept": "application/vnd.github+json",
|
|
},
|
|
)
|
|
|
|
if runs_resp.status_code != 200:
|
|
print(f"Failed to fetch workflow runs: {runs_resp.status_code}")
|
|
continue
|
|
|
|
for run in runs_resp.json().get("workflow_runs", []):
|
|
# Skip runs created before our dispatch (with 10s tolerance)
|
|
run_created = datetime.fromisoformat(
|
|
run["created_at"].replace("Z", "+00:00")
|
|
).timestamp()
|
|
if run_created < dispatch_time - 10:
|
|
continue
|
|
|
|
# Match by display_title (set by workflow's run-name directive)
|
|
# This is immediately available, unlike job names which require waiting
|
|
display_title = run.get("display_title", "")
|
|
if display_title == expected_title:
|
|
print(
|
|
f"Found matching workflow run: {run['id']} with title '{display_title}'"
|
|
)
|
|
return run["html_url"]
|
|
|
|
print(f"Could not find workflow run after {max_wait} seconds")
|
|
return None
|
|
|
|
|
|
def get_env_var(name):
|
|
val = os.getenv(name)
|
|
if not val:
|
|
print(f"Error: Environment variable {name} not set.")
|
|
sys.exit(1)
|
|
return val
|
|
|
|
|
|
def load_permissions(user_login):
|
|
"""
|
|
Reads the permissions JSON from the local file system and returns
|
|
the permissions dict for the specific user.
|
|
"""
|
|
try:
|
|
print(f"Loading permissions from {PERMISSIONS_FILE_PATH}...")
|
|
if not os.path.exists(PERMISSIONS_FILE_PATH):
|
|
print(f"Error: Permissions file not found at {PERMISSIONS_FILE_PATH}")
|
|
return None
|
|
|
|
with open(PERMISSIONS_FILE_PATH, "r") as f:
|
|
data = json.load(f)
|
|
|
|
user_perms = data.get(user_login)
|
|
|
|
if not user_perms:
|
|
print(f"User '{user_login}' not found in permissions file.")
|
|
return None
|
|
|
|
return user_perms
|
|
|
|
except Exception as e:
|
|
print(f"Failed to load or parse permissions file: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
def has_sgl_kernel_changes(pr):
|
|
"""
|
|
Check if the PR has changes to the sgl-kernel directory.
|
|
This is used to determine if we need a full workflow rerun
|
|
(to rebuild the kernel) vs just rerunning failed jobs.
|
|
"""
|
|
try:
|
|
files = pr.get_files()
|
|
for f in files:
|
|
if f.filename.startswith("sgl-kernel/"):
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
print(f"Warning: Could not check PR files for sgl-kernel changes: {e}")
|
|
# Default to False to avoid unnecessary full reruns
|
|
return False
|
|
|
|
|
|
def handle_tag_run_ci(gh_repo, pr, comment, user_perms, react_on_success=True):
|
|
"""
|
|
Handles the /tag-run-ci-label command.
|
|
Returns True if action was taken, False otherwise.
|
|
"""
|
|
if not user_perms.get("can_tag_run_ci_label", False):
|
|
print("Permission denied: can_tag_run_ci_label is false.")
|
|
return False
|
|
|
|
print("Permission granted. Adding 'run-ci' label.")
|
|
pr.add_to_labels("run-ci")
|
|
|
|
if react_on_success:
|
|
comment.create_reaction("+1")
|
|
print("Label added and comment reacted.")
|
|
else:
|
|
print("Label added (reaction suppressed).")
|
|
|
|
return True
|
|
|
|
|
|
def handle_rerun_failed_ci(gh_repo, pr, comment, user_perms, react_on_success=True):
|
|
"""
|
|
Handles the /rerun-failed-ci command.
|
|
Reruns workflows with 'failure' or 'skipped' conclusions.
|
|
Returns True if action was taken, False otherwise.
|
|
"""
|
|
if not user_perms.get("can_rerun_failed_ci", False):
|
|
print("Permission denied: can_rerun_failed_ci is false.")
|
|
return False
|
|
|
|
print("Permission granted. Triggering rerun of failed or skipped workflows.")
|
|
|
|
# Check if PR has sgl-kernel changes - if so, we may need full reruns
|
|
# to ensure sgl-kernel-build-wheels runs and produces fresh artifacts.
|
|
# However, if the wheel already built successfully for this commit,
|
|
# we can just rerun failed jobs — the artifact is already there.
|
|
sgl_kernel_changes = has_sgl_kernel_changes(pr)
|
|
if sgl_kernel_changes:
|
|
print("PR has sgl-kernel changes - checking if kernel wheel already built")
|
|
|
|
# Get the SHA of the latest commit in the PR
|
|
head_sha = pr.head.sha
|
|
print(f"Checking workflows for commit: {head_sha}")
|
|
|
|
# If PR has sgl-kernel changes, check whether ALL wheel builds already
|
|
# succeeded for this commit (CUDA + ARM). If so, we can use
|
|
# rerun_failed_jobs and avoid retriggering all tests. If any wheel
|
|
# build is pending/failed, a dependent job could fail for missing
|
|
# artifacts, so fall back to full rerun.
|
|
# Check-runs display names: "Build Wheel (<python>, <cuda>)" (CUDA) and
|
|
# "Build Wheel Arm (<python>, <cuda>)" (ARM). The YAML job ids
|
|
# sgl-kernel-build-wheels{,-arm} are NOT what the check-runs API
|
|
# returns — it returns the job's `name:` field.
|
|
kernel_wheel_built = False
|
|
if sgl_kernel_changes:
|
|
try:
|
|
wheel_builds = [
|
|
cr
|
|
for cr in gh_repo.get_commit(head_sha).get_check_runs()
|
|
if cr.name.startswith("Build Wheel")
|
|
]
|
|
kernel_wheel_built = bool(wheel_builds) and all(
|
|
cr.conclusion == "success" for cr in wheel_builds
|
|
)
|
|
print(
|
|
f"All {len(wheel_builds)} kernel wheel build(s) passed - using rerun_failed_jobs"
|
|
if kernel_wheel_built
|
|
else f"Kernel wheel not fully built "
|
|
f"({sum(1 for c in wheel_builds if c.conclusion == 'success')}"
|
|
f"/{len(wheel_builds)} success) - will use full rerun"
|
|
)
|
|
except Exception as e:
|
|
print(
|
|
f"Failed to check kernel wheel status: {e} - falling back to full rerun"
|
|
)
|
|
|
|
# Rerun workflows with conclusion=failure or conclusion=skipped.
|
|
#
|
|
# - failure: use rerun_failed_jobs() which reruns failed jobs *and their
|
|
# dependent jobs* (GitHub API). Fast-fail cascades call
|
|
# core.setFailed(...) so their conclusion is "failure" and are covered.
|
|
# - skipped: the entire run was skipped (no jobs ran), so there are no
|
|
# failed jobs for rerun_failed_jobs() to target. Use run.rerun().
|
|
# - kernel wheel escape: if the PR touches sgl-kernel and not all wheel
|
|
# builds are success yet, full-rerun failure runs too — Build Wheel
|
|
# lives in pr-test-sgl-kernel.yml, consumers in pr-test.yml, and
|
|
# rerun_failed_jobs() is scoped to a single workflow run.
|
|
runs = gh_repo.get_workflow_runs(head_sha=head_sha)
|
|
|
|
rerun_count = 0
|
|
for run in runs:
|
|
if run.status != "completed":
|
|
continue
|
|
if run.conclusion not in ("failure", "skipped"):
|
|
continue
|
|
|
|
print(f"Processing {run.conclusion} workflow: {run.name} (ID: {run.id})")
|
|
try:
|
|
if run.conclusion == "skipped" or (
|
|
sgl_kernel_changes and not kernel_wheel_built
|
|
):
|
|
print(" Full rerun")
|
|
run.rerun()
|
|
else:
|
|
print(" rerun_failed_jobs")
|
|
run.rerun_failed_jobs()
|
|
rerun_count += 1
|
|
except Exception as e:
|
|
print(f"Failed to rerun workflow {run.id}: {e}")
|
|
|
|
if rerun_count > 0:
|
|
print(f"Triggered rerun for {rerun_count} workflows.")
|
|
if react_on_success:
|
|
comment.create_reaction("+1")
|
|
return True
|
|
else:
|
|
print("No failed or skipped workflows found to rerun.")
|
|
return False
|
|
|
|
|
|
MULTIMODAL_TEST_DIR = "python/sglang/multimodal_gen/test"
|
|
|
|
MULTIMODAL_PATH_TO_RUNNER = {
|
|
"2_gpu": "2-gpu-h100",
|
|
"2-gpu": "2-gpu-h100",
|
|
}
|
|
MULTIMODAL_DEFAULT_RUNNER = "1-gpu-h100"
|
|
|
|
|
|
def _known_test_groups():
|
|
groups = []
|
|
for group_dir in glob.glob("test/registered/*"):
|
|
if os.path.isdir(group_dir):
|
|
groups.append(os.path.basename(group_dir))
|
|
return sorted(groups)
|
|
|
|
|
|
def resolve_test_group_specs(group_name):
|
|
"""
|
|
Resolve a test group name into /rerun-test specs.
|
|
|
|
A group maps to a directory under test/registered/. For example,
|
|
"hicache" maps to all test_*.py files under test/registered/hicache/.
|
|
|
|
Returns (test_specs, error_message). On success error_message is None.
|
|
"""
|
|
group_name = group_name.strip().strip("/")
|
|
if (
|
|
not group_name
|
|
or group_name.startswith(".")
|
|
or "/." in group_name
|
|
or ".." in group_name.split("/")
|
|
):
|
|
return [], f"Invalid test group `{group_name}`."
|
|
|
|
group_dir = os.path.join("test", "registered", group_name)
|
|
if not os.path.isdir(group_dir):
|
|
known = ", ".join(f"`{g}`" for g in _known_test_groups())
|
|
return (
|
|
[],
|
|
f"Unknown test group `{group_name}`.\n\nKnown groups: {known}",
|
|
)
|
|
|
|
test_files = sorted(
|
|
glob.glob(os.path.join(group_dir, "**", "test_*.py"), recursive=True)
|
|
)
|
|
if not test_files:
|
|
return [], f"No registered test files found in `{group_dir}`."
|
|
|
|
return [os.path.relpath(path, "test") for path in test_files], None
|
|
|
|
|
|
def resolve_test_file(file_part):
|
|
"""
|
|
Resolve a user-provided file path to a path relative to test/ or full path for multimodal.
|
|
|
|
Supports:
|
|
- Full path: test/registered/core/test_srt_endpoint.py
|
|
- Relative to test/: registered/core/test_srt_endpoint.py
|
|
- Bare filename: test_srt_endpoint.py (glob-matched, must be unique)
|
|
- Multimodal paths: python/sglang/multimodal_gen/test/server/test_server_a.py
|
|
|
|
Returns (resolved_path, is_multimodal, error_message). On success error_message is None.
|
|
"""
|
|
# Check if it's explicitly a multimodal path
|
|
multimodal_prefixes = [
|
|
"python/sglang/multimodal_gen/test/",
|
|
"sglang/multimodal_gen/test/",
|
|
"multimodal_gen/test/",
|
|
]
|
|
for prefix in multimodal_prefixes:
|
|
if file_part.startswith(prefix):
|
|
full_path = (
|
|
file_part
|
|
if file_part.startswith("python/")
|
|
else f"python/sglang/multimodal_gen/test/{file_part[len(prefix):]}"
|
|
)
|
|
if not os.path.isfile(full_path):
|
|
return None, False, f"File not found: `{full_path}`"
|
|
return full_path, True, None
|
|
|
|
# Existing logic for test/registered/ paths
|
|
if file_part.startswith("test/"):
|
|
file_part = file_part[len("test/") :]
|
|
|
|
if "/" not in file_part:
|
|
# Try test/registered/ first
|
|
matches = glob.glob(f"test/registered/**/{file_part}", recursive=True)
|
|
|
|
# Try multimodal test directory
|
|
mm_matches = glob.glob(f"{MULTIMODAL_TEST_DIR}/**/{file_part}", recursive=True)
|
|
# Filter to only test files
|
|
mm_matches = [m for m in mm_matches if os.path.basename(m).startswith("test_")]
|
|
|
|
if len(matches) == 1 and len(mm_matches) == 0:
|
|
return matches[0][len("test/") :], False, None
|
|
if len(matches) == 0 and len(mm_matches) == 1:
|
|
return mm_matches[0], True, None
|
|
|
|
all_matches = matches + mm_matches
|
|
if len(all_matches) == 0:
|
|
return (
|
|
None,
|
|
False,
|
|
f"No test file found matching `{file_part}` under `test/registered/` or `{MULTIMODAL_TEST_DIR}/`.",
|
|
)
|
|
if len(all_matches) > 1:
|
|
match_list = "\n".join(f"- `{m}`" for m in sorted(all_matches))
|
|
return (
|
|
None,
|
|
False,
|
|
(
|
|
f"Ambiguous filename `{file_part}` — matched {len(all_matches)} files:\n\n"
|
|
f"{match_list}\n\n"
|
|
f"Please provide the full path, e.g. `/rerun-test {all_matches[0]}`"
|
|
),
|
|
)
|
|
# Shouldn't reach here, but handle gracefully
|
|
if mm_matches:
|
|
return mm_matches[0], True, None
|
|
return matches[0][len("test/") :], False, None
|
|
|
|
# Path with directory - check test/ location
|
|
full_path = f"test/{file_part}"
|
|
if os.path.isfile(full_path):
|
|
return file_part, False, None
|
|
|
|
return None, False, f"File not found: `{full_path}`"
|
|
|
|
|
|
def detect_multimodal_suite(file_path):
|
|
"""
|
|
Determine runner for a multimodal gen test file based on its path.
|
|
|
|
Returns (runner_label, error_message).
|
|
"""
|
|
# Check path components and basename for GPU count hints
|
|
for pattern, runner in MULTIMODAL_PATH_TO_RUNNER.items():
|
|
if pattern in file_path:
|
|
return runner, None
|
|
return MULTIMODAL_DEFAULT_RUNNER, None
|
|
|
|
|
|
def _extract_runner_config(content):
|
|
"""Pull `runner_config` and the args string from a `register_cuda_ci(...)` call."""
|
|
args = re.search(r"^[^#\n]*register_cuda_ci\s*\(([^)]*)\)", content, re.MULTILINE)
|
|
if not args:
|
|
return None, None
|
|
m = re.search(r'runner_config\s*=\s*["\']([^"\']+)["\']', args.group(1))
|
|
return (m.group(1), args.group(1)) if m else (None, None)
|
|
|
|
|
|
def detect_suite(file_path_from_test):
|
|
"""
|
|
Read a test file and extract dispatch info from register_cuda_ci or
|
|
register_cpu_ci.
|
|
|
|
CUDA tests must use `register_cuda_ci(stage=..., runner_config=...)`;
|
|
runner label, install script, timeout, and rdma_devices are all resolved
|
|
from scripts/ci/runner_configs.yml — the same single source of truth that
|
|
drives the main PR test pipeline.
|
|
|
|
CPU tests (`register_cpu_ci(...)`) dispatch to the CPU job (ubuntu-latest).
|
|
|
|
Returns dict with keys: suite, runner_label, install_script,
|
|
install_timeout, rdma_devices, is_cpu, error.
|
|
"""
|
|
full_path = f"test/{file_path_from_test}"
|
|
with open(full_path, "r") as f:
|
|
content = f.read()
|
|
|
|
def _err(suite, msg):
|
|
return {
|
|
"suite": suite,
|
|
"runner_label": None,
|
|
"install_script": "",
|
|
"install_timeout": "",
|
|
"rdma_devices": "",
|
|
"is_cpu": False,
|
|
"error": msg,
|
|
}
|
|
|
|
rc, args_str = _extract_runner_config(content)
|
|
if rc:
|
|
configs = _runner_configs.load()
|
|
cfg = configs.get(rc)
|
|
if cfg is None:
|
|
known = ", ".join(f"`{k}`" for k in sorted(configs))
|
|
return _err(
|
|
rc,
|
|
f"Unknown runner_config `{rc}` in `{full_path}` "
|
|
f"— not in scripts/ci/runner_configs.yml.\n\n"
|
|
f"Known runner_configs: {known}",
|
|
)
|
|
install_script = cfg["install"]
|
|
if not _ALLOWED_INSTALL_SCRIPT.match(install_script):
|
|
return _err(
|
|
rc,
|
|
f"Disallowed `install` value `{install_script}` for runner_config "
|
|
f"`{rc}` in scripts/ci/runner_configs.yml. The slash handler "
|
|
f"passes this string verbatim into a shell step, so it must "
|
|
f"match `scripts/ci/cuda/*.sh`.",
|
|
)
|
|
runs_on = cfg.get("runs_on")
|
|
# Resolve $b200_runner sentinel: rerun-test never builds sgl-kernel,
|
|
# so always pick the non-kernel b200 pool.
|
|
if runs_on == "$b200_runner":
|
|
runs_on = _B200_DEFAULT_RUNNER
|
|
stage_m = re.search(r'stage\s*=\s*["\']([^"\']+)["\']', args_str)
|
|
suite = f"{stage_m.group(1)}-test-{rc}" if stage_m else rc
|
|
return {
|
|
"suite": suite,
|
|
"runner_label": runs_on,
|
|
"install_script": install_script,
|
|
"install_timeout": str(cfg["install_timeout"]),
|
|
"rdma_devices": cfg.get("rdma_devices", ""),
|
|
"is_cpu": False,
|
|
"error": None,
|
|
}
|
|
|
|
if re.search(r"^[^#\n]*register_cpu_ci\s*\(", content, re.MULTILINE):
|
|
return {
|
|
"suite": "cpu",
|
|
"runner_label": "ubuntu-latest",
|
|
"install_script": "",
|
|
"install_timeout": "",
|
|
"rdma_devices": "",
|
|
"is_cpu": True,
|
|
"error": None,
|
|
}
|
|
|
|
return _err(
|
|
None,
|
|
f"No `register_cuda_ci(runner_config=...)` or `register_cpu_ci()` "
|
|
f"found in `{full_path}`. /rerun-test only supports tests registered "
|
|
f"via the new-style yml-driven API; nightly/weekly tests aren't "
|
|
f"dispatchable through this command.",
|
|
)
|
|
|
|
|
|
def _resolve_test_spec(test_spec):
|
|
"""
|
|
Resolve a single test spec into its components without dispatching.
|
|
|
|
Returns a dict with keys: spec, test_command, mode, runs_on,
|
|
install_script, install_timeout, rdma_devices, error.
|
|
"""
|
|
if "::" in test_spec:
|
|
file_part, test_selector = test_spec.split("::", 1)
|
|
else:
|
|
file_part = test_spec
|
|
test_selector = None
|
|
|
|
file_part = file_part.strip()
|
|
if test_selector:
|
|
test_selector = test_selector.strip()
|
|
|
|
resolved_path, is_multimodal, err = resolve_test_file(file_part)
|
|
if err:
|
|
return {"spec": test_spec, "error": err}
|
|
|
|
if is_multimodal:
|
|
runner_label, err = detect_multimodal_suite(resolved_path)
|
|
if err:
|
|
return {"spec": test_spec, "error": err}
|
|
|
|
# For multimodal pytest tests, use :: separator for test selection
|
|
test_command = resolved_path
|
|
if test_selector:
|
|
test_command = f"{resolved_path}::{test_selector}"
|
|
|
|
print(
|
|
f"Resolved (multimodal_gen): file={resolved_path}, selector={test_selector}, "
|
|
f"runner={runner_label}, command='{test_command}'"
|
|
)
|
|
return {
|
|
"spec": test_spec,
|
|
"test_command": test_command,
|
|
"mode": "multimodal_gen",
|
|
"runs_on": runner_label,
|
|
"install_script": "",
|
|
"install_timeout": "",
|
|
"rdma_devices": "",
|
|
"error": None,
|
|
}
|
|
|
|
info = detect_suite(resolved_path)
|
|
if info["error"]:
|
|
return {"spec": test_spec, "error": info["error"]}
|
|
|
|
test_command = resolved_path
|
|
if test_selector:
|
|
test_command = f"{resolved_path} {test_selector}"
|
|
|
|
mode = "cpu" if info["is_cpu"] else "cuda"
|
|
print(
|
|
f"Resolved: file={resolved_path}, selector={test_selector}, "
|
|
f"suite={info['suite']}, mode={mode}, runs_on={info['runner_label']}, "
|
|
f"install={info['install_script']}, rdma={info['rdma_devices']}, "
|
|
f"command='{test_command}'"
|
|
)
|
|
return {
|
|
"spec": test_spec,
|
|
"test_command": test_command,
|
|
"mode": mode,
|
|
"runs_on": info["runner_label"],
|
|
"install_script": info["install_script"],
|
|
"install_timeout": info["install_timeout"],
|
|
"rdma_devices": info["rdma_devices"],
|
|
"error": None,
|
|
}
|
|
|
|
|
|
def _dispatch_batch(gh_repo, pr, batch, token, reply_comment_id="", reply_marker=""):
|
|
"""
|
|
Dispatch a single workflow run for a batch of resolved test specs that
|
|
share the same dispatch shape (mode + runs_on + install_script +
|
|
install_timeout + rdma_devices).
|
|
|
|
Returns a dict with keys: specs, success, test_commands, runner_label, run_url, error.
|
|
"""
|
|
test_commands = [r["test_command"] for r in batch]
|
|
mode = batch[0]["mode"]
|
|
runs_on = batch[0]["runs_on"]
|
|
install_script = batch[0]["install_script"]
|
|
install_timeout = batch[0]["install_timeout"]
|
|
rdma_devices = batch[0]["rdma_devices"]
|
|
|
|
# Join multiple commands with newlines for the workflow to iterate over
|
|
combined_command = "\n".join(test_commands)
|
|
|
|
try:
|
|
workflow_name = "Rerun Test"
|
|
workflows = gh_repo.get_workflows()
|
|
target_workflow = None
|
|
for wf in workflows:
|
|
if wf.name == workflow_name:
|
|
target_workflow = wf
|
|
break
|
|
|
|
if not target_workflow:
|
|
return {
|
|
"specs": [r["spec"] for r in batch],
|
|
"success": False,
|
|
"error": f"{workflow_name} workflow not found",
|
|
}
|
|
|
|
is_fork = (
|
|
pr.head.repo is None or pr.head.repo.owner.login != gh_repo.owner.login
|
|
)
|
|
|
|
pr_head_sha = None
|
|
inputs = {
|
|
"mode": mode,
|
|
"test_command": combined_command,
|
|
"runs_on": runs_on or "",
|
|
"install_script": install_script,
|
|
"install_timeout": install_timeout or "20",
|
|
"rdma_devices": rdma_devices,
|
|
"reply_comment_id": str(reply_comment_id) if reply_comment_id else "",
|
|
"reply_marker": reply_marker,
|
|
}
|
|
if is_fork:
|
|
ref = "main"
|
|
pr_head_sha = pr.head.sha
|
|
inputs["pr_head_sha"] = pr_head_sha
|
|
else:
|
|
ref = pr.head.ref
|
|
|
|
dispatch_time = time.time()
|
|
|
|
dispatch_url = f"https://api.github.com/repos/{gh_repo.full_name}/actions/workflows/{target_workflow.id}/dispatches"
|
|
dispatch_resp = requests.post(
|
|
dispatch_url,
|
|
json={"ref": ref, "inputs": inputs},
|
|
headers={
|
|
"Authorization": f"Bearer {token}",
|
|
"Accept": "application/vnd.github+json",
|
|
},
|
|
)
|
|
success = dispatch_resp.status_code in (200, 204)
|
|
if not success:
|
|
print(f"Dispatch failed: {dispatch_resp.status_code} {dispatch_resp.text}")
|
|
return {
|
|
"specs": [r["spec"] for r in batch],
|
|
"success": False,
|
|
"error": f"Dispatch failed: {dispatch_resp.status_code}",
|
|
}
|
|
|
|
print(f"Successfully triggered rerun-test: {combined_command}")
|
|
|
|
run_url = find_workflow_run_url(
|
|
gh_repo,
|
|
target_workflow.id,
|
|
ref,
|
|
"rerun-test",
|
|
token,
|
|
dispatch_time,
|
|
pr_head_sha=pr_head_sha,
|
|
max_wait=30,
|
|
test_command=combined_command,
|
|
)
|
|
return {
|
|
"specs": [r["spec"] for r in batch],
|
|
"success": True,
|
|
"test_commands": test_commands,
|
|
"mode": mode,
|
|
"runs_on": runs_on,
|
|
"run_url": run_url,
|
|
"reply_marker": reply_marker,
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Error triggering rerun-test for batch: {e}")
|
|
return {
|
|
"specs": [r["spec"] for r in batch],
|
|
"success": False,
|
|
"error": str(e),
|
|
}
|
|
|
|
|
|
def _check_rerun_test_permissions(gh_repo, pr, comment, user_perms, command_name):
|
|
"""
|
|
Check permissions shared by /rerun-test and /rerun-group.
|
|
"""
|
|
# SECURITY: These commands check out and execute code from the PR branch on
|
|
# self-hosted GPU runners, so fork PRs require a trusted collaborator.
|
|
is_fork = pr.head.repo is None or pr.head.repo.owner.login != gh_repo.owner.login
|
|
if is_fork:
|
|
commenter = comment.user.login
|
|
perm = gh_repo.get_collaborator_permission(commenter)
|
|
if perm not in ("admin", "write"):
|
|
print(f"Permission denied: /{command_name} on fork PR by {commenter}.")
|
|
comment.create_reaction("confused")
|
|
pr.create_issue_comment(
|
|
f"⛔ `/{command_name}` is not available for fork PRs unless the commenter "
|
|
"has write permission on the repo.\n\n"
|
|
"Please ask a maintainer to run this command, or use the normal CI flow."
|
|
)
|
|
return False
|
|
print(f"Fork PR, but commenter {commenter} has write+ permission. Proceeding.")
|
|
|
|
if not (
|
|
user_perms.get("can_rerun_test", False)
|
|
or user_perms.get("can_rerun_stage", False)
|
|
):
|
|
print("Permission denied: neither can_rerun_test nor can_rerun_stage is true.")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def handle_rerun_test(
|
|
gh_repo, pr, comment, user_perms, test_specs, token, skip_permission_check=False
|
|
):
|
|
"""
|
|
Handles the /rerun-test command. Resolves all test specs, groups them by
|
|
dispatch shape (mode + runs_on + install_script + install_timeout +
|
|
rdma_devices), and dispatches one workflow per group.
|
|
"""
|
|
if not skip_permission_check and not _check_rerun_test_permissions(
|
|
gh_repo, pr, comment, user_perms, "rerun-test"
|
|
):
|
|
return False
|
|
|
|
if not test_specs:
|
|
comment.create_reaction("confused")
|
|
pr.create_issue_comment(
|
|
"⛔ Please specify a test: `/rerun-test <file>::<TestClass.test_method>`\n\n"
|
|
"Examples:\n"
|
|
"- `/rerun-test test/registered/core/test_srt_endpoint.py::TestSRTEndpoint.test_simple_decode`\n"
|
|
"- `/rerun-test registered/core/test_srt_endpoint.py::TestSRTEndpoint`\n"
|
|
"- `/rerun-test test_srt_endpoint.py`\n"
|
|
"- `/rerun-test test_a.py test_b.py test_c.py` (multiple tests)"
|
|
)
|
|
return False
|
|
|
|
allowed, gate_msg = _check_rebase_gate(gh_repo, pr, token)
|
|
if not allowed:
|
|
comment.create_reaction("confused")
|
|
pr.create_issue_comment(gate_msg)
|
|
return False
|
|
|
|
# Phase 1: Resolve all specs
|
|
resolved = []
|
|
resolve_failures = []
|
|
for spec in test_specs:
|
|
r = _resolve_test_spec(spec)
|
|
if r.get("error"):
|
|
resolve_failures.append(r)
|
|
else:
|
|
resolved.append(r)
|
|
|
|
# Phase 2: Group by dispatch shape.
|
|
groups = {}
|
|
for r in resolved:
|
|
key = (
|
|
r["mode"],
|
|
r["runs_on"],
|
|
r["install_script"],
|
|
r["install_timeout"],
|
|
r["rdma_devices"],
|
|
)
|
|
groups.setdefault(key, []).append(r)
|
|
|
|
# Phase 3a: Create placeholder reply comment so we have its ID before
|
|
# dispatching workflows. This lets each dispatched run write its
|
|
# success/failure result back to the right line in this comment.
|
|
reply_comment = pr.create_issue_comment("🚀 Dispatching rerun-test workflow(s)...")
|
|
|
|
# Phase 3b: Dispatch one workflow per group, with a unique per-batch
|
|
# marker each. The marker is an HTML comment that the writeback step
|
|
# uses to locate the line and replace 🚀 with ✅/❌.
|
|
dispatch_results = []
|
|
for idx, batch in enumerate(groups.values()):
|
|
marker = f"<!--rrt:{idx}-->"
|
|
dispatch_results.append(
|
|
_dispatch_batch(
|
|
gh_repo,
|
|
pr,
|
|
batch,
|
|
token,
|
|
reply_comment_id=reply_comment.id,
|
|
reply_marker=marker,
|
|
)
|
|
)
|
|
|
|
# Build consolidated comment body (markers placed at line ends so the
|
|
# writeback step can locate and update each line).
|
|
lines = []
|
|
for dr in dispatch_results:
|
|
if dr["success"]:
|
|
if dr["mode"] == "multimodal_gen":
|
|
cmds = "\n".join(
|
|
f"python3 -m pytest {cmd} -x" for cmd in dr["test_commands"]
|
|
)
|
|
else:
|
|
cmds = "\n".join(
|
|
f"cd test/ && python3 {cmd}" for cmd in dr["test_commands"]
|
|
)
|
|
marker = dr.get("reply_marker", "")
|
|
label = dr["runs_on"] or dr["mode"]
|
|
if dr.get("run_url"):
|
|
lines.append(
|
|
f"🚀 `{label}` ({len(dr['test_commands'])} test{'s' if len(dr['test_commands']) > 1 else ''}): "
|
|
f"⏳ [View workflow run]({dr['run_url']}) {marker}\n"
|
|
f"```\n{cmds}\n```"
|
|
)
|
|
else:
|
|
lines.append(
|
|
f"🚀 `{label}` ({len(dr['test_commands'])} test{'s' if len(dr['test_commands']) > 1 else ''}): ⏳ {marker}\n"
|
|
f"```\n{cmds}\n```\n"
|
|
f"⚠️ Could not retrieve workflow run URL. "
|
|
f"Check the [Actions tab](https://github.com/{gh_repo.full_name}/actions) for progress."
|
|
)
|
|
else:
|
|
specs_str = ", ".join(f"`{s}`" for s in dr["specs"])
|
|
lines.append(f"⛔ {specs_str}: {dr['error']}")
|
|
|
|
for r in resolve_failures:
|
|
lines.append(f"⛔ `{r['spec']}`: {r['error']}")
|
|
|
|
body = "\n\n".join(lines)
|
|
|
|
successes = [dr for dr in dispatch_results if dr["success"]]
|
|
if successes:
|
|
comment.create_reaction("+1")
|
|
if not successes and (resolve_failures or dispatch_results):
|
|
comment.create_reaction("confused")
|
|
|
|
reply_comment.edit(body)
|
|
return len(successes) > 0
|
|
|
|
|
|
def handle_rerun_group(gh_repo, pr, comment, user_perms, group_names, token):
|
|
"""
|
|
Handles the /rerun-group command. Expands one or more registered test
|
|
groups into test file specs, then reuses /rerun-test dispatch behavior.
|
|
"""
|
|
if not _check_rerun_test_permissions(
|
|
gh_repo, pr, comment, user_perms, "rerun-group"
|
|
):
|
|
return False
|
|
|
|
if not group_names:
|
|
comment.create_reaction("confused")
|
|
pr.create_issue_comment(
|
|
"⛔ Please specify a test group: `/rerun-group <group>`\n\n"
|
|
"Example:\n"
|
|
"- `/rerun-group hicache`"
|
|
)
|
|
return False
|
|
|
|
test_specs = []
|
|
failures = []
|
|
seen = set()
|
|
for group_name in group_names:
|
|
specs, err = resolve_test_group_specs(group_name)
|
|
if err:
|
|
failures.append((group_name, err))
|
|
continue
|
|
|
|
for spec in specs:
|
|
if spec not in seen:
|
|
test_specs.append(spec)
|
|
seen.add(spec)
|
|
|
|
if failures:
|
|
comment.create_reaction("confused")
|
|
lines = [f"⛔ `{group}`: {err}" for group, err in failures]
|
|
pr.create_issue_comment("\n\n".join(lines))
|
|
return False
|
|
|
|
return handle_rerun_test(
|
|
gh_repo,
|
|
pr,
|
|
comment,
|
|
user_perms,
|
|
test_specs,
|
|
token,
|
|
skip_permission_check=True,
|
|
)
|
|
|
|
|
|
def main():
|
|
# 1. Load Environment Variables
|
|
token = get_env_var("GITHUB_TOKEN")
|
|
repo_name = get_env_var("REPO_FULL_NAME")
|
|
pr_number = int(get_env_var("PR_NUMBER"))
|
|
comment_id = int(get_env_var("COMMENT_ID"))
|
|
comment_body = get_env_var("COMMENT_BODY").strip()
|
|
user_login = get_env_var("USER_LOGIN")
|
|
|
|
# 2. Load Permissions (local file check first to avoid unnecessary API calls)
|
|
user_perms = load_permissions(user_login)
|
|
|
|
# 3. Initialize GitHub API with Auth
|
|
auth = Auth.Token(token)
|
|
g = Github(auth=auth)
|
|
|
|
repo = g.get_repo(repo_name)
|
|
pr = repo.get_pull(pr_number)
|
|
comment = repo.get_issue(pr_number).get_comment(comment_id)
|
|
|
|
# PR authors can always rerun failed CI and rerun individual UTs on their own PRs,
|
|
# even if they are not listed in CI_PERMISSIONS.json.
|
|
# Note: /tag-run-ci-label still requires CI_PERMISSIONS.json.
|
|
# Note: /rerun-test is blocked entirely for fork PRs in handle_rerun_test() itself.
|
|
if pr.user.login == user_login:
|
|
if user_perms is None:
|
|
print(
|
|
f"User {user_login} is the PR author (not in CI_PERMISSIONS.json). "
|
|
"Granting CI rerun permissions."
|
|
)
|
|
user_perms = {}
|
|
else:
|
|
print(
|
|
f"User {user_login} is the PR author and has existing CI permissions."
|
|
)
|
|
user_perms["can_rerun_failed_ci"] = True
|
|
user_perms["can_rerun_test"] = True
|
|
|
|
if not user_perms:
|
|
print(f"User {user_login} does not have any configured permissions. Exiting.")
|
|
return
|
|
|
|
# 4. Parse Command and Execute
|
|
first_line = comment_body.split("\n")[0].strip()
|
|
|
|
if first_line.startswith("/tag-run-ci-label"):
|
|
handle_tag_run_ci(repo, pr, comment, user_perms)
|
|
|
|
elif first_line.startswith("/rerun-failed-ci"):
|
|
handle_rerun_failed_ci(repo, pr, comment, user_perms)
|
|
|
|
elif first_line.startswith("/tag-and-rerun-ci"):
|
|
# Perform both actions, but suppress individual reactions
|
|
print("Processing combined command: /tag-and-rerun-ci")
|
|
|
|
tagged = handle_tag_run_ci(
|
|
repo, pr, comment, user_perms, react_on_success=False
|
|
)
|
|
|
|
# Wait for the label to propagate before triggering rerun
|
|
if tagged:
|
|
print("Waiting 5 seconds for label to propagate...")
|
|
time.sleep(5)
|
|
|
|
rerun = handle_rerun_failed_ci(
|
|
repo, pr, comment, user_perms, react_on_success=False
|
|
)
|
|
|
|
# If at least one action was successful, add the reaction here
|
|
if tagged or rerun:
|
|
comment.create_reaction("+1")
|
|
print("Combined command processed successfully; reaction added.")
|
|
else:
|
|
print("Combined command finished, but no actions were taken.")
|
|
|
|
elif first_line.startswith("/rerun-stage"):
|
|
print("/rerun-stage is deprecated; posting deprecation notice.")
|
|
comment.create_reaction("-1")
|
|
pr.create_issue_comment(
|
|
"⚠️ **`/rerun-stage` has been deprecated.**\n\n"
|
|
"Stage granularity is too coarse — a stage usually doesn't map to one "
|
|
"feature, so rerunning a stage re-pays the cost of unrelated tests. "
|
|
"If you don't know which exact test files to rerun, you shouldn't be "
|
|
"using `/rerun-stage` or `/rerun-test` in the first place.\n\n"
|
|
"**Use one of these instead:**\n"
|
|
"- **Selective tests** (you know exactly which files to rerun):\n"
|
|
" ```\n"
|
|
" /rerun-test test_foo.py test_bar.py\n"
|
|
" ```\n"
|
|
"- **Rerun only failed jobs**:\n"
|
|
" ```\n"
|
|
" /rerun-failed-ci\n"
|
|
" ```\n"
|
|
"- **Full CI rerun** (with extra coverage): add the `run-ci` or "
|
|
"`run-ci-extra` label and push a new commit (or use `/tag-and-rerun-ci`).\n\n"
|
|
"**AMD CI**: stage-level dispatch is still available via "
|
|
"Actions UI → *PR Test (AMD)* / *PR Test ROCm 7.2 (AMD)* → "
|
|
"*Run workflow* → pick a stage from the dropdown."
|
|
)
|
|
|
|
elif first_line.startswith("/rerun-group"):
|
|
group_names = first_line.split()[1:]
|
|
handle_rerun_group(repo, pr, comment, user_perms, group_names or None, token)
|
|
|
|
elif first_line.startswith("/rerun-test"):
|
|
test_specs = first_line.split()[1:]
|
|
handle_rerun_test(repo, pr, comment, user_perms, test_specs or None, token)
|
|
|
|
else:
|
|
print(f"Unknown or ignored command: {first_line}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|