[TRTLLM-12838][infra] CBTS: coverage-based test selection

jenkins/L0_MergeRequest.groovy

@@ -155,6 +155,10 @@ 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
 
 def testFilter = [
     (REUSE_TEST): gitlabParamsFromBot.get(REUSE_TEST, null),
@@ -846,8 +850,9 @@ 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"
 
-        // Shadow audit: download the latest merged touch DB and log its health + HEAD coverage gap (diagnostic only).
-        _cbtsCoverageAudit(pipeline)
+        // 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)
 
         // Ask Python which file patterns need diffs, fetch them.
         def patternsOut = sh(
@@ -872,10 +877,11 @@ def getCbtsResult(pipeline, testFilter, globalVars)
         def inputPath = "${LLM_ROOT}/cbts_input.json"
         writeFile file: inputPath, text: inputJson
 
-        def output = sh(
-            script: "cd ${LLM_ROOT} && python3 jenkins/scripts/cbts/main.py cbts_input.json",
-            returnStdout: true,
-        )
+        def mainCmd = "cd ${LLM_ROOT} && python3 jenkins/scripts/cbts/main.py cbts_input.json"
+        if (coverageDb) {
+            mainCmd += " --coverage-db ${coverageDb.path} --coverage-db-meta ${coverageDb.meta}"
+        }
+        def output = sh(script: mainCmd, returnStdout: true)
 
         def result = _cbtsParseSelectionResult(output)
         if (result.scope == null) {
@@ -917,30 +923,45 @@ def getCbtsResult(pipeline, testFilter, globalVars)
     }
 }
 
-// Download the latest merged touch DB and run coverage_audit.py on it; best-effort, never changes the CBTS decision.
+// 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.
 def _cbtsCoverageAudit(pipeline)
 {
     try {
-        def covDir = "${LLM_ROOT}/cbts_cov"
-        def url = sh(
-            script: "cd ${LLM_ROOT} && python3 jenkins/scripts/cbts/coverage_selection/artifact.py --print-url || true",
-            returnStdout: true,
-        ).trim()
-        if (!url) {
-            pipeline.echo("CBTS audit: no coverage DB artifact found — skipping")
-            return
+        // artifact.py resolves, downloads and unpacks; paths come back
+        // ${LLM_ROOT}-relative, matching the main.py caller's `cd ${LLM_ROOT}`.
+        // The checked-out revision is the PR head; its merge base is what drift is measured against.
+        def prHead = env.gitlabMergeRequestLastCommit ?: ""
+        def readyJson = ""
+        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",
+                returnStdout: true,
+            ).trim()
         }
-        sh "mkdir -p ${covDir}"
-        // wget the tarball (retrying) and extract the sqlite.
-        trtllm_utils.llmExecStepWithRetry(pipeline, script:
-            "wget -nv '${url}' -O ${covDir}/cbts_pystart_report.tar.gz && " +
-            "tar xzf ${covDir}/cbts_pystart_report.tar.gz -C ${covDir}")
-        sh "python3 ${LLM_ROOT}/jenkins/scripts/cbts/tools/coverage_audit.py " +
-           "--db ${covDir}/cbts_touchmap.sqlite"
+        if (!readyJson) {
+            pipeline.echo("CBTS audit: no coverage DB could be prepared — skipping Tier 2")
+            return null
+        }
+        def ready = new groovy.json.JsonSlurper().parseText(readyJson)
+        sh "cd ${LLM_ROOT} && python3 jenkins/scripts/cbts/tools/coverage_audit.py --db ${ready.path}"
+        return ready
     } catch (InterruptedException e) {
         throw e
     } catch (Exception e) {
         pipeline.echo("CBTS audit: skipped (non-fatal): ${e.message}")
+        return null
     }
 }
 
@@ -1033,6 +1054,8 @@ def _cbtsParseSelectionResult(String text)
         // Explicit null check preserves `false`; default True is safe.
         sanity_required: data.sanity_required != null ? data.sanity_required : true,
         perfsanity_required: data.perfsanity_required != null ? data.perfsanity_required : true,
+        // Coverage tier omits multi-GPU stages; L0_Test re-adds them under this flag.
+        enable_multi_gpu: data.enable_multi_gpu ?: false,
     ]
 }
 

jenkins/L0_Test.groovy

@@ -2549,6 +2549,56 @@ def cbtsResizeSplits(configs) {
     return resized
 }
 
