feat(tests): add vacuous test audit tooling and CI ratchet

.github/workflows/test-code-quality.yml

@@ -131,6 +131,9 @@ jobs:
       - name: memory_test
         run: uv run --no-sync python ./tests/code_coverage_tests/memory_test.py
 
+      - name: vacuous_test_ratchet
+        run: uv run --no-sync python ./tests/vacuous_tests/inventory.py --check
+
       - name: documentation_test_env_keys
         run: uv run --no-sync python ./tests/documentation_tests/test_env_keys.py
 

tests/vacuous_tests/README.md

@@ -0,0 +1,77 @@
+# Vacuous test audit
+
+A vacuous test is one that cannot fail when the code it claims to test is broken. It passes CI, adds runtime, and gives false confidence, which is worse than having no test at all
+
+This directory holds the tooling behind two things: a CI ratchet that stops new vacuous tests from landing, and a daily automation that fixes existing ones
+
+## Stage A: candidate inventory
+
+`inventory.py` walks `tests/` and classifies test functions into buckets. It is static, so it only produces *candidates*, never verdicts
+
+| bucket | meaning |
+| --- | --- |
+| `dead_skip` | unconditionally skipped, so it can never fail |
+| `swallowed_failure` | the assertion sits in a `try` whose handler swallows the failure |
+| `trivial_assert` | `assert True`, `assert <literal>`, or a value compared with itself |
+| `mock_tautology` | both sides of the comparison are values the test configured on a mock |
+| `no_assert` | no `assert`, `pytest.raises`/`fail`, or `assert_*` call anywhere in the body |
+
+`no_assert` is the noisiest bucket: a test that asserts by not raising is legitimate. That is what Stage B is for
+
+Commands:
+
+```bash
+python tests/vacuous_tests/inventory.py --report
+python tests/vacuous_tests/inventory.py --check            # CI ratchet
+python tests/vacuous_tests/inventory.py --update-baseline  # after a cleanup
+python tests/vacuous_tests/inventory.py --queue 15 --area tests/test_litellm/proxy
+```
+
+`inventory_baseline.json` records per-file candidate counts. `--check` fails when any count grows, so the number can only go down. If you are adding a deliberate assert-by-not-raising test, say so in the test's docstring and regenerate the baseline
+
+## Stage B: does the test actually have teeth
+
+`mutation_probe.py` decides. For one test it runs the test under coverage, subtracts the coverage floor of a no-op test in the same directory (so import-time lines are not counted), mutates only the lines the test itself executed, and re-runs the test against each mutant
+
+```bash
+python tests/vacuous_tests/mutation_probe.py "tests/x/test_y.py::test_z" --record
+```
+
+Mutants never touch the working tree: each one is written into a temp directory that symlinks the whole repo except the mutated file, and pytest runs with that directory as cwd, so a killed or crashed probe cannot leave mutated source behind
+
+Verdicts: `vacuous` (survived every mutant), `not_vacuous` (a mutant killed it, recorded in `verified_not_vacuous.json` so it is never re-flagged), `dead` (skipped in this environment), `already_failing`, or `inconclusive`
+
+Only a `vacuous` verdict authorizes editing a test
+
+## Fixing a vacuous test
+
+Refactor first: add the assertion the test's own name and docstring imply, then re-run the probe and confirm the mutant that used to survive now dies. Delete only when the behaviour is provably covered somewhere else, and cite that test id. When neither is possible, leave it alone and record why: a human should look at it
+
+## Not becoming flaky
+
+`flake_gate.py` runs every touched test five times under different hash seeds, runs the owning file as a whole, and statically rejects sleeps, wall-clock reads, unseeded randomness, and unmocked network calls
+
+```bash
+python tests/vacuous_tests/flake_gate.py "tests/x/test_y.py::test_z"
+```
+
+## Guardrails
+
+`guardrails.py` runs against the diff and rejects anything that games the metric: files outside `tests/`, edits to `conftest.py` or CI config, edits to this directory's own logic, test removals without a citation, and assertion counts dropping without tests being removed
+
+```bash
+python tests/vacuous_tests/guardrails.py --base origin/litellm_internal_staging
+python tests/vacuous_tests/guardrails.py --base origin/litellm_internal_staging --allow-removals removals.json
+```
+
+Each removal needs its own entry, keyed by the removed test id:
+
+```json
+{"tests/x/test_y.py::test_z": "tests/x/test_y.py::test_z_rejects_bad_input covers this"}
+```
+
+## Daily automation
+
+The scheduled run pulls the next batch from `--queue`, probes each candidate, fixes only the confirmed ones, clears the rest into `verified_not_vacuous.json`, runs the flake gate and the guardrails, then opens a single PR capped at 15 tests in one area. Anything it cannot fix honestly is reported rather than patched
+
+It stops rather than lowering the bar: if fewer than three candidates survive probing it opens no PR that day, and if three or more of its own PRs are still open it skips the run entirely

tests/vacuous_tests/flake_gate.py

