[None][infra] cbts-v2 coverage pilot allowlist

jenkins/L0_MergeRequest.groovy

@@ -155,10 +155,6 @@ def DISABLE_CBTS = "disable_cbts"
 // Kill switch for CBTS per-test coverage; official post-merge pipeline only, single-GPU stages only in Phase 1.
 @Field
 def ENABLE_CBTS_COVERAGE = true
-// Rollout switch for pre-merge Tier 2 coverage-based narrowing. Keep collection
-// enabled above while this remains off so a later pilot allowlist has fresh data.
-@Field
-def ENABLE_CBTS_COVERAGE_TIER = false
 @Field
 def OSS_COMPLIANCE_FILE_CHANGED = "oss_compliance_file_changed"
 
@@ -853,9 +849,8 @@ def getCbtsResult(pipeline, testFilter, globalVars)
         // pyyaml is needed by main.py's blocks.py to parse test-db YAMLs.
         sh "apt-get update -qq && apt-get install -y -qq python3-yaml"
 
-        // Download the touch DB only when Tier 2 is enabled. Tier 1 rules still
-        // run while the coverage tier is disabled during the initial rollout.
-        def coverageDb = _cbtsCoverageDb(pipeline)
+        // Download the touch DB only for PRs in the coverage-tier pilot.
+        def coverageDb = _cbtsCoverageAudit(pipeline)
 
         // Ask Python which file patterns need diffs, fetch them.
         def patternsOut = sh(
@@ -926,18 +921,8 @@ def getCbtsResult(pipeline, testFilter, globalVars)
     }
 }
 