+// CBTS Layer 2: replace the normal stage set with the selector's affected
+// stages while retaining the baseline sanity and multi-GPU gates.
+def filterCbtsStageJobs(parallelJobs, parallelJobsFiltered, multiGpuJobs, testFilter) {
+    def cbts = testFilter[(CBTS_RESULT)]
+    if (cbts == null) {
+        return parallelJobsFiltered
+    }
+
+    // cbtsResizeSplits renames only narrowed stages (those in
+    // affected_stage_split_counts) to `-cbts`; affected-but-not-narrowed
+    // stages keep their original name, so match each per its actual key.
+    def stageSuffix = cbts.cbts_test_db_artifact_path ? CBTS_STAGE_SUFFIX : ""
+    def narrowed = (cbts.affected_stage_split_counts ?: [:]).keySet()
+    def affectedSet = (cbts.affected_stages ?: []).collect {
+        (stageSuffix && narrowed.contains(it)) ? (it + stageSuffix) : it
+    } as Set
+    def needsSanity = cbts.sanity_required
+    def needsPerfSanity = cbts.perfsanity_required
+    def filtered = parallelJobs.findAll { key, _ ->
+        if (key.contains("-OnDemand-")) {
+            return false
+        }
+        if (key =~ /Post-Merge/) return affectedSet.contains(key)
+        return affectedSet.contains(key) ||
+               (needsSanity && key =~ /PackageSanityCheck/) ||
+               (needsPerfSanity && key =~ /PerfSanity/)
+    }
+    if (affectedSet.isEmpty()) {
+        if (filtered.isEmpty()) {
+            echo "CBTS [${cbts.scope}]: trigger-mode mismatch + nothing force-kept → no-op"
+        } else {
+            echo "CBTS [${cbts.scope}]: trigger-mode mismatch — running " +
+                 "${filtered.size()} force-kept stage(s) only"
+        }
+    } else if (filtered) {
+        echo "CBTS [${cbts.scope}]: limiting to ${filtered.size()} stages " +
+             "(sanity_required=${needsSanity}, perfsanity_required=${needsPerfSanity})"
+    } else {
+        echo "CBTS [${cbts.scope}]: empty stage set after filtering"
+    }
+
+    // The coverage tier omits multi-GPU; re-add it under the baseline gate.
+    if (cbts.enable_multi_gpu && testFilter[(MULTI_GPU_FILE_CHANGED)]) {
+        filtered += multiGpuJobs
+        echo "CBTS [${cbts.scope}]: multi-GPU file changed → running " +
+             "${multiGpuJobs.size()} multi-GPU stage(s) at baseline"
+    }
+    return filtered
+}
+
 // True when an exception indicates the K8s dispatcher pod this SLURM stage runs
 // inside died mid-run -- kubelet eviction, container termination, or the JNLP
 // agent otherwise going offline. Retrying inside such a pod is futile (every
@@ -6539,41 +6589,8 @@ def launchTestJobs(pipeline, testFilter, globalVars)
         checkStageNameSet(testFilter[(EXTRA_STAGE_LIST)], fullSet, EXTRA_STAGE_LIST)
     }
 