@@ -0,0 +1,190 @@
+"""Anti-flake gate for tests the vacuous-test audit rewrote.
+
+A test that is vacuous today must not become flaky tomorrow, so every test the
+daily run touches has to clear this before its PR opens: repeated runs under
+different hash seeds, a full run of the owning file, and a static scan for the
+usual sources of nondeterminism.
+
+Usage:
+    python tests/vacuous_tests/flake_gate.py tests/x/test_y.py::test_z [more ...]
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import os
+import subprocess
+import sys
+from dataclasses import dataclass
+from typing import Dict, FrozenSet, List, Optional, Sequence, Tuple, Union
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+# (dotted call or attribute, why it is a flake risk)
+FLAKY_CALLS: Tuple[Tuple[str, str], ...] = (
+    ("time.sleep", "wall-clock sleep: slow and racy under load"),
+    ("asyncio.sleep", "wall-clock sleep: racy under load, prefer awaiting the real signal"),
+    ("datetime.now", "current time in a test makes it depend on when it runs"),
+    ("datetime.utcnow", "current time in a test makes it depend on when it runs"),
+    ("time.time", "current time in a test makes it depend on when it runs"),
+    ("random.random", "unseeded randomness"),
+    ("random.choice", "unseeded randomness"),
+    ("uuid.uuid4", "unseeded randomness in an assertion is unpredictable"),
+    ("requests.get", "real network call"),
+    ("requests.post", "real network call"),
+    ("httpx.get", "real network call"),
+    ("httpx.post", "real network call"),
+    ("litellm.completion", "hits a live provider unless mocked or replayed"),
+    ("litellm.acompletion", "hits a live provider unless mocked or replayed"),
+)
+MOCK_MARKERS = ("mock", "patch", "respx", "vcr", "cassette", "monkeypatch", "AsyncMock", "MagicMock")
+# Safe when the transport is patched for the call, unlike time and randomness,
+# which stay nondeterministic however the test is written.
+NETWORK_CALLS = frozenset(
+    {
+        "requests.get",
+        "requests.post",
+        "httpx.get",
+        "httpx.post",
+        "litellm.completion",
+        "litellm.acompletion",
+    }
+)
+
+
+@dataclass(frozen=True)
+class Finding:
+    test_id: str
+    problem: str
+
+
+def _env(seed: str) -> Dict[str, str]:
+    return {
+        **os.environ,
+        "LITELLM_LOCAL_MODEL_COST_MAP": "True",
+        "PYTHONHASHSEED": seed,
+        "PYTHONDONTWRITEBYTECODE": "1",
+    }
+
+
+def _run(target: str, seed: str, timeout: int) -> Tuple[int, str]:
+    command: Sequence[str] = (
+        sys.executable,
+        "-m",
+        "pytest",
+        target,
+        "-q",
+        "--no-header",
+        "-p",
+        "no:cacheprovider",
+        f"--timeout={timeout}",
+    )
+    completed = subprocess.run(command, cwd=REPO_ROOT, env=_env(seed), capture_output=True, text=True)
+    return completed.returncode, (completed.stdout + completed.stderr)[-2000:]
+
+
+TestFunction = Union[ast.FunctionDef, ast.AsyncFunctionDef]
+
+
+def _find_test(path: str, name: str) -> Optional[TestFunction]:
+    with open(os.path.join(REPO_ROOT, path), "r", encoding="utf-8") as handle:
+        tree = ast.parse(handle.read())
+    wanted = name.split("::")[-1]
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == wanted:
+            return node
+    return None
+
+
+def _dotted(node: ast.expr) -> str:
+    if isinstance(node, ast.Name):
+        return node.id
+    if isinstance(node, ast.Attribute):
+        prefix = _dotted(node.value)
+        return f"{prefix}.{node.attr}" if prefix else node.attr
+    return ""
+
+
+def _mocked(marker_source: str) -> bool:
+    return any(marker in marker_source for marker in MOCK_MARKERS)
+
+
+def mocked_lines(node: TestFunction) -> FrozenSet[int]:
+    """Lines where a patch is in force, either from a decorator or an enclosing with.
+
+    A mock anywhere in the test body is not enough: a test can patch one client
+    and still call a live provider two lines later.
+    """
+    if any(_mocked(ast.unparse(decorator)) for decorator in node.decorator_list):
+        return frozenset(range(node.lineno, (node.end_lineno or node.lineno) + 1))
+    return frozenset(
+        line
+        for statement in ast.walk(node)
+        if isinstance(statement, (ast.With, ast.AsyncWith))
+        and any(_mocked(ast.unparse(item.context_expr)) for item in statement.items)
+        for line in range(statement.lineno, (statement.end_lineno or statement.lineno) + 1)
+    )
+
+
+def static_findings(test_id: str) -> List[Finding]:
+    path, _, name = test_id.partition("::")
+    node = _find_test(path, name)
+    if node is None:
+        return [Finding(test_id, "test not found in file")]
+    patched = mocked_lines(node)
+    reasons: Dict[str, str] = dict(FLAKY_CALLS)
+    findings: List[Finding] = []
+    for call in ast.walk(node):
+        if not isinstance(call, ast.Call):
+            continue
+        dotted = _dotted(call.func)
+        matched = next((known for known in reasons if dotted == known or dotted.endswith(f".{known}")), None)
+        if matched is None or (matched in NETWORK_CALLS and call.lineno in patched):
+            continue
+        findings.append(Finding(test_id, f"uses `{matched}`: {reasons[matched]}"))
+    return findings
+
+
+def dynamic_findings(test_id: str, repeat: int, timeout: int) -> List[Finding]:
+    findings: List[Finding] = []
+    for index in range(repeat):
+        seed = str(index * 7919 + 1)
+        code, output = _run(test_id, seed, timeout)
+        if code != 0:
+            findings.append(
+                Finding(test_id, f"failed on repeat {index + 1}/{repeat} with PYTHONHASHSEED={seed}:\n{output[-600:]}")
+            )
+            break
+    path = test_id.partition("::")[0]
+    code, output = _run(path, "0", timeout)
+    if code != 0:
+        findings.append(Finding(test_id, f"owning file {path} does not pass as a whole:\n{output[-600:]}"))
+    return findings
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("test_ids", nargs="+")
+    parser.add_argument("--repeat", type=int, default=5)
+    parser.add_argument("--timeout", type=int, default=180)
+    parser.add_argument("--static-only", action="store_true")
+    args = parser.parse_args()
+
+    findings: List[Finding] = []
+    for test_id in args.test_ids:
+        findings.extend(static_findings(test_id))
+        if not args.static_only:
+            findings.extend(dynamic_findings(test_id, args.repeat, args.timeout))
+
+    if findings:
+        print("flake gate FAILED")
+        for finding in findings:
+            print(f"  - {finding.test_id}: {finding.problem}")
+        return 1
+    print(f"flake gate OK for {len(args.test_ids)} test(s)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/vacuous_tests/guardrails.py

@@ -0,0 +1,136 @@
+"""Guardrails on what the vacuous-test audit is allowed to change.
+
+The failure mode of an automated "make the vacuous-test count go down" loop is
+gaming the metric: deleting tests, weakening assertions, or editing production
+code until the suite agrees. This runs against a diff and rejects it unless the
+change is confined to test bodies.
+
+Rules:
+  1. Only files under tests/ may change, and never conftest.py, CI config, or
+     the audit tooling itself.
+  2. The number of test functions may not drop, unless --allow-removals is
+     passed with a citations file naming the test that now covers each removal.
+  3. Assertions may not be net removed from a test file.
+
+Usage:
+    python tests/vacuous_tests/guardrails.py --base origin/litellm_internal_staging
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import json
+import os
+import subprocess
+import sys
+from typing import Dict, FrozenSet, List, Optional
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+FORBIDDEN_NAMES = ("conftest.py", "pytest.ini", "pyproject.toml", "ruff.toml", "Makefile")
+FORBIDDEN_PREFIXES = (".github/",)
+# The audit's own logic is off limits; its JSON ledgers are exactly what the
+# daily run has to update.
+TOOLING_DIR = "tests/vacuous_tests/"
+
+
+def _git(*args: str) -> str:
+    return subprocess.run(("git", *args), cwd=REPO_ROOT, capture_output=True, text=True, check=True).stdout
+
+
+def changed_files(base: str) -> List[str]:
+    output = _git("diff", "--name-only", f"{base}...HEAD")
+    return [line for line in output.splitlines() if line.strip()]
+
+
+def _test_names(source: str) -> FrozenSet[str]:
+    try:
+        tree = ast.parse(source)
+    except SyntaxError:
+        return frozenset()
+    return frozenset(
+        node.name
+        for node in ast.walk(tree)
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("test_")
+    )
+
+
+def _assert_count(source: str) -> int:
+    try:
+        tree = ast.parse(source)
+    except SyntaxError:
+        return 0
+    return sum(1 for node in ast.walk(tree) if isinstance(node, ast.Assert))
+
+
+def _blob(ref: str, path: str) -> str:
+    try:
+        return _git("show", f"{ref}:{path}")
+    except subprocess.CalledProcessError:
+        return ""
+
+
+def check(base: str, allow_removals: Optional[str]) -> List[str]:
+    violations: List[str] = []
+    files = changed_files(base)
+    if not files:
+        return ["diff is empty"]
+
+    for path in files:
+        if not path.startswith("tests/"):
+            violations.append(f"{path}: outside tests/; this automation may only change tests")
+        if os.path.basename(path) in FORBIDDEN_NAMES or path.startswith(FORBIDDEN_PREFIXES):
+            violations.append(f"{path}: shared test or CI configuration is off limits")
+        if path.startswith(TOOLING_DIR) and path.endswith(".py"):
+            violations.append(f"{path}: the audit may not edit its own detection logic")
+
+    citations: Dict[str, str] = {}
+    if allow_removals:
+        with open(allow_removals, "r", encoding="utf-8") as handle:
+            citations = json.load(handle)
+
+    for path in files:
+        if not path.endswith(".py"):
+            continue
+        before_source = _blob(base, path)
+        after_source = _blob("HEAD", path)
+        before_names = _test_names(before_source)
+        after_names = _test_names(after_source)
+        violations.extend(uncited_removals(path, before_names - after_names, citations))
+        before_asserts = _assert_count(before_source)
+        after_asserts = _assert_count(after_source)
+        if after_asserts < before_asserts and len(after_names) >= len(before_names):
+            violations.append(
+                f"{path}: assertion count dropped from {before_asserts} to {after_asserts} "
+                "without removing tests, which looks like weakening"
+            )
+    return violations
+
+
+def uncited_removals(path: str, removed: FrozenSet[str], citations: Dict[str, str]) -> List[str]:
+    """Every removed test needs its own citation naming the test that now covers it."""
+    return [
+        f'{path}::{name} was removed without a citation; add "{path}::{name}": "<test id that now covers this>"'
+        for name in sorted(removed)
+        if not citations.get(f"{path}::{name}", "").strip()
+    ]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base", required=True, help="base ref to diff against")
+    parser.add_argument("--allow-removals", metavar="PATH", help="JSON map of removed test id -> covering test id")
+    args = parser.parse_args()
+
+    violations = check(args.base, args.allow_removals)
+    if violations:
+        print("vacuous-test guardrails FAILED")
+        for violation in violations:
+            print(f"  - {violation}")
+        return 1
+    print("vacuous-test guardrails OK")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/vacuous_tests/inventory.py

