diff --git a/jenkins/L0_MergeRequest.groovy b/jenkins/L0_MergeRequest.groovy index 57bc4123c2d5..962a87152339 100644 --- a/jenkins/L0_MergeRequest.groovy +++ b/jenkins/L0_MergeRequest.groovy @@ -774,6 +774,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) + // Ask Python which file patterns need diffs, fetch them. def patternsOut = sh( script: "cd ${LLM_ROOT} && python3 jenkins/scripts/cbts/main.py --list-needed-diffs", @@ -842,6 +845,33 @@ 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. +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 + } + 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" + } catch (InterruptedException e) { + throw e + } catch (Exception e) { + pipeline.echo("CBTS audit: skipped (non-fatal): ${e.message}") + } +} + // Post one CBTS decision record to OpenSearch (best-effort; never blocks CI). // decisionJson null for deferred; reason used only then. Context/creds via env. def _cbtsReportDecision(pipeline, globalVars, String status, String reason, String decisionJson) diff --git a/jenkins/scripts/cbts/coverage_selection/artifact.py b/jenkins/scripts/cbts/coverage_selection/artifact.py new file mode 100644 index 000000000000..74bb0159fffb --- /dev/null +++ b/jenkins/scripts/cbts/coverage_selection/artifact.py @@ -0,0 +1,182 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. +"""Resolve and fetch the latest merged CBTS touch DB from Artifactory. + +The tarball is uploaded per post-merge run to +`//cbts-coverage/cbts_pystart_report.tar.gz` (sqlite at +the tar root plus `cbts_report/`). + +`latest_tarball_url()` reads the newest build number from the Jenkins REST API, +then walks builds down, probing Artifactory with a 1-byte ranged GET until it +finds one whose tarball exists. + +Two entry points for the Groovy wiring: + * `--print-url` — resolve and print the tarball URL only (no download). + * `--dest DIR` — download + extract, printing the local sqlite path. +""" + +from __future__ import annotations + +import argparse +import json +import shutil +import sys +import tarfile +import urllib.error +import urllib.request +from pathlib import Path +from typing import Optional + +# Merged-artifact base for the main-branch L0_PostMerge job. +ARTIFACT_BASE = "sw-tensorrt-generic/llm-artifacts/LLM/main/L0_PostMerge" +TARBALL_NAME = "cbts_pystart_report.tar.gz" +SQLITE_NAME = "cbts_touchmap.sqlite" + +_URM = "https://urm.nvidia.com/artifactory" +_JENKINS_BASE = "https://prod.blsm.nvidia.com/sw-tensorrt-top-1/job/LLM/job/main/job/L0_PostMerge" +# Max builds to walk back when recent builds have no tarball. +_MAX_PROBE = 10 +# Per-request timeout in seconds. +_TIMEOUT = 15 + + +def _get(url: str) -> tuple[Optional[int], Optional[bytes]]: + try: + with urllib.request.urlopen(url, timeout=_TIMEOUT) as resp: + return resp.status, resp.read() + except urllib.error.HTTPError as e: + return e.code, None + except OSError as e: + print(f"[artifact] error fetching {url}: {e}", file=sys.stderr) + return None, None + + +def _exists(url: str) -> bool: + """True if the artifact exists — a 1-byte ranged GET; 200/206 means present.""" + req = urllib.request.Request(url, headers={"Range": "bytes=0-0"}) + try: + with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp: + return resp.status in (200, 206) + except urllib.error.HTTPError: + return False + except OSError as e: + print(f"[artifact] error probing {url}: {e}", file=sys.stderr) + return False + + +def latest_build_number(jenkins_base: str = _JENKINS_BASE) -> Optional[int]: + """Newest build number via the Jenkins REST API (lastBuild, then lastCompletedBuild).""" + for kind in ("lastBuild", "lastCompletedBuild"): + status, data = _get(f"{jenkins_base}/{kind}/api/json") + if status == 200 and data: + try: + return int(json.loads(data)["number"]) + except (json.JSONDecodeError, KeyError, ValueError): + pass + return None + + +def tarball_url(build: int, artifact_base: str = ARTIFACT_BASE) -> str: + return f"{_URM}/{artifact_base}/{build}/cbts-coverage/{TARBALL_NAME}" + + +def latest_tarball_url( + artifact_base: str = ARTIFACT_BASE, + jenkins_base: str = _JENKINS_BASE, + max_probe: int = _MAX_PROBE, +) -> Optional[str]: + """URL of the newest build whose coverage tarball actually exists, or None.""" + build = latest_build_number(jenkins_base) + if build is None: + print("[artifact] could not resolve latest build number", file=sys.stderr) + return None + floor = max(0, build - max_probe) + while build > floor: + url = tarball_url(build, artifact_base) + if _exists(url): + return url + print(f"[artifact] build {build} has no tarball, trying {build - 1}", file=sys.stderr) + build -= 1 + print(f"[artifact] no tarball in the last {max_probe} builds", file=sys.stderr) + return None + + +def extract_touch_db(tarball: Path | str, dest_dir: Path | str) -> Optional[Path]: + """Extract `cbts_touchmap.sqlite` from a downloaded tarball; return its path.""" + dest_dir = Path(dest_dir) + dest_dir.mkdir(parents=True, exist_ok=True) + with tarfile.open(tarball) as tf: + member = next((m for m in tf.getmembers() if m.name.endswith(SQLITE_NAME)), None) + if member is None: + return None + member.name = SQLITE_NAME + tf.extract(member, dest_dir) + return dest_dir / SQLITE_NAME + + +def fetch_latest_touch_db(dest_dir: Path | str, url: Optional[str] = None) -> Optional[Path]: + """Download + extract the latest post-merge touch DB; return local sqlite Path or None. + + `url` pins an explicit tarball (skips latest-build resolution); any failure returns None. + """ + dest_dir = Path(dest_dir) + dest_dir.mkdir(parents=True, exist_ok=True) + url = url or latest_tarball_url() + if url is None: + return None + tarball = dest_dir / TARBALL_NAME + try: + with urllib.request.urlopen(url, timeout=_TIMEOUT) as resp, open(tarball, "wb") as f: + shutil.copyfileobj(resp, f) + return extract_touch_db(tarball, dest_dir) + except OSError as e: + print(f"[artifact] download/extract failed {url}: {e}", file=sys.stderr) + return None + + +def main(argv: Optional[list[str]] = None) -> int: + ap = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + ap.add_argument("--dest", help="download + extract into DIR; prints the local sqlite path") + ap.add_argument( + "--print-url", action="store_true", help="resolve and print the tarball URL only" + ) + ap.add_argument( + "--build", type=int, default=None, help="pin a build number (skip auto-resolve)" + ) + args = ap.parse_args(argv) + + url = tarball_url(args.build) if args.build is not None else None + + if args.print_url: + url = url or latest_tarball_url() + if url is None: + return 1 + print(url) + return 0 + + if args.dest: + path = fetch_latest_touch_db(args.dest, url=url) + if path is None: + return 1 + print(path) + return 0 + + ap.error("one of --print-url or --dest is required") + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/jenkins/scripts/cbts/coverage_selection/touch_db.py b/jenkins/scripts/cbts/coverage_selection/touch_db.py new file mode 100644 index 000000000000..d6ff12abc686 --- /dev/null +++ b/jenkins/scripts/cbts/coverage_selection/touch_db.py @@ -0,0 +1,232 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. +"""Read-only accessor for the merged CBTS touch DB (`cbts_touchmap.sqlite`). + +Implements the TOUCH_DB_CONTRACT.md queries. Two real-data details: + - the `test` column is `/` (stage-prefixed); + - unit tests are recorded wrapped as + `test_unittests.py::test_unittests_v2[]`, while test-db YAML lists + the bare `` entry. +""" + +from __future__ import annotations + +import re +import sqlite3 +from pathlib import Path +from typing import Optional + +from blocks import normalize_test_id + +# Canonicalize an absolute co_filename to the DB `file` form (`tensorrt_llm/...`). +_CANON_RE = re.compile(r"(tensorrt_llm/.*)$") + +# A DB test value is `/`; unit tests wrap the inner entry. +_UNITTEST_WRAP_RE = re.compile(r"::test_unittests_v2\[(?P.+)\]$") + +# Completeness-heuristic constants consumed by `untrusted_tests()`. +_WORKER_SENTINEL = "tensorrt_llm/_torch/pyexecutor/py_executor.py" +_LAUNCH_MARKERS: tuple[tuple[str, str], ...] = ( + ("tensorrt_llm/llmapi/llm.py", "generate"), + ("tensorrt_llm/executor/executor.py", "GenerationExecutor.generate"), +) +_SERVING_PATH_MARKERS: tuple[str, ...] = ("disaggregated/",) +_MIN_FUNCS = 30 + + +def canon(path: str) -> str: + """Canonicalize a path to the DB's `file` form (`tensorrt_llm/...`).""" + m = _CANON_RE.search(path) + return m.group(1) if m else path + + +def split_stage(test: str) -> tuple[str, str]: + """Split a DB `test` value `/` into `(stage, nodeid)`; `("", test)` if no `/`.""" + stage, sep, nodeid = test.partition("/") + return (stage, nodeid) if sep else ("", test) + + +def unwrap_unittest(nodeid: str) -> Optional[str]: + """Return the inner `unittest/...` entry of a wrapped unittest nodeid, else None. + + `test_unittests.py::test_unittests_v2[unittest/x.py -m "part0"]` + -> `unittest/x.py -m "part0"` + """ + m = _UNITTEST_WRAP_RE.search(nodeid) + return m.group("inner") if m else None + + +def db_key(entry: str) -> Optional[str]: + """Map a test-db YAML `tests:` entry to the DB nodeid form, or None if not 1:1. + + Unit tests wrap as `test_unittests.py::test_unittests_v2[]`; a `-k` + keyword entry expands to many nodeids (no single DB key) -> None. + """ + e = normalize_test_id(entry) + if e.startswith("unittest/"): + return f"test_unittests.py::test_unittests_v2[{e}]" + if " -k " in e: + return None + return e + + +class TouchDB: + """Read-only view over a merged `cbts_touchmap.sqlite`.""" + + def __init__(self, conn: sqlite3.Connection) -> None: + self._conn = conn + + @classmethod + def open(cls, sqlite_path: Path | str) -> "TouchDB": + """Open the DB read-only (`mode=ro`) and verify the `touch` schema.""" + uri = f"file:{Path(sqlite_path).resolve()}?mode=ro" + conn = sqlite3.connect(uri, uri=True) + cols = {row[1] for row in conn.execute("PRAGMA table_info(touch)")} + if not {"test", "file", "qualname"} <= cols: + conn.close() + raise ValueError(f"unexpected touch schema, columns={sorted(cols)}") + return cls(conn) + + def close(self) -> None: + self._conn.close() + + def __enter__(self) -> "TouchDB": + return self + + def __exit__(self, *_exc) -> None: + self.close() + + # -- meta (every key optional; read with a default) -- + + def meta(self, key: str, default: Optional[str] = None) -> Optional[str]: + try: + row = self._conn.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone() + except sqlite3.OperationalError: + return default + return row[0] if row is not None else default + + def schema_version(self) -> Optional[str]: + return self.meta("schema_version") + + def collection_commit(self) -> Optional[str]: + """Commit the DB was collected at, or None if not recorded.""" + return self.meta("commit") or self.meta("collection_commit") + + # -- reverse lookup (the core of selection); always `test != ''` -- + + def tests_touching_file(self, file: str) -> set[str]: + """Stage-prefixed tests that entered any function in `file` (file-level).""" + return { + row[0] + for row in self._conn.execute( + "SELECT DISTINCT test FROM touch WHERE file=? AND test!=''", (file,) + ) + } + + def tests_touching_func(self, file: str, qualname: str) -> set[str]: + """Stage-prefixed tests that entered `qualname` in `file` (function-level).""" + return { + row[0] + for row in self._conn.execute( + "SELECT DISTINCT test FROM touch WHERE file=? AND qualname=? AND test!=''", + (file, qualname), + ) + } + + def file_has_touch_rows(self, file: str) -> bool: + """True iff any instrumented test entered a function in `file`.""" + row = self._conn.execute( + "SELECT 1 FROM touch WHERE file=? AND test!='' LIMIT 1", (file,) + ).fetchone() + return row is not None + + # -- universe / per-stage -- + + def known_tests(self) -> set[str]: + """Every stage-prefixed test with coverage data.""" + return { + row[0] for row in self._conn.execute("SELECT DISTINCT test FROM touch WHERE test!=''") + } + + def per_test_footprint(self) -> dict[str, int]: + """`{test -> functions entered}` over all stage-prefixed tests.""" + return { + row[0]: row[1] + for row in self._conn.execute( + "SELECT test, COUNT(*) FROM touch WHERE test!='' GROUP BY test" + ) + } + + def instrumented_stages(self) -> set[str]: + """Stage names the DB has data for — the stages coverage may narrow.""" + return {stage for stage, _ in map(split_stage, self.known_tests()) if stage} + + def known_by_stage(self) -> dict[str, set[str]]: + """`{stage -> {bare nodeid, ...}}` over all known tests.""" + out: dict[str, set[str]] = {} + for test in self.known_tests(): + stage, nodeid = split_stage(test) + if stage: + out.setdefault(stage, set()).add(nodeid) + return out + + # -- forward lookup (debug / explain-why) -- + + def files_touched_by(self, test: str) -> list[tuple[str, str]]: + """`(file, qualname)` rows for a stage-prefixed `test`.""" + return [ + (row[0], row[1]) + for row in self._conn.execute("SELECT file, qualname FROM touch WHERE test=?", (test,)) + ] + + # -- coverage-completeness heuristic -- + + def untrusted_tests( + self, + worker_file: str, + launch_markers: tuple[tuple[str, str], ...], + serving_path_markers: tuple[str, ...], + min_funcs: int, + ) -> set[str]: + """Stage-prefixed tests whose per-test capture looks incomplete (must always run). + + Flags a test that drove execution/serving but is missing `worker_file` — + matched by a `launch_markers` `(file, qualname_substring)` call or a + `serving_path_markers` nodeid substring — or that entered fewer than + `min_funcs` functions total. + """ + drove_execution: set[str] = set() + for file, qual_substr in launch_markers: + drove_execution |= { + row[0] + for row in self._conn.execute( + "SELECT DISTINCT test FROM touch WHERE file=? AND qualname LIKE ? AND test!=''", + (file, f"%{qual_substr}%"), + ) + } + if serving_path_markers: + drove_execution |= { + test + for test in self.known_tests() + if any(marker in test for marker in serving_path_markers) + } + missing_worker = drove_execution - self.tests_touching_file(worker_file) + tiny = { + row[0] + for row in self._conn.execute( + "SELECT test FROM touch WHERE test!='' GROUP BY test HAVING COUNT(*) < ?", + (min_funcs,), + ) + } + return missing_worker | tiny diff --git a/jenkins/scripts/cbts/tools/coverage_audit.py b/jenkins/scripts/cbts/tools/coverage_audit.py new file mode 100644 index 000000000000..b9e2309b6152 --- /dev/null +++ b/jenkins/scripts/cbts/tools/coverage_audit.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. +r"""Audit a CBTS touch DB (`cbts_touchmap.sqlite`) — format, scale, and coverage completeness. + +Reports the format (stage prefix, schema_version, collection commit), scale, +per-stage known counts, the per-test footprint distribution, and the tests +whose capture looks incomplete (the same heuristic the selector uses). + +Example:: + + python3 jenkins/scripts/cbts/tools/coverage_audit.py \\ + --db cbts_touchmap.sqlite --list-untrusted +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +THIS = Path(__file__).resolve() +CBTS = THIS.parent.parent +sys.path.insert(0, str(CBTS)) +sys.path.insert(0, str(CBTS / "coverage_selection")) + +from blocks import YAMLIndex, block_matches_stage, parse_stages_from_groovy # noqa: E402 +from touch_db import ( # noqa: E402 + _LAUNCH_MARKERS, + _MIN_FUNCS, + _SERVING_PATH_MARKERS, + _WORKER_SENTINEL, + TouchDB, + db_key, + split_stage, +) + +_DEFAULT_TEST_DB = CBTS.parents[2] / "tests/integration/test_lists/test-db" +_DEFAULT_GROOVY = CBTS.parents[2] / "jenkins/L0_Test.groovy" + + +def _fmt_pct(n: int, d: int) -> str: + return f"{100.0 * n / d:.0f}%" if d else "n/a" + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + ap.add_argument("--db", required=True, help="path to cbts_touchmap.sqlite") + ap.add_argument("--list-untrusted", action="store_true", help="print every untrusted test") + ap.add_argument( + "--min-funcs", type=int, default=_MIN_FUNCS, help=f"near-empty floor (default {_MIN_FUNCS})" + ) + ap.add_argument( + "--test-db", + default=str(_DEFAULT_TEST_DB), + help="test-db dir to diff against the DB (HEAD coverage gap); '' to skip", + ) + ap.add_argument( + "--groovy", default=str(_DEFAULT_GROOVY), help="Groovy file to parse stage defs from" + ) + ap.add_argument( + "--list-not-in-db", action="store_true", help="print every gap case (default: first 15)" + ) + args = ap.parse_args(argv) + + db = TouchDB.open(args.db) + known = db.known_tests() + stages = db.known_by_stage() + footprint = db.per_test_footprint() + + print(f"=== CBTS coverage DB audit: {args.db} ===\n") + + # -- Format -- + print("## Format") + if stages: + print(f" test field: stage-prefixed ({len(stages)} instrumented stage(s) derivable)") + elif known: + print( + " test field: BARE nodeid !! WARNING: no stage prefix -> per-stage narrowing impossible" + ) + sv = db.schema_version() + commit = db.collection_commit() + print(f" schema_version: {sv or 'MISSING (selector cannot hard-fail on format drift)'}") + print( + f" collection commit: {commit or 'MISSING (no staleness gating; zero-touch lever stays off)'}" + ) + + # -- Scale -- + print("\n## Scale") + print( + f" known tests: {len(known)} | meta: tests={db.meta('tests')} files={db.meta('files')} " + f"functions={db.meta('functions')}" + ) + fr, qr = db.meta("file_rate_pct"), db.meta("func_rate_pct") + if fr or qr: + print(f" coverage rate: files {fr}% functions {qr}%") + + # -- Per-stage -- + print("\n## Instrumented stages") + for stage in sorted(stages): + print(f" {stage}: {len(stages[stage])} known") + + # -- Completeness -- + untrusted = db.untrusted_tests( + _WORKER_SENTINEL, _LAUNCH_MARKERS, _SERVING_PATH_MARKERS, args.min_funcs + ) + + def reason(test: str) -> str: + if any(m in test for m in _SERVING_PATH_MARKERS): + return "disagg-path (servers uninstrumented)" + if footprint[test] < args.min_funcs: + return f"near-empty (<{args.min_funcs} funcs)" + return "worker-lost (drove inference, no py_executor)" + + print("\n## Coverage completeness") + if footprint: + print( + f" per-test footprint (functions entered): min={min(footprint.values())} " + f"max={max(footprint.values())} (few funcs => likely lost subprocess capture)" + ) + else: + print(" per-test footprint: none (no test != '' rows — no usable per-test coverage)") + trusted_fp = [footprint[t] for t in known if t not in untrusted] + untrusted_fp = [footprint[t] for t in untrusted] + if trusted_fp and untrusted_fp: + print( + f" footprint gap: untrusted max={max(untrusted_fp)} | trusted min={min(trusted_fp)}" + ) + print( + f" UNTRUSTED (incomplete capture): {len(untrusted)}/{len(known)} ({_fmt_pct(len(untrusted), len(known))})" + ) + by_reason: dict[str, int] = {} + by_stage: dict[str, int] = {} + for t in untrusted: + by_reason[reason(t)] = by_reason.get(reason(t), 0) + 1 + by_stage[split_stage(t)[0]] = by_stage.get(split_stage(t)[0], 0) + 1 + for r, n in sorted(by_reason.items(), key=lambda kv: -kv[1]): + print(f" - {r}: {n}") + print(f" by stage: {dict(sorted(by_stage.items()))}") + print( + f"\n TRUSTED skippable universe: {len(known) - len(untrusted)}/{len(known)} " + f"(only these may ever be skipped)" + ) + + if args.list_untrusted: + print("\n## Untrusted tests") + for t in sorted(untrusted): + print(f" [{footprint[t]:>5} funcs] {t}\n -> {reason(t)}") + + # -- HEAD coverage gap: cases on an instrumented stage with no DB row -- + if args.test_db and Path(args.test_db).is_dir() and Path(args.groovy).is_file(): + yaml_index = YAMLIndex.load(Path(args.test_db)) + all_stages = parse_stages_from_groovy(Path(args.groovy), include_post_merge=True) + bare_known = {split_stage(t)[1] for t in known} + per_stage: dict[str, set[str]] = {} + for name in sorted(set(all_stages) & set(stages)): + stage = all_stages[name] + missing = { + entry + for block in yaml_index.blocks + if block.yaml_stem == stage.yaml_stem and block_matches_stage(block, stage) + for entry in block.tests + if (k := db_key(entry)) is not None and k not in bare_known + } + if missing: + per_stage[name] = missing + gap = sorted(set().union(*per_stage.values())) if per_stage else [] + print("\n## HEAD coverage gap on instrumented stages") + print( + f" {len(gap)} unique case(s) render on an instrumented single-GPU stage but have " + f"NO DB row -> always must-run (new/renamed or never captured)" + ) + for name in sorted(per_stage, key=lambda n: -len(per_stage[n])): + print(f" {name}: {len(per_stage[name])} not-in-DB") + preview = gap if args.list_not_in_db else gap[:15] + if preview: + print(" cases:") + for t in preview: + print(f" - {t}") + if not args.list_not_in_db and len(gap) > 15: + print(f" ... (+{len(gap) - 15}; use --list-not-in-db)") + + return 0 + + +if __name__ == "__main__": + sys.exit(main())