-    // CBTS Layer 2: replace `parallelJobsFiltered` with affected stages plus
-    // PackageSanityCheck (kept iff sanity_required) and PerfSanity (kept iff
-    // perfsanity_required). Pure -Perf- stages run only when CBTS selects them
-    // (present in affected_stages). Post-Merge stages are never force-kept;
-    // they only run when explicitly listed in affected_stages.
-    def cbts = testFilter[(CBTS_RESULT)]
-    if (cbts != null) {
-        // Match the -cbts rename cbtsResizeSplits applies to narrowed stages.
-        def stageSuffix = cbts.cbts_test_db_artifact_path ? CBTS_STAGE_SUFFIX : ""
-        def affectedSet = (cbts.affected_stages ?: []).collect { it + stageSuffix } as Set
-        def needsSanity = cbts.sanity_required
-        def needsPerfSanity = cbts.perfsanity_required
-        parallelJobsFiltered = parallelJobs.findAll { key, _ ->
-            if (key.contains("-OnDemand-")) {
-                return false
-            }
-            if (key =~ /Post-Merge/) return affectedSet.contains(key)
-            return affectedSet.contains(key) ||
-                   (needsSanity && key =~ /PackageSanityCheck/) ||
-                   (needsPerfSanity && key =~ /PerfSanity/)
-        }
-        if (affectedSet.isEmpty()) {
-            if (parallelJobsFiltered.isEmpty()) {
-                echo "CBTS [${cbts.scope}]: trigger-mode mismatch + nothing force-kept → no-op"
-            } else {
-                echo "CBTS [${cbts.scope}]: trigger-mode mismatch — running " +
-                     "${parallelJobsFiltered.size()} force-kept stage(s) only"
-            }
-        } else if (parallelJobsFiltered) {
-            echo "CBTS [${cbts.scope}]: limiting to ${parallelJobsFiltered.size()} stages " +
-                 "(sanity_required=${needsSanity}, perfsanity_required=${needsPerfSanity})"
-        } else {
-            echo "CBTS [${cbts.scope}]: empty stage set after filtering"
-        }
-    }
+    parallelJobsFiltered = filterCbtsStageJobs(
+        parallelJobs, parallelJobsFiltered, multiGpuJobs, testFilter)
 
     if (globalVars[RUN_MODE] == "nightly_release") {
         parallelJobsFiltered = sanityCheckJobs

jenkins/scripts/cbts/README.md

@@ -5,6 +5,16 @@ run, based on what the PR changed. New rules are added in Python only.
 
 ---
 
+## Selection tiers
+
+| Tier | Where | Basis |
+|---|---|---|
+| **1. Rules** | `rules/` | Path patterns and diffs. Each rule claims the files it understands and narrows to the blocks they affect. |
+| **2. Coverage** | `coverage_selection/`, `coverage_tier.py` | Runs only on what Tier 1 left unclaimed (the *residual*), and only when every residual file is core Python present in the touch DB. Maps changed lines to qualnames and removes entries no changed function reaches. Declines to the full run otherwise. See `coverage_selection/SELECTION.md`. |
+
+The touch DB is produced by the post-merge collection under `coverage_utils/`;
+see `coverage_utils/COLLECTION.md` for what it does and does not record.
+
 ## Consumption layers
 
 CBTS narrows test cases only; Build always runs.
@@ -75,8 +85,18 @@ jenkins/scripts/cbts/
 │   ├── visual_gen_rule.py
 │   ├── spec_dec_rule.py
 │   └── out_of_scope_rule.py
+├── coverage_tier.py       Tier 2 entry: applies the selector to the test-db YAMLs, classifies every candidate entry
+├── coverage_selection/
+│   ├── SELECTION.md       how a decision is made: qualname concepts, decline gates, narrowing
+│   ├── selector.py        CoverageSelector.decide(): changed lines → qualnames → impacted / skippable per stage family
+│   ├── qualname_map.py    changed lines → co_qualname, plus the import-time and closure classifications
+│   ├── touch_db.py        read-only accessor over cbts_touchmap.sqlite + the untrusted-capture signals
+│   └── artifact.py        resolve which post-merge touch DB to use (by collected revision)
+├── coverage_utils/        post-merge collection that produces the touch DB (see its README / COLLECTION.md)
 └── tools/
     ├── dryrun.py          replay CBTS over historical commits → per-PR summary.txt + filtered YAMLs + INDEX.md (debug only)
+    ├── coverage_audit.py  report a touch DB's format, scale, untrusted rate and HEAD coverage gap
+    ├── coverage_explain.py  explain one commit's decision case by case (delegates to CoverageSelector)
     └── report_cbts_decision.py  post the decision (hit-stage count, case-level skip rate, fallback) to OpenSearch for CI-health monitoring
 ```
 
@@ -294,6 +314,14 @@ CBTS defers to the existing filter chain when:
   but cannot decide; e.g. testdef blast-radius cap, testlist structural
   YAML edit)
 - Combined scope is `None` (incompatible mix)
+- Tier 2 declines: a residual file is not core Python, is absent from the
+  touch DB, has an import-executed change (module / class body, signature or
+  decorator line), has no usable patch, has unparsable source, or has a closure
+  change with no wider row set (see `coverage_selection/SELECTION.md` §3-4)
+- No touch DB artifact could be resolved — Tier 2 never runs
+- The resolved DB sits more than `--coverage-max-drift` commits from the PR's
+  base commit, on either side, or an unmeasurable distance from it — Tier 2
+  declines (`coverage_freshness` = `stale` / `unknown`)
 - Layer 3 narrowing would empty a block — block keeps original tests
 - `cbts_test_db` tarball upload or download/extraction fails — renderTestDB falls back to source
 - Narrowed YAML missing/empty on a stage agent — renderTestDB falls back

jenkins/scripts/cbts/blocks.py

@@ -25,6 +25,7 @@
 import json
 import math
 import re
+import shutil
 from dataclasses import dataclass, field
 from fnmatch import fnmatch
 from pathlib import Path
@@ -433,6 +434,9 @@ def _classify_map_var(var_name: str) -> Optional[str]:
     return None
 
 
+# Stages CBTS always runs, whatever the decision.
+ALWAYS_RUN_STAGE_PREFIX = "CPU-"
+
 # Backend name -> mako value. Same patterns as getMakoArgsFromStageName in
 # jenkins/L0_Test.groovy (line ~2079). IMPORTANT: keep this list in sync.
 _BACKEND_PATTERNS = [
@@ -441,6 +445,7 @@ def _classify_map_var(var_name: str) -> Optional[str]:
     ("-Triton-", "triton"),
     ("-FMHA-", "fmha"),
     ("-AutoDeploy-", "autodeploy"),
+    ("-Generic-", "generic"),
     ("-Verl-", "verl"),
 ]
 
@@ -488,6 +493,9 @@ def derive_mako_from_stage(stage_name: str) -> dict[str, str]:
         mako["gpu"] = gpu_match.group(1).lower()
     count_match = _GPU_COUNT_RE.search(stage_name)
     mako["system_gpu_count"] = count_match.group(1) if count_match else "1"
+    # renderTestDB hardcodes system_gpu_count=0 for CPU- stages (no N_GPUs token).
+    if stage_name.startswith("CPU-"):
+        mako["system_gpu_count"] = "0"
 
     return mako
 
@@ -749,6 +757,8 @@ def write_filtered_test_db(
     tests are kept (prevents silent skip from typo'd waive ids or granularity
     mismatch). The block itself is still kept either way.
     """
+    # Clear first: a leftover YAML from an earlier run would ship in the artifact.
+    shutil.rmtree(output_dir, ignore_errors=True)
     output_dir.mkdir(parents=True, exist_ok=True)
 
     affected_stems = {stem for stem, _ in block_filters}