-// Resolve the optional Tier 2 input behind an explicit rollout gate. Keeping
-// this separate makes the follow-up pilot allowlist a small policy change.
-def _cbtsCoverageDb(pipeline)
-{
-    if (!ENABLE_CBTS_COVERAGE_TIER) {
-        pipeline.echo("CBTS: coverage tier disabled — running Tier 1 only")
-        return null
-    }
-    return _cbtsCoverageAudit(pipeline)
-}
-
-// Fetch the touch DB and audit it; artifact.py's {path, meta} verbatim, or null on failure.
+// Check pilot eligibility, then fetch and audit the touch DB; artifact.py's
+// {path, meta} verbatim, or null on failure.
 def _cbtsCoverageAudit(pipeline)
 {
     try {
@@ -946,12 +931,23 @@ def _cbtsCoverageAudit(pipeline)
         // The checked-out revision is the PR head; its merge base is what drift is measured against.
         def prHead = env.gitlabMergeRequestLastCommit ?: ""
         def readyJson = ""
+        def pilotEligible = false
         withCredentials([usernamePassword(credentialsId: 'github-cred-trtllm-ci', usernameVariable: 'NOT_USED_YET', passwordVariable: 'GITHUB_API_TOKEN')]) {
-            readyJson = sh(
-                script: "cd ${LLM_ROOT} && python3 jenkins/scripts/cbts/coverage_selection/artifact.py " +
-                        "--prepare cbts_cov${prHead ? " --pr-head ${prHead}" : ""} || true",
+            pilotEligible = sh(
+                script: "cd ${LLM_ROOT} && python3 jenkins/scripts/cbts/coverage_pilot.py",
                 returnStdout: true,
-            ).trim()
+            ).trim() == "true"
+            if (pilotEligible) {
+                readyJson = sh(
+                    script: "cd ${LLM_ROOT} && python3 jenkins/scripts/cbts/coverage_selection/artifact.py " +
+                            "--prepare cbts_cov${prHead ? " --pr-head ${prHead}" : ""} || true",
+                    returnStdout: true,
+                ).trim()
+            }
+        }
+        if (!pilotEligible) {
+            pipeline.echo("CBTS: coverage tier disabled for this PR — running Tier 1 only")
+            return null
         }
         if (!readyJson) {
             pipeline.echo("CBTS audit: no coverage DB could be prepared — skipping Tier 2")

jenkins/scripts/cbts/coverage_pilot.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Fail-closed pilot allowlist for the CBTS coverage tier."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+import urllib.error
+import urllib.request
+from collections.abc import Mapping, Set
+from typing import Optional
+
+PILOT_USERS: frozenset[str] = frozenset(
+    {
+        "crazydemo",
+        "QiJune",
+        "sunnyqgg",
+        "Barry-Delaney",
+        "xxi-nv",
+        "leslie-fang25",
+        "rosong11",
+        "tongyuantongyu",
+    }
+)
+
+_GITHUB_TOKEN_ENV = "GITHUB_API_TOKEN"
+_TRIGGER_PHRASE_ENV = "gitlabTriggerPhrase"
+_PR_API_URL_RE = re.compile(r"https://api\.github\.com/repos/NVIDIA/TensorRT-LLM/pulls/\d+\Z")
+_REQUEST_TIMEOUT_SECONDS = 15
+
+
+def pr_api_url_from_trigger_phrase(trigger_phrase: str) -> str:
+    """Return the bot-provided GitHub PR API URL, or an empty string."""
+    if not trigger_phrase:
+        return ""
+    try:
+        payload = json.loads(trigger_phrase)
+    except json.JSONDecodeError:
+        return ""
+    if not isinstance(payload, Mapping):
+        return ""
+    value = payload.get("github_pr_api_url")
+    return value.strip() if isinstance(value, str) else ""
+
+
+def evaluate_pr_info(pr_info: object, pilot_users: Set[str] = PILOT_USERS) -> tuple[bool, str, str]:
+    """Return ``(eligible, normalized_login, reason)`` for one PR response."""
+    if not isinstance(pr_info, Mapping):
+        return False, "", "PR API response is not an object"
+    user = pr_info.get("user")
+    if not isinstance(user, Mapping):
+        return False, "", "PR API response has no user"
+    raw_login = user.get("login")
+    login = raw_login.strip() if isinstance(raw_login, str) else ""
+    if not login:
+        return False, "", "PR API response has no author login"
+    normalized_pilot_users = {pilot_user.casefold() for pilot_user in pilot_users}
+    if login.casefold() not in normalized_pilot_users:
+        return False, login, "author is not allowlisted"
+    return True, login, "author is allowlisted"
+
+
+def check_pilot_eligibility(
+    pr_api_url: str,
+    token: str = "",
+    pilot_users: Set[str] = PILOT_USERS,
+) -> tuple[bool, str, str]:
+    """Fetch one trusted GitHub PR endpoint and evaluate its author."""
+    if not _PR_API_URL_RE.fullmatch(pr_api_url):
+        return False, "", "missing or unexpected GitHub PR API URL"
+
+    headers = {
+        "Accept": "application/vnd.github+json",
+        "User-Agent": "tensorrt-llm-cbts-pilot",
+    }
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    request = urllib.request.Request(pr_api_url, headers=headers)
+    try:
+        with urllib.request.urlopen(request, timeout=_REQUEST_TIMEOUT_SECONDS) as response:
+            pr_info = json.loads(response.read())
+    except (
+        json.JSONDecodeError,
+        UnicodeDecodeError,
+        urllib.error.HTTPError,
+        urllib.error.URLError,
+        TimeoutError,
+        OSError,
+    ) as error:
+        return False, "", f"PR author lookup failed: {error}"
+    return evaluate_pr_info(pr_info, pilot_users)
+
+
+def main(argv: Optional[list[str]] = None) -> int:
+    parser = argparse.ArgumentParser(description="Check CBTS coverage-tier pilot eligibility.")
+    parser.add_argument(
+        "--pr-api-url",
+        default=None,
+        help="GitHub PR API URL; defaults to the bot trigger payload.",
+    )
+    args = parser.parse_args(argv)
+
+    pr_api_url = args.pr_api_url
+    if pr_api_url is None:
+        pr_api_url = pr_api_url_from_trigger_phrase(os.environ.get(_TRIGGER_PHRASE_ENV, ""))
+    eligible, login, reason = check_pilot_eligibility(
+        pr_api_url,
+        token=os.environ.get(_GITHUB_TOKEN_ENV, ""),
+    )
+    print(
+        "CBTS coverage pilot: "
+        f"pr_author={login or 'unknown'}, eligible={str(eligible).lower()}, reason={reason}",
+        file=sys.stderr,
+    )
+    # Jenkins consumes stdout; diagnostics stay on stderr in the console log.
+    print(str(eligible).lower())
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/unittest/scripts/test_cbts_coverage_pilot.py

@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import importlib.util
+import json
+import urllib.error
+import urllib.request
+from pathlib import Path
+from types import ModuleType, TracebackType
+from typing import NoReturn, TypeAlias, Union
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent
+PILOT_PATH = REPO_ROOT / "jenkins" / "scripts" / "cbts" / "coverage_pilot.py"
+PR_API_URL = "https://api.github.com/repos/NVIDIA/TensorRT-LLM/pulls/123"
+TEST_PILOT_USERS = frozenset({"pilot-user"})
+JSONValue: TypeAlias = Union[
+    None,
+    bool,
+    int,
+    float,
+    str,
+    list["JSONValue"],
+    dict[str, "JSONValue"],
+]
+
+
+@pytest.fixture(scope="module")
+def pilot_module() -> ModuleType:
+    spec = importlib.util.spec_from_file_location("coverage_pilot", PILOT_PATH)
+    assert spec is not None
+    assert spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class _Response:
+    def __init__(self, payload: JSONValue) -> None:
+        self._payload = json.dumps(payload).encode()
+
+    def __enter__(self) -> "_Response":
+        return self
+
+    def __exit__(
+        self,
+        _exc_type: type[BaseException] | None,
+        _exc_value: BaseException | None,
+        _traceback: TracebackType | None,
+    ) -> None:
+        return None
+
+    def read(self) -> bytes:
+        return self._payload
+
+
+@pytest.mark.parametrize(
+    ("pr_info", "expected"),
+    (
+        ({"user": {"login": " Pilot-User "}}, (True, "Pilot-User", "author is allowlisted")),
+        (
+            {"user": {"login": "someone-else"}},
+            (False, "someone-else", "author is not allowlisted"),
+        ),
+        ({"user": {}}, (False, "", "PR API response has no author login")),
+        ({}, (False, "", "PR API response has no user")),
+        ([], (False, "", "PR API response is not an object")),
+    ),
+)
+def test_evaluate_pr_info(
+    pilot_module: ModuleType,
+    pr_info: JSONValue,
+    expected: tuple[bool, str, str],
+) -> None:
+    assert pilot_module.evaluate_pr_info(pr_info, pilot_users=TEST_PILOT_USERS) == expected
+
+
+def test_check_pilot_eligibility_uses_token(
+    pilot_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    def urlopen(request: urllib.request.Request, timeout: int) -> _Response:
+        assert request.full_url == PR_API_URL
+        assert request.get_header("Authorization") == "Bearer token"
+        assert timeout == 15
+        return _Response({"user": {"login": "pilot-user"}})
+
+    monkeypatch.setattr(pilot_module.urllib.request, "urlopen", urlopen)
+
+    assert pilot_module.check_pilot_eligibility(
+        PR_API_URL, token="token", pilot_users=TEST_PILOT_USERS
+    ) == (
+        True,
+        "pilot-user",
+        "author is allowlisted",
+    )
+
+
+def test_check_pilot_eligibility_rejects_untrusted_url(
+    pilot_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    def unexpected_urlopen(
+        _request: urllib.request.Request,
+        _timeout: int,
+    ) -> NoReturn:
+        raise AssertionError("untrusted URLs must not be requested")
+
+    monkeypatch.setattr(pilot_module.urllib.request, "urlopen", unexpected_urlopen)
+
+    eligible, login, reason = pilot_module.check_pilot_eligibility(
+        "https://example.com/pulls/123", token="token"
+    )
+    assert not eligible
+    assert not login
+    assert reason == "missing or unexpected GitHub PR API URL"
+
+
+def test_check_pilot_eligibility_fails_closed_on_api_error(
+    pilot_module: ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    def urlopen(_request: urllib.request.Request, *, timeout: int) -> NoReturn:
+        assert timeout == 15
+        raise urllib.error.URLError("unavailable")
+
+    monkeypatch.setattr(pilot_module.urllib.request, "urlopen", urlopen)
+
+    eligible, login, reason = pilot_module.check_pilot_eligibility(PR_API_URL, token="token")
+    assert not eligible
+    assert not login
+    assert reason.startswith("PR author lookup failed:")
+
+
+def test_main_reads_bot_trigger_payload(
+    pilot_module: ModuleType,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    trigger_phrase = json.dumps({"github_pr_api_url": PR_API_URL})
+    monkeypatch.setenv("gitlabTriggerPhrase", trigger_phrase)
+    monkeypatch.setenv("GITHUB_API_TOKEN", "token")
+
+    def check_pilot_eligibility(
+        pr_api_url: str,
+        *,
+        token: str,
+    ) -> tuple[bool, str, str]:
+        assert pr_api_url == PR_API_URL
+        assert token == "token"
+        return True, "pilot-user", "author is allowlisted"
+
+    monkeypatch.setattr(pilot_module, "check_pilot_eligibility", check_pilot_eligibility)
+
+    assert pilot_module.main([]) == 0
+    captured = capsys.readouterr()
+    assert captured.out == "true\n"
+    assert "pr_author=pilot-user, eligible=true" in captured.err