@@ -0,0 +1,542 @@
+"""Stage A of the vacuous-test audit: a static inventory of *candidate* vacuous tests.
+
+A vacuous test is one that cannot fail when the code it claims to test is
+broken. That property is not statically decidable, so this script only
+produces candidates; `mutation_probe.py` (Stage B) is what actually decides,
+by mutating the lines a single test covers and checking whether the test
+notices.
+
+Two jobs:
+
+1. Ratchet (CI). `--check` compares the per-file candidate counts against
+   `inventory_baseline.json` and fails when any count grows, so new vacuous
+   tests cannot land. Regenerate with `--update-baseline` after a cleanup.
+2. Queue (automation). `--queue N` prints the next N candidates for the daily
+   run, skipping anything Stage B has already cleared in
+   `verified_not_vacuous.json`.
+
+Usage:
+    python tests/vacuous_tests/inventory.py --report
+    python tests/vacuous_tests/inventory.py --check
+    python tests/vacuous_tests/inventory.py --update-baseline
+    python tests/vacuous_tests/inventory.py --queue 15 --area tests/litellm_utils_tests
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import json
+import os
+import sys
+import warnings
+from dataclasses import dataclass
+from typing import Dict, Iterable, List, Optional, Set, Tuple, Union
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+TOOL_DIR = os.path.join(REPO_ROOT, "tests", "vacuous_tests")
+BASELINE_PATH = os.path.join(TOOL_DIR, "inventory_baseline.json")
+CLEARED_PATH = os.path.join(TOOL_DIR, "verified_not_vacuous.json")
+TESTS_ROOT = os.path.join(REPO_ROOT, "tests")
+
+# Buckets, most to least specific. A test is reported under exactly one bucket
+# (the first that matches) so counts stay stable when a test has several
+# problems.
+BUCKETS = (
+    "dead_skip",
+    "swallowed_failure",
+    "trivial_assert",
+    "mock_tautology",
+    "no_assert",
+)
+
+# Directories whose tests are out of scope: they talk to live providers, take
+# minutes, or measure performance rather than behaviour, so "does a mutant kill
+# it" is either unaffordable or meaningless.
+EXCLUDED_DIRS = (
+    "tests/load_tests",
+    "tests/benchmarks",
+    "tests/e2e",
+    "tests/multi_instance_e2e_tests",
+    "tests/old_proxy_tests",
+    "tests/pass_through_tests",
+    "tests/vacuous_tests",
+)
+
+MOCK_FACTORIES = {"Mock", "MagicMock", "AsyncMock", "NonCallableMock", "patch"}
+ASSERT_CALL_PREFIXES = ("assert_", "assert", "check_", "verify_", "expect_")
+PYTEST_ASSERT_FUNCS = {"raises", "fail", "approx", "warns", "deprecated_call"}
+# Handler bodies made up only of these are swallowing the failure.
+SWALLOWING_CALLS = {"skip", "xfail", "print", "warn", "debug", "info", "warning"}
+# Attributes that only ever hold what the test itself configured or recorded on a mock.
+MOCK_CONFIG_ATTRS = ("return_value", "side_effect", "call_args", "await_args", "call_args_list")
+
+
+@dataclass(frozen=True)
+class Candidate:
+    path: str  # repo-relative
+    lineno: int
+    name: str  # dotted: Class.test_method or test_func
+    bucket: str
+    evidence: str
+
+    @property
+    def test_id(self) -> str:
+        return f"{self.path}::{'::'.join(self.name.split('.'))}"
+
+    def to_json(self) -> Dict[str, object]:
+        return {
+            "test_id": self.test_id,
+            "path": self.path,
+            "lineno": self.lineno,
+            "name": self.name,
+            "bucket": self.bucket,
+            "evidence": self.evidence,
+        }
+
+
+TestFunction = Union[ast.FunctionDef, ast.AsyncFunctionDef]
+
+
+def _decorator_names(node: TestFunction) -> List[str]:
+    names = []
+    for dec in node.decorator_list:
+        target = dec.func if isinstance(dec, ast.Call) else dec
+        names.append(ast.unparse(target))
+    return names
+
+
+def _decorators(node: TestFunction) -> List[ast.expr]:
+    return list(node.decorator_list)
+
+
+def _call_name(node: ast.AST) -> Optional[str]:
+    if not isinstance(node, ast.Call):
+        return None
+    func = node.func
+    if isinstance(func, ast.Name):
+        return func.id
+    if isinstance(func, ast.Attribute):
+        return func.attr
+    return None
+
+
+def _mock_expression(value: ast.expr, mock_names: Set[str]) -> bool:
+    """True for a mock factory call or an attribute chain rooted at a known mock.
+
+    Deliberately narrow: an expression that merely *passes* a mock into real code
+    (`result = handler(mock_client)`) is not mock-derived, because production code
+    still decides the outcome.
+    """
+    node = value.value if isinstance(value, ast.Await) else value
+    if isinstance(node, ast.Call):
+        return _call_name(node) in MOCK_FACTORIES
+    if isinstance(node, ast.Name):
+        return node.id in mock_names
+    while isinstance(node, ast.Attribute):
+        node = node.value
+    return isinstance(node, ast.Name) and node.id in mock_names
+
+
+def _mock_bound_names(fn: TestFunction) -> Set[str]:
+    """Names in `fn` bound to a Mock/patch result, transitively."""
+    mock_names: Set[str] = set()
+
+    def rhs_is_mock(value: ast.expr) -> bool:
+        return _mock_expression(value, mock_names)
+
+    # Two passes so `b = a.child` picks up a mock bound later in source order.
+    for _ in range(2):
+        for node in ast.walk(fn):
+            targets: List[ast.expr] = []
+            value: Optional[ast.expr] = None
+            if isinstance(node, ast.Assign):
+                targets, value = list(node.targets), node.value
+            elif isinstance(node, ast.AnnAssign) and node.value is not None:
+                targets, value = [node.target], node.value
+            elif isinstance(node, (ast.With, ast.AsyncWith)):
+                for item in node.items:
+                    if item.optional_vars is not None and rhs_is_mock(item.context_expr):
+                        targets.append(item.optional_vars)
+                        value = item.context_expr
+            if value is None or not rhs_is_mock(value):
+                continue
+            for target in targets:
+                for sub in ast.walk(target):
+                    if isinstance(sub, ast.Name):
+                        mock_names.add(sub.id)
+    # Fixture/decorator-injected mocks: `@patch(...)` passes the mock in as an
+    # argument, conventionally named mock_*.
+    for arg in fn.args.args:
+        if arg.arg.startswith("mock") or arg.arg.endswith("_mock"):
+            mock_names.add(arg.arg)
+    return mock_names
+
+
+def _local_names(node: ast.AST) -> Set[str]:
+    return {sub.id for sub in ast.walk(node) if isinstance(sub, ast.Name)}
+
+
+def _is_assertive_node(node: ast.AST) -> bool:
+    if isinstance(node, ast.Assert):
+        return True
+    name = _call_name(node)
+    if name is None:
+        return False
+    if name in PYTEST_ASSERT_FUNCS:
+        return True
+    if name.startswith(ASSERT_CALL_PREFIXES):
+        return True
+    return False
+
+
+def _has_assertion(fn: TestFunction) -> bool:
+    return any(_is_assertive_node(node) for node in ast.walk(fn))
+
+
+def _is_truthy_constant(node: ast.expr) -> bool:
+    if isinstance(node, ast.Constant):
+        return bool(node.value)
+    if isinstance(node, (ast.List, ast.Tuple, ast.Set)):
+        return bool(node.elts)
+    if isinstance(node, ast.Dict):
+        return bool(node.keys)
+    return False
+
+
+def _constant_locals(fn: TestFunction) -> Set[str]:
+    """Names assigned a truthy literal exactly once and never rebound."""
+    assigned: Dict[str, int] = {}
+    literal: Set[str] = set()
+    for node in ast.walk(fn):
+        if not isinstance(node, ast.Assign) or len(node.targets) != 1:
+            continue
+        target = node.targets[0]
+        if not isinstance(target, ast.Name):
+            continue
+        assigned[target.id] = assigned.get(target.id, 0) + 1
+        if _is_truthy_constant(node.value):
+            literal.add(target.id)
+    return {name for name in literal if assigned.get(name) == 1}
+
+
+def _unconditional_skip(fn: TestFunction, decorators: Iterable[ast.expr]) -> Optional[str]:
+    for dec in decorators:
+        target = dec.func if isinstance(dec, ast.Call) else dec
+        rendered = ast.unparse(target)
+        if rendered.endswith("mark.skip"):
+            return f"@{rendered}"
+        if rendered.endswith("mark.skipif") and isinstance(dec, ast.Call) and dec.args:
+            if _is_truthy_constant(dec.args[0]):
+                return f"@{rendered}(<always true>)"
+    body = [s for s in fn.body if not _is_docstring(s)]
+    if body and isinstance(body[0], ast.Expr):
+        name = _call_name(body[0].value)
+        if name in {"skip", "xfail"} and isinstance(body[0].value, ast.Call):
+            func = body[0].value.func
+            if isinstance(func, ast.Attribute) and ast.unparse(func).startswith("pytest."):
+                return f"unconditional pytest.{name}() as first statement"
+    return None
+
+
+def _is_docstring(stmt: ast.stmt) -> bool:
+    return isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Constant) and isinstance(stmt.value.value, str)
+
+
+def _handler_swallows(handler: ast.ExceptHandler) -> bool:
+    """True when the handler cannot surface the failure it caught."""
+    body = [s for s in handler.body if not _is_docstring(s)]
+    if not body:
+        return True
+    for stmt in body:
+        if isinstance(stmt, (ast.Pass, ast.Continue, ast.Break)):
+            continue
+        if isinstance(stmt, ast.Return) and stmt.value is None:
+            continue
+        if isinstance(stmt, ast.Expr):
+            name = _call_name(stmt.value)
+            if name in SWALLOWING_CALLS:
+                continue
+        return False
+    return True
+
+
+def _swallowed_assertion(fn: TestFunction) -> Optional[str]:
+    for node in ast.walk(fn):
+        if not isinstance(node, ast.Try):
+            continue
+        if not any(_is_assertive_node(sub) for stmt in node.body for sub in ast.walk(stmt)):
+            continue
+        for handler in node.handlers:
+            if handler.type is not None and "AssertionError" not in ast.unparse(handler.type):
+                # Only assertion-swallowing matters; `except KeyError: pass`
+                # around an assert is usually deliberate setup tolerance.
+                if not _catches_broad_exception(handler):
+                    continue
+            if _handler_swallows(handler):
+                return f"assert inside try/{ast.unparse(handler.type) if handler.type else 'except'} whose handler swallows the failure (line {handler.lineno})"
+    return None
+
+
+def _catches_broad_exception(handler: ast.ExceptHandler) -> bool:
+    if handler.type is None:
+        return True
+    rendered = ast.unparse(handler.type)
+    return "Exception" in rendered or "BaseException" in rendered
+
+
+def _trivial_assert(fn: TestFunction, constant_names: Set[str]) -> Optional[str]:
+    for node in ast.walk(fn):
+        if not isinstance(node, ast.Assert):
+            continue
+        test = node.test
+        if _is_truthy_constant(test):
+            return f"`assert {ast.unparse(test)}` (line {node.lineno})"
+        if isinstance(test, ast.Name) and test.id in constant_names:
+            return f"`assert {test.id}` where {test.id} is a literal (line {node.lineno})"
+        if isinstance(test, ast.Compare) and len(test.comparators) == 1:
+            if ast.unparse(test.left) == ast.unparse(test.comparators[0]):
+                return f"`assert {ast.unparse(test)}` compares a value with itself (line {node.lineno})"
+    return None
+
+
+def _mock_tautology(fn: TestFunction, mock_names: Set[str]) -> Optional[str]:
+    if not mock_names:
+        return None
+    for node in ast.walk(fn):
+        if not isinstance(node, ast.Assert):
+            continue
+        test = node.test
+        if not isinstance(test, ast.Compare) or len(test.comparators) != 1:
+            continue
+        sides = (test.left, test.comparators[0])
+        if not all(_mock_expression(side, mock_names) for side in sides):
+            continue
+        if not all(any(attr in ast.unparse(side) for attr in MOCK_CONFIG_ATTRS) for side in sides):
+            continue
+        if _local_names(test) - mock_names:
+            continue
+        return f"`assert {ast.unparse(test)}` compares two mock-derived values (line {node.lineno})"
+    return None
+
+
+def _module_level_skip(tree: ast.Module) -> Optional[str]:
+    for node in tree.body:
+        targets = node.targets if isinstance(node, ast.Assign) else []
+        if not any(isinstance(t, ast.Name) and t.id == "pytestmark" for t in targets):
+            continue
+        rendered = ast.unparse(node.value)
+        if "mark.skip" in rendered and "skipif" not in rendered:
+            return "module-level pytestmark skip"
+    return None
+
+
+def classify_file(path: str, source: str) -> List[Candidate]:
+    rel = os.path.relpath(path, REPO_ROOT).replace(os.sep, "/")
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", SyntaxWarning)
+            tree = ast.parse(source)
+    except SyntaxError:
+        return []
+    module_skip = _module_level_skip(tree)
+    candidates: List[Candidate] = []
+
+    def visit(node: Union[ast.Module, ast.ClassDef], prefix: str) -> None:
+        for child in node.body:
+            if isinstance(child, ast.ClassDef):
+                if child.name.startswith("Test"):
+                    visit(child, f"{prefix}{child.name}.")
+                continue
+            if not isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                continue
+            if not child.name.startswith("test_"):
+                continue
+            found = classify_test(child, module_skip)
+            if found is not None:
+                bucket, evidence = found
+                candidates.append(
+                    Candidate(
+                        path=rel,
+                        lineno=child.lineno,
+                        name=f"{prefix}{child.name}",
+                        bucket=bucket,
+                        evidence=evidence,
+                    )
+                )
+
+    visit(tree, "")
+    return candidates
+
+
+def classify_test(fn: TestFunction, module_skip: Optional[str]) -> Optional[Tuple[str, str]]:
+    decorators = _decorators(fn)
+    skip = module_skip or _unconditional_skip(fn, decorators)
+    if skip:
+        return "dead_skip", skip
+    swallowed = _swallowed_assertion(fn)
+    if swallowed:
+        return "swallowed_failure", swallowed
+    trivial = _trivial_assert(fn, _constant_locals(fn))
+    if trivial:
+        return "trivial_assert", trivial
+    mock_names = _mock_bound_names(fn)
+    tautology = _mock_tautology(fn, mock_names)
+    if tautology:
+        return "mock_tautology", tautology
+    if not _has_assertion(fn):
+        return "no_assert", "no assert, pytest.raises/fail, or assert_* call in the test body"
+    return None
+
+
+def iter_test_files(root: str = TESTS_ROOT) -> Iterable[str]:
+    for dirpath, dirnames, filenames in os.walk(root):
+        rel_dir = os.path.relpath(dirpath, REPO_ROOT).replace(os.sep, "/")
+        if any(rel_dir == ex or rel_dir.startswith(ex + "/") for ex in EXCLUDED_DIRS):
+            dirnames[:] = []
+            continue
+        dirnames[:] = sorted(d for d in dirnames if d not in {"__pycache__", ".pytest_cache"})
+        for filename in sorted(filenames):
+            if filename.endswith(".py") and (filename.startswith("test_") or filename.endswith("_test.py")):
+                yield os.path.join(dirpath, filename)
+
+
+def collect(root: str = TESTS_ROOT) -> List[Candidate]:
+    candidates: List[Candidate] = []
+    for path in iter_test_files(root):
+        with open(path, "r", encoding="utf-8", errors="ignore") as handle:
+            candidates.extend(classify_file(path, handle.read()))
+    return sorted(candidates, key=lambda c: (c.path, c.lineno))
+
+
+def to_counts(candidates: Iterable[Candidate]) -> Dict[str, Dict[str, int]]:
+    counts: Dict[str, Dict[str, int]] = {}
+    for candidate in candidates:
+        counts.setdefault(candidate.path, {})
+        counts[candidate.path][candidate.bucket] = counts[candidate.path].get(candidate.bucket, 0) + 1
+    return {path: dict(sorted(buckets.items())) for path, buckets in sorted(counts.items())}
+
+
+def load_json(path: str, default: object) -> object:
+    if not os.path.exists(path):
+        return default
+    with open(path, "r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def cleared_ids() -> Set[str]:
+    data = load_json(CLEARED_PATH, {})
+    if isinstance(data, dict):
+        return set(data.get("test_ids", []))
+    return set()
+
+
+def write_baseline(counts: Dict[str, Dict[str, int]]) -> None:
+    totals: Dict[str, int] = {}
+    for buckets in counts.values():
+        for bucket, count in buckets.items():
+            totals[bucket] = totals.get(bucket, 0) + count
+    payload = {
+        "_comment": (
+            "Ratchet baseline for tests/vacuous_tests/inventory.py. Counts may only "
+            "decrease; regenerate with --update-baseline after a cleanup."
+        ),
+        "totals": dict(sorted(totals.items())),
+        "files": counts,
+    }
+    with open(BASELINE_PATH, "w", encoding="utf-8") as handle:
+        json.dump(payload, handle, indent=2, sort_keys=False)
+        handle.write("\n")
+
+
+def regressions(counts: Dict[str, Dict[str, int]], baseline_files: Dict[str, Dict[str, int]]) -> List[str]:
+    return sorted(
+        f"{path}: {bucket} went from {baseline_files.get(path, {}).get(bucket, 0)} to {count}"
+        for path, buckets in counts.items()
+        for bucket, count in buckets.items()
+        if count > baseline_files.get(path, {}).get(bucket, 0)
+    )
+
+
+def check_against_baseline(counts: Dict[str, Dict[str, int]]) -> int:
+    baseline = load_json(BASELINE_PATH, None)
+    if baseline is None:
+        print(
+            f"ERROR: no baseline at {os.path.relpath(BASELINE_PATH, REPO_ROOT)}; run with --update-baseline",
+            file=sys.stderr,
+        )
+        return 1
+    base_files: Dict[str, Dict[str, int]] = baseline["files"]
+    failures = regressions(counts, base_files)
+    if failures:
+        print("Vacuous-test ratchet failed. New candidate vacuous tests:\n", file=sys.stderr)
+        for line in failures:
+            print(f"  - {line}", file=sys.stderr)
+        print(
+            "\nEach bucket is explained in tests/vacuous_tests/README.md. Make the new "
+            "test assert something a mutant can break; if this is a deliberate "
+            "assert-by-not-raising test, add a docstring saying so and regenerate the "
+            "baseline with:\n"
+            "  python tests/vacuous_tests/inventory.py --update-baseline",
+            file=sys.stderr,
+        )
+        return 1
+    improvements = 0
+    for path, buckets in base_files.items():
+        for bucket, count in buckets.items():
+            improvements += max(0, count - counts.get(path, {}).get(bucket, 0))
+    print(f"Vacuous-test ratchet OK ({improvements} candidate(s) below baseline).")
+    return 0
+
+
+def print_report(candidates: List[Candidate]) -> None:
+    totals: Dict[str, int] = {bucket: 0 for bucket in BUCKETS}
+    for candidate in candidates:
+        totals[candidate.bucket] += 1
+    print(f"candidate vacuous tests: {len(candidates)} across {len({c.path for c in candidates})} files")
+    for bucket in BUCKETS:
+        print(f"  {bucket:<20} {totals[bucket]}")
+
+
+def print_queue(candidates: List[Candidate], limit: int, area: Optional[str]) -> None:
+    cleared = cleared_ids()
+    queue = [c for c in candidates if c.test_id not in cleared and (area is None or c.path.startswith(area))]
+    # Most-specific buckets first: they are the highest-confidence candidates,
+    # so the daily run spends its mutation budget where it pays off.
+    order = {bucket: index for index, bucket in enumerate(BUCKETS)}
+    queue.sort(key=lambda c: (order[c.bucket], c.path, c.lineno))
+    print(json.dumps([c.to_json() for c in queue[:limit]], indent=2))
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--check", action="store_true", help="fail if counts grew over baseline")
+    parser.add_argument("--update-baseline", action="store_true")
+    parser.add_argument("--report", action="store_true", help="print per-bucket totals")
+    parser.add_argument("--json", metavar="PATH", help="write the full candidate list")
+    parser.add_argument("--queue", type=int, metavar="N", help="print the next N candidates")
+    parser.add_argument("--area", help="restrict --queue to a path prefix")
+    parser.add_argument("--root", default=TESTS_ROOT, help="tests root to scan")
+    args = parser.parse_args()
+
+    candidates = collect(args.root)
+    counts = to_counts(candidates)
+
+    if args.json:
+        with open(args.json, "w", encoding="utf-8") as handle:
+            json.dump([c.to_json() for c in candidates], handle, indent=2)
+            handle.write("\n")
+    if args.update_baseline:
+        write_baseline(counts)
+        print(f"wrote {os.path.relpath(BASELINE_PATH, REPO_ROOT)}")
+    if args.queue:
+        print_queue(candidates, args.queue, args.area)
+    if args.report or not (args.check or args.update_baseline or args.queue or args.json):
+        print_report(candidates)
+    if args.check:
+        return check_against_baseline(counts)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/vacuous_tests/mutation_probe.py

@@ -0,0 +1,526 @@
+"""Stage B of the vacuous-test audit: decide whether one test is actually vacuous.
+
+Runs a single test under coverage to find the production lines it executes,
+mutates only those lines, and re-runs the same test against each mutant. A test
+that survives every mutant cannot fail when the code it covers is broken, which
+is the working definition of vacuous. A single kill proves the test has teeth,
+and is recorded in verified_not_vacuous.json so the daily run stops re-flagging
+it.
+
+Unlike a whole-folder mutmut run (see .github/workflows/mutation-test.yml) this
+is scoped to one test and a handful of mutants, so it finishes in minutes.
+
+Usage:
+    python tests/vacuous_tests/mutation_probe.py \
+        "tests/test_litellm/foo/test_bar.py::test_baz" --json report.json
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import json
+import os
+import subprocess
+import sys
+import tempfile
+from contextlib import contextmanager
+from dataclasses import dataclass, field
+from typing import Callable, Dict, Iterable, Iterator, List, Optional, Sequence, Set, Tuple
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+TOOL_DIR = os.path.join(REPO_ROOT, "tests", "vacuous_tests")
+CLEARED_PATH = os.path.join(TOOL_DIR, "verified_not_vacuous.json")
+
+# Files that are configuration or generated data rather than behaviour: mutating
+# them says nothing about whether a test has teeth.
+SKIPPED_SOURCES = ("litellm/types/", "litellm/proxy/_types.py", "litellm/litellm_core_utils/model_param_helper.py")
+
+# Below this, "coverage" of a file is incidental (a lazy import, a decorator) rather
+# than the test exercising it.
+MIN_LINES_PER_FILE = 3
+
+COMPARE_SWAPS: Dict[type, type] = {
+    ast.Eq: ast.NotEq,
+    ast.NotEq: ast.Eq,
+    ast.Lt: ast.GtE,
+    ast.LtE: ast.Gt,
+    ast.Gt: ast.LtE,
+    ast.GtE: ast.Lt,
+    ast.Is: ast.IsNot,
+    ast.IsNot: ast.Is,
+    ast.In: ast.NotIn,
+    ast.NotIn: ast.In,
+}
+BINOP_SWAPS: Dict[type, type] = {
+    ast.Add: ast.Sub,
+    ast.Sub: ast.Add,
+    ast.Mult: ast.FloorDiv,
+    ast.Div: ast.Mult,
+}
+
+
+@dataclass(frozen=True)
+class Mutant:
+    path: str
+    lineno: int
+    description: str
+    source: str
+
+
+@dataclass
+class MutantResult:
+    path: str
+    lineno: int
+    description: str
+    outcome: str  # killed | survived | timeout | broken
+
+
+@dataclass
+class ProbeReport:
+    test_id: str
+    verdict: str
+    detail: str
+    covered_files: Dict[str, int] = field(default_factory=dict)
+    mutants: List[MutantResult] = field(default_factory=list)
+
+    def to_json(self) -> Dict[str, object]:
+        return {
+            "test_id": self.test_id,
+            "verdict": self.verdict,
+            "detail": self.detail,
+            "covered_files": self.covered_files,
+            "mutants": [vars(m) for m in self.mutants],
+        }
+
+
+def _pytest_env() -> Dict[str, str]:
+    return {
+        **os.environ,
+        "LITELLM_LOCAL_MODEL_COST_MAP": "True",
+        "PYTHONDONTWRITEBYTECODE": "1",
+    }
+
+
+def run_test(test_id: str, timeout: int, overlay: Optional[str] = None) -> Tuple[int, str]:
+    command: Sequence[str] = (
+        sys.executable,
+        "-m",
+        "pytest",
+        test_id,
+        "-q",
+        "--no-header",
+        "-p",
+        "no:cacheprovider",
+        f"--timeout={timeout}",
+    )
+    try:
+        completed = subprocess.run(
+            command,
+            cwd=overlay or REPO_ROOT,
+            env=_pytest_env(),
+            capture_output=True,
+            text=True,
+            timeout=timeout + 60,
+        )
+    except subprocess.TimeoutExpired:
+        return 124, "pytest wall-clock timeout"
+    return completed.returncode, (completed.stdout + completed.stderr)[-4000:]
+
+
+def _coverage_of(test_id: str, timeout: int) -> Dict[str, Set[int]]:
+    import coverage
+
+    with tempfile.TemporaryDirectory() as tmp:
+        data_file = os.path.join(tmp, ".coverage")
+        command: Sequence[str] = (
+            sys.executable,
+            "-m",
+            "coverage",
+            "run",
+            f"--data-file={data_file}",
+            "--source=litellm",
+            "-m",
+            "pytest",
+            test_id,
+            "-q",
+            "--no-header",
+            "-p",
+            "no:cacheprovider",
+            f"--timeout={timeout}",
+        )
+        subprocess.run(
+            command,
+            cwd=REPO_ROOT,
+            env=_pytest_env(),
+            capture_output=True,
+            text=True,
+            timeout=timeout + 120,
+        )
+        data = coverage.CoverageData(basename=data_file)
+        data.read()
+        result: Dict[str, Set[int]] = {}
+        for measured in data.measured_files():
+            rel = os.path.relpath(measured, REPO_ROOT).replace(os.sep, "/")
+            if not rel.startswith("litellm/") or rel.startswith(SKIPPED_SOURCES):
+                continue
+            lines = data.lines(measured) or []
+            if lines:
+                result[rel] = set(lines)
+        return result
+
+
+def covered_lines(test_id: str, timeout: int) -> Dict[str, List[int]]:
+    """Lines the test itself exercises, with import-time coverage subtracted.
+
+    Collecting any test in a directory imports litellm and that directory's
+    conftest, which lights up thousands of module-level lines. Those lines are
+    covered no matter what the test does, so mutating them measures the import,
+    not the test. A no-op test in the same directory gives the floor to subtract.
+    """
+    test_path = test_id.split("::", 1)[0]
+    with _noop_test(os.path.dirname(os.path.join(REPO_ROOT, test_path))) as noop_id:
+        floor = _coverage_of(noop_id, timeout)
+    actual = _coverage_of(test_id, timeout)
+    result: Dict[str, List[int]] = {}
+    for path, lines in actual.items():
+        own = sorted(lines - floor.get(path, set()))
+        if own:
+            result[path] = own
+    return result
+
+
+@contextmanager
+def _noop_test(directory: str) -> Iterator[str]:
+    handle = tempfile.NamedTemporaryFile(
+        mode="w", suffix=".py", prefix="test_vacuous_probe_floor_", dir=directory, delete=False
+    )
+    try:
+        handle.write("def test_noop() -> None:\n    assert True\n")
+        handle.close()
+        rel = os.path.relpath(handle.name, REPO_ROOT).replace(os.sep, "/")
+        yield f"{rel}::test_noop"
+    finally:
+        os.unlink(handle.name)
+
+
+NodeKey = Tuple[str, int, int]
+
+
+def _position(node: ast.AST) -> Tuple[int, int]:
+    if isinstance(node, (ast.expr, ast.stmt)):
+        return (node.lineno, node.col_offset)
+    return (-1, -1)
+
+
+def _key(node: ast.AST) -> NodeKey:
+    lineno, col_offset = _position(node)
+    return (type(node).__name__, lineno, col_offset)
+
+
+Mutation = Callable[[ast.AST], ast.AST]
+
+
+def _mutant_source(original: str, key: NodeKey, mutate: Mutation) -> str:
+    class Transformer(ast.NodeTransformer):
+        def visit(self, node: ast.AST) -> ast.AST:
+            if _key(node) == key:
+                return mutate(node)
+            return super().visit(node)
+
+    mutated = Transformer().visit(ast.parse(original))
+    ast.fix_missing_locations(mutated)
+    return ast.unparse(mutated)
+
+
+def function_body_lines(tree: ast.Module) -> Set[int]:
+    """Line numbers inside a function body.
+
+    Module and class level lines run at import, so a mutant there is killed (or
+    not) by whether the module still imports, which says nothing about the test.
+    """
+    lines: Set[int] = set()
+    for node in ast.walk(tree):
+        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            continue
+        for statement in node.body:
+            end = statement.end_lineno or statement.lineno
+            lines.update(range(statement.lineno, end + 1))
+    return lines
+
+
+def behavioural_lines(path: str, own_lines: Iterable[int]) -> List[int]:
+    with open(os.path.join(REPO_ROOT, path), "r", encoding="utf-8") as handle:
+        tree = ast.parse(handle.read())
+    inside = function_body_lines(tree)
+    return sorted(set(own_lines) & inside)
+
+
+def imported_modules(test_path: str) -> Set[str]:
+    with open(os.path.join(REPO_ROOT, test_path), "r", encoding="utf-8") as handle:
+        tree = ast.parse(handle.read())
+    modules: Set[str] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            modules.update(alias.name for alias in node.names)
+        elif isinstance(node, ast.ImportFrom) and node.module:
+            modules.add(node.module)
+            modules.update(f"{node.module}.{alias.name}" for alias in node.names)
+    return {module for module in modules if module.startswith("litellm")}
+
+
+def _module_name(path: str) -> str:
+    trimmed = path[: -len(".py")] if path.endswith(".py") else path
+    return trimmed.replace("/", ".").removesuffix(".__init__")
+
+
+def _is_under_test(path: str, imports: Set[str]) -> bool:
+    """True when the test file imports this module, directly or as a parent package."""
+    module = _module_name(path)
+    return any(imported == module or imported.startswith(module + ".") for imported in imports)
+
+
+def generate_mutants(path: str, lines: Iterable[int]) -> List[Mutant]:
+    absolute = os.path.join(REPO_ROOT, path)
+    with open(absolute, "r", encoding="utf-8") as handle:
+        original = handle.read()
+    tree = ast.parse(original)
+    covered = set(lines)
+    mutants: List[Mutant] = []
+
+    def emit(node: ast.AST, description: str, mutate: Mutation) -> None:
+        lineno, _ = _position(node)
+        if lineno not in covered:
+            return
+        try:
+            source = _mutant_source(original, _key(node), mutate)
+        except Exception:
+            return
+        mutants.append(Mutant(path=path, lineno=lineno, description=description, source=source))
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Compare) and len(node.ops) == 1 and type(node.ops[0]) in COMPARE_SWAPS:
+            swap = COMPARE_SWAPS[type(node.ops[0])]
+            emit(
+                node,
+                f"flip `{type(node.ops[0]).__name__}` to `{swap.__name__}` in `{_snippet(node)}`",
+                lambda n, swap=swap: ast.Compare(left=n.left, ops=[swap()], comparators=n.comparators),
+            )
+        elif isinstance(node, ast.BoolOp):
+            swap = ast.Or if isinstance(node.op, ast.And) else ast.And
+            emit(
+                node,
+                f"swap `{'and' if isinstance(node.op, ast.And) else 'or'}` in `{_snippet(node)}`",
+                lambda n, swap=swap: ast.BoolOp(op=swap(), values=n.values),
+            )
+        elif isinstance(node, ast.BinOp) and type(node.op) in BINOP_SWAPS:
+            swap = BINOP_SWAPS[type(node.op)]
+            emit(
+                node,
+                f"swap `{type(node.op).__name__}` for `{swap.__name__}` in `{_snippet(node)}`",
+                lambda n, swap=swap: ast.BinOp(left=n.left, op=swap(), right=n.right),
+            )
+        elif isinstance(node, ast.Constant) and isinstance(node.value, bool):
+            emit(
+                node,
+                f"replace `{node.value}` with `{not node.value}`",
+                lambda n: ast.Constant(value=not n.value),
+            )
+        elif isinstance(node, ast.Constant) and isinstance(node.value, int):
+            emit(
+                node,
+                f"replace `{node.value}` with `{node.value + 1}`",
+                lambda n: ast.Constant(value=n.value + 1),
+            )
+        elif isinstance(node, ast.Constant) and isinstance(node.value, str) and node.value:
+            emit(
+                node,
+                f"replace string `{node.value[:30]}` with a different value",
+                lambda n: ast.Constant(value=f"litellm_mutant_{n.value}"),
+            )
+        elif isinstance(node, ast.Return) and node.value is not None and not _is_none(node.value):
+            emit(
+                node,
+                f"replace `return {_snippet(node.value)}` with `return None`",
+                lambda n: ast.Return(value=ast.Constant(value=None)),
+            )
+    return mutants
+
+
+def _is_none(node: ast.expr) -> bool:
+    return isinstance(node, ast.Constant) and node.value is None
+
+
+def _snippet(node: ast.AST, limit: int = 60) -> str:
+    try:
+        rendered = ast.unparse(node)
+    except Exception:
+        return "<expr>"
+    return rendered if len(rendered) <= limit else rendered[: limit - 3] + "..."
+
+
+def select_mutants(mutants: Sequence[Mutant], limit: int) -> List[Mutant]:
+    """Round-robin over distinct lines so the budget spreads across the code path."""
+    by_line: Dict[Tuple[str, int], List[Mutant]] = {}
+    for mutant in mutants:
+        by_line.setdefault((mutant.path, mutant.lineno), []).append(mutant)
+    ordered: List[Mutant] = []
+    depth = 0
+    while len(ordered) < limit:
+        added = False
+        for key in sorted(by_line):
+            bucket = by_line[key]
+            if depth < len(bucket):
+                ordered.append(bucket[depth])
+                added = True
+                if len(ordered) == limit:
+                    break
+        if not added:
+            break
+        depth += 1
+    return ordered
+
+
+def probe(test_id: str, max_mutants: int, max_files: int, timeout: int) -> ProbeReport:
+    code, output = run_test(test_id, timeout)
+    if code == 5:
+        return ProbeReport(test_id, "inconclusive", "test not collected (renamed or parametrized away)")
+    if code != 0:
+        return ProbeReport(test_id, "already_failing", f"baseline run failed:\n{output[-800:]}")
+    if " skipped" in output and " passed" not in output:
+        return ProbeReport(test_id, "dead", "test is skipped in this environment, so it can never fail")
+
+    coverage_map = covered_lines(test_id, timeout)
+    behavioural = {
+        path: lines
+        for path, lines in ((path, behavioural_lines(path, own)) for path, own in coverage_map.items())
+        if len(lines) >= MIN_LINES_PER_FILE
+    }
+    if not behavioural:
+        return ProbeReport(
+            test_id,
+            "vacuous",
+            "test executes no litellm function body of its own; it only imports modules",
+        )
+
+    imports = imported_modules(test_id.partition("::")[0])
+    ranked = sorted(
+        behavioural.items(),
+        key=lambda item: (not _is_under_test(item[0], imports), -len(item[1]), item[0]),
+    )[:max_files]
+    report = ProbeReport(
+        test_id,
+        "inconclusive",
+        "",
+        covered_files={path: len(lines) for path, lines in ranked},
+    )
+    candidates: List[Mutant] = []
+    for path, lines in ranked:
+        candidates.extend(generate_mutants(path, lines))
+    if not candidates:
+        report.detail = "no mutable statements on the covered lines"
+        return report
+
+    kills = 0
+    for mutant in select_mutants(candidates, max_mutants):
+        outcome = _run_mutant(mutant, test_id, timeout)
+        report.mutants.append(MutantResult(mutant.path, mutant.lineno, mutant.description, outcome))
+        if outcome in {"killed", "timeout"}:
+            kills += 1
+            break
+
+    if kills:
+        report.verdict = "not_vacuous"
+        report.detail = f"killed by mutant: {report.mutants[-1].description} ({report.mutants[-1].path}:{report.mutants[-1].lineno})"
+        return report
+    tested = [m for m in report.mutants if m.outcome != "broken"]
+    if len(tested) < 3:
+        report.detail = f"only {len(tested)} usable mutant(s); not enough signal"
+        return report
+    report.verdict = "vacuous"
+    report.detail = f"survived all {len(tested)} mutants on the code it covers"
+    return report
+
+
+def _mirror(source: str, destination: str, remaining: Sequence[str]) -> None:
+    os.makedirs(destination, exist_ok=True)
+    head = remaining[0]
+    for entry in os.scandir(source):
+        if entry.name != head:
+            os.symlink(entry.path, os.path.join(destination, entry.name))
+    if len(remaining) > 1:
+        _mirror(os.path.join(source, head), os.path.join(destination, head), remaining[1:])
+
+
+@contextmanager
+def mutant_overlay(mutant: Mutant) -> Iterator[str]:
+    """A working directory where only the mutated file differs from the real tree.
+
+    Everything except the mutated file's own directory chain is symlinked, so the
+    real source is never written to: a crashed or killed probe cannot leave a
+    mutant behind in the repo. Running pytest with this as cwd puts it at the
+    front of sys.path, so the mutated module wins over the installed one.
+    """
+    with tempfile.TemporaryDirectory(prefix="vacuous_mutant_") as overlay:
+        parts = mutant.path.split("/")
+        _mirror(REPO_ROOT, overlay, parts)
+        with open(os.path.join(overlay, mutant.path), "w", encoding="utf-8") as handle:
+            handle.write(mutant.source)
+        yield overlay
+
+
+def _run_mutant(mutant: Mutant, test_id: str, timeout: int) -> str:
+    with mutant_overlay(mutant) as overlay:
+        code, output = run_test(test_id, timeout, overlay=overlay)
+    if code == 124:
+        return "timeout"
+    if code == 0:
+        return "survived"
+    if "ImportError" in output or "SyntaxError" in output or code == 5:
+        return "broken"
+    return "killed"
+
+
+def record_cleared(test_id: str) -> None:
+    payload = {"test_ids": []}
+    if os.path.exists(CLEARED_PATH):
+        with open(CLEARED_PATH, "r", encoding="utf-8") as handle:
+            payload = json.load(handle)
+    ids = sorted(set(payload.get("test_ids", [])) | {test_id})
+    payload["test_ids"] = ids
+    payload.setdefault(
+        "_comment",
+        "Tests a mutation probe proved have teeth. inventory.py --queue skips these.",
+    )
+    with open(CLEARED_PATH, "w", encoding="utf-8") as handle:
+        json.dump(payload, handle, indent=2)
+        handle.write("\n")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("test_id", help="pytest node id, e.g. tests/x/test_y.py::test_z")
+    parser.add_argument("--max-mutants", type=int, default=12)
+    parser.add_argument("--max-files", type=int, default=2)
+    parser.add_argument("--timeout", type=int, default=180, help="per-test-run timeout in seconds")
+    parser.add_argument("--json", metavar="PATH", help="write the report as JSON")
+    parser.add_argument("--record", action="store_true", help="record a not_vacuous verdict")
+    args = parser.parse_args()
+
+    report = probe(args.test_id, args.max_mutants, args.max_files, args.timeout)
+    if args.json:
+        with open(args.json, "w", encoding="utf-8") as handle:
+            json.dump(report.to_json(), handle, indent=2)
+            handle.write("\n")
+    print(f"{report.verdict}: {report.test_id}")
+    print(f"  {report.detail}")
+    for mutant in report.mutants:
+        print(f"  [{mutant.outcome}] {mutant.path}:{mutant.lineno} {mutant.description}")
+    if args.record and report.verdict == "not_vacuous":
+        record_cleared(report.test_id)
+        print(f"  recorded in {os.path.relpath(CLEARED_PATH, REPO_ROOT)}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/vacuous_tests/test_vacuous_tooling.py

@@ -0,0 +1,298 @@
+"""Tests for the vacuous-test audit tooling.
+
+Both directions are pinned: the classifier flags what it claims to flag, and it
+leaves healthy tests alone. A false positive here sends the daily automation to
+rewrite a working test
+"""
+
+from __future__ import annotations
+
+import ast
+import os
+import sys
+import textwrap
+from typing import List, Optional
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import flake_gate
+import guardrails
+import inventory
+import mutation_probe
+
+
+def findings_for(source: str, monkeypatch, tmp_path) -> List[str]:
+    path = tmp_path / "test_target.py"
+    path.write_text(textwrap.dedent(source))
+    monkeypatch.setattr(flake_gate, "REPO_ROOT", str(tmp_path))
+    return [finding.problem for finding in flake_gate.static_findings("test_target.py::test_thing")]
+
+
+def bucket_of(source: str, name: str = "test_thing") -> Optional[str]:
+    candidates = inventory.classify_file("/repo/tests/test_sample.py", textwrap.dedent(source))
+    for candidate in candidates:
+        if candidate.name.endswith(name):
+            return candidate.bucket
+    return None
+
+
+def test_flags_trivial_assert() -> None:
+    assert bucket_of("def test_thing():\n    do_work()\n    assert True\n") == "trivial_assert"
+
+
+def test_flags_self_comparison() -> None:
+    source = """
+    def test_thing():
+        value = compute()
+        assert value == value
+    """
+    assert bucket_of(source) == "trivial_assert"
+
+
+def test_flags_missing_assertion() -> None:
+    assert bucket_of("def test_thing():\n    result = compute()\n    print(result)\n") == "no_assert"
+
+
+def test_flags_swallowed_assertion() -> None:
+    source = """
+    def test_thing():
+        try:
+            assert compute() == 3
+        except Exception:
+            pass
+    """
+    assert bucket_of(source) == "swallowed_failure"
+
+
+def test_flags_unconditional_skip() -> None:
+    source = """
+    @pytest.mark.skip(reason="broken")
+    def test_thing():
+        assert compute() == 3
+    """
+    assert bucket_of(source) == "dead_skip"
+
+
+def test_flags_mock_only_comparison() -> None:
+    source = """
+    def test_thing():
+        client = MagicMock()
+        other = MagicMock()
+        assert client.send.return_value == other.send.return_value
+    """
+    assert bucket_of(source) == "mock_tautology"
+
+
+def test_ignores_real_assertion() -> None:
+    source = """
+    def test_thing():
+        assert compute() == 3
+    """
+    assert bucket_of(source) is None
+
+
+def test_ignores_pytest_raises_only_test() -> None:
+    source = """
+    def test_thing():
+        with pytest.raises(ValueError):
+            compute()
+    """
+    assert bucket_of(source) is None
+
+
+def test_ignores_assertion_in_shared_helper() -> None:
+    source = """
+    def test_thing():
+        assert_response_matches(compute(), expected)
+    """
+    assert bucket_of(source) is None
+
+
+def test_ignores_mock_passed_into_real_code() -> None:
+    source = """
+    def test_thing():
+        client = MagicMock()
+        result = handler(client)
+        assert result == client.send.return_value
+    """
+    assert bucket_of(source) is None
+
+
+def test_ignores_conditional_skip() -> None:
+    source = """
+    @pytest.mark.skipif(sys.platform == "win32", reason="posix only")
+    def test_thing():
+        assert compute() == 3
+    """
+    assert bucket_of(source) is None
+
+
+def test_classifies_methods_of_test_classes() -> None:
+    source = """
+    class TestThings:
+        def test_thing(self):
+            compute()
+    """
+    candidates = inventory.classify_file("/repo/tests/test_sample.py", textwrap.dedent(source))
+    assert [c.name for c in candidates] == ["TestThings.test_thing"]
+
+
+def test_ratchet_rejects_new_candidates() -> None:
+    failures = inventory.regressions(
+        {"tests/test_sample.py": {"no_assert": 2}},
+        {"tests/test_sample.py": {"no_assert": 1}},
+    )
+    assert failures == ["tests/test_sample.py: no_assert went from 1 to 2"]
+
+
+def test_ratchet_allows_fewer_candidates_and_untouched_files() -> None:
+    baseline = {"tests/test_sample.py": {"no_assert": 2}, "tests/test_other.py": {"dead_skip": 1}}
+    assert inventory.regressions({"tests/test_sample.py": {"no_assert": 1}}, baseline) == []
+
+
+def test_ratchet_rejects_candidates_in_a_new_file() -> None:
+    failures = inventory.regressions({"tests/test_new.py": {"trivial_assert": 1}}, {})
+    assert failures == ["tests/test_new.py: trivial_assert went from 0 to 1"]
+
+
+def _mutant_descriptions(source: str, lines: List[int], tmp_path: str) -> List[str]:
+    path = os.path.join(tmp_path, "module.py")
+    with open(path, "w", encoding="utf-8") as handle:
+        handle.write(textwrap.dedent(source))
+    relative = os.path.relpath(path, mutation_probe.REPO_ROOT)
+    return [mutant.description for mutant in mutation_probe.generate_mutants(relative, lines)]
+
+
+def test_mutates_only_covered_lines(tmp_path) -> None:
+    source = """
+    def covered(value):
+        return value > 3
+
+    def uncovered(value):
+        return value < 9
+    """
+    descriptions = _mutant_descriptions(source, [3], str(tmp_path))
+    assert any("value > 3" in description for description in descriptions)
+    assert not any("value < 9" in description for description in descriptions)
+
+
+def test_mutant_source_is_valid_python(tmp_path) -> None:
+    source = """
+    def covered(value):
+        if value == 3 and value is not None:
+            return "three"
+        return None
+    """
+    path = os.path.join(str(tmp_path), "module.py")
+    with open(path, "w", encoding="utf-8") as handle:
+        handle.write(textwrap.dedent(source))
+    relative = os.path.relpath(path, mutation_probe.REPO_ROOT)
+    mutants = mutation_probe.generate_mutants(relative, [3, 4, 5])
+    assert mutants
+    for mutant in mutants:
+        ast.parse(mutant.source)
+    assert any("flip `Eq`" in mutant.description for mutant in mutants)
+
+
+def test_import_time_lines_are_not_behavioural(tmp_path) -> None:
+    source = """
+    DEFAULT = True
+
+    class Config:
+        enabled = False
+
+        def check(self):
+            return self.enabled is True
+    """
+    path = os.path.join(str(tmp_path), "module.py")
+    with open(path, "w", encoding="utf-8") as handle:
+        handle.write(textwrap.dedent(source))
+    relative = os.path.relpath(path, mutation_probe.REPO_ROOT)
+    assert mutation_probe.behavioural_lines(relative, range(1, 9)) == [8]
+
+
+def test_overlay_isolates_the_mutant_from_the_real_tree(tmp_path, monkeypatch) -> None:
+    root = tmp_path / "root"
+    (root / "pkg" / "sub").mkdir(parents=True)
+    (root / "tests").mkdir()
+    target = root / "pkg" / "sub" / "mod.py"
+    target.write_text("X = 1\n")
+    (root / "pkg" / "other.py").write_text("Y = 1\n")
+    monkeypatch.setattr(mutation_probe, "REPO_ROOT", str(root))
+    mutant = mutation_probe.Mutant(path="pkg/sub/mod.py", lineno=1, description="d", source="X = 2\n")
+
+    with mutation_probe.mutant_overlay(mutant) as overlay:
+        assert open(os.path.join(overlay, "pkg", "sub", "mod.py"), encoding="utf-8").read() == "X = 2\n"
+        assert not os.path.islink(os.path.join(overlay, "pkg", "sub", "mod.py"))
+        assert os.path.islink(os.path.join(overlay, "tests"))
+        assert os.path.islink(os.path.join(overlay, "pkg", "other.py"))
+        assert target.read_text() == "X = 1\n"
+
+    assert not os.path.exists(overlay)
+    assert target.read_text() == "X = 1\n"
+
+
+def test_module_under_test_is_recognised_from_imports() -> None:
+    imports = {"litellm.llms.bedrock.base_aws_llm", "litellm.llms.bedrock.base_aws_llm.BaseAWSLLM"}
+    assert mutation_probe._is_under_test("litellm/llms/bedrock/base_aws_llm.py", imports)
+    assert not mutation_probe._is_under_test("litellm/caching/dual_cache.py", imports)
+
+
+def test_flake_gate_flags_sleep_and_wall_clock(monkeypatch, tmp_path) -> None:
+    source = """
+    def test_thing():
+        time.sleep(0.5)
+        started = datetime.datetime.now()
+        assert started
+    """
+    problems = findings_for(source, monkeypatch, tmp_path)
+    assert any("time.sleep" in problem for problem in problems)
+    assert any("datetime.now" in problem for problem in problems)
+
+
+def test_flake_gate_flags_live_call_outside_the_patch(monkeypatch, tmp_path) -> None:
+    source = """
+    def test_thing():
+        with patch("litellm.main.completion") as mocked:
+            litellm.completion(model="gpt-5", messages=[])
+        litellm.acompletion(model="gpt-5", messages=[])
+    """
+    problems = findings_for(source, monkeypatch, tmp_path)
+    assert problems == ["uses `litellm.acompletion`: hits a live provider unless mocked or replayed"]
+
+
+def test_flake_gate_accepts_patched_network_call(monkeypatch, tmp_path) -> None:
+    source = """
+    @patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post")
+    def test_thing(mocked_post):
+        response = litellm.completion(model="gpt-5", messages=[])
+        assert response.choices
+    """
+    assert findings_for(source, monkeypatch, tmp_path) == []
+
+
+def test_flake_gate_does_not_excuse_sleep_in_a_patched_test(monkeypatch, tmp_path) -> None:
+    source = """
+    @patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post")
+    def test_thing(mocked_post):
+        time.sleep(1)
+        assert mocked_post.called
+    """
+    assert findings_for(source, monkeypatch, tmp_path) == [
+        "uses `time.sleep`: wall-clock sleep: slow and racy under load"
+    ]
+
+
+def test_removals_need_a_citation_each() -> None:
+    removed = frozenset({"test_one", "test_two"})
+    citations = {"tests/test_sample.py::test_one": "tests/test_sample.py::test_covers_one"}
+    problems = guardrails.uncited_removals("tests/test_sample.py", removed, citations)
+    assert len(problems) == 1
+    assert "test_two was removed without a citation" in problems[0]
+
+
+def test_blank_citation_does_not_count() -> None:
+    problems = guardrails.uncited_removals(
+        "tests/test_sample.py", frozenset({"test_one"}), {"tests/test_sample.py::test_one": "  "}
+    )
+    assert len(problems) == 1

tests/vacuous_tests/verified_not_vacuous.json

@@ -0,0 +1,4 @@
+{
+  "_comment": "Tests a mutation probe proved have teeth. inventory.py --queue skips these.",
+  "test_ids": []
+}