scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

ci(lint): ratcheted type-discipline gate (mutable collections, casts, guards, kwargs, suppressions) (#30500)

Browse source 
* ci(lint): enforce type-discipline budget for casts and type guards

Add a ratcheted gate that blocks net-new typing.cast() usage and bans
TypeGuard/TypeIs outright, layered on the existing ruff-strict budget setup.

- ruff-strict.toml: ban cast/TypeGuard/TypeIs (typing + typing_extensions)
  via flake8-tidy-imports banned-api (TID251) for a coarse import-level freeze.
- ruff-strict-budget.json: bump TID251 baseline 2404 -> 2662 to absorb the
  ~258 pre-existing usages now matched by the new banned-api entries.
- scripts/check_type_discipline.py: AST checker adding LIT006 (cast call sites,
  suppress with `# cast-ok: <reason>`) and LIT007 (TypeGuard/TypeIs annotations,
  suppress with `# guard-ok: <reason>`) for per-call-site granularity.
- scripts/type_discipline_gate.py: baseline+slack gate with delta-vs-base,
  mirroring ruff_strict_gate.py.
- type-discipline-budget.json: LIT006 baseline 1013 (slack 10), LIT007 0/0.
- test-linting.yml: run the gate in CI against the PR base SHA.

* ci(lint): enforce suppression-reason budgets and guard budgets against loosening

- wire the **kwargs ban (LIT008) into the vendored type-discipline checker so it
  matches the budget that already referenced it
- freeze LIT003/LIT004 (noqa / type-ignore without codes or reason) and LIT005
  (*-ok suppression without a reason) at slack 0 so any net-new unexplained
  suppression trips the type-discipline gate
- add scripts/budget_ratchet_check.py and a separate, non-gating budget-ratchet CI
  job that turns red when any *-budget.json ceiling is raised, a rule is dropped,
  or a budget file is deleted

* ci(lint): ban mutable collections in annotations and all mutable construction

Expand LIT001 from coarse builtins at interfaces to any mutable collection
in any annotation (builtins, typing aliases, collections concretes, mutable
ABCs) across signatures, class attributes, locals, and globals. Add LIT009 to
flag mutable-collection construction (literals, comprehensions, constructors)
so the unannotated seed-then-mutate pattern is caught too. Enumerate any-ok in
LIT005 so its reason requirement holds even when only the stdlib checker runs.
Budget LIT001 (21452) and LIT009 (25222) with slack 10 to ratchet down.

* ci(lint): recommend pydantic at boundaries and add functional-refactor guidance

Drop the msgspec mention from the cast banned-api messages so the recommended
validation path matches the codebase's primary pattern (pydantic). Add a note to
CLAUDE.md that lint / type-discipline failures should be resolved by refactoring
to functional, immutable patterns rather than reaching for mutable structures or
`# mutable-ok`.

* style: make CLAUDE.md more concise

* chore: update CLAUDE.md guidelines

* ci(lint): renumber mutable construction LIT009 -> LIT002 next to LIT001

Group the mutable-collection family together: LIT001 (mutable collection in any
annotation) and the construction rule now sit adjacent at LIT001/LIT002. The
freed LIT009 slot is taken by the sibling Any gate (check_any_discipline.py,
#30379), which moves its Any-typed-value rule LIT002 -> LIT009 in lockstep so
the shared LIT namespace stays contiguous with no holes. Budget, gate docstring,
and the checker's own docstring/messages are updated to match.

* fix: numbering in CLAUDE.md

* test(lint): test type-discipline checker, scope LIT007 to return types

Add regression tests for check_type_discipline.py (every LIT rule, its
suppression, and the comment scanner) and for budget_ratchet_check.py.

Confine LIT007 to function return annotations, the only place TypeGuard/TypeIs
are valid, so a runtime name that merely reads those identifiers is no longer
flagged. Switch scan_comments to io.StringIO(source).readline, the standard
readline that returns '' at EOF, dropping the iter(...).__next__ idiom.

* fix(lint): best-effort worktree teardown so cleanup can't mask the real error

base_counts ran `git worktree remove` through the raising `_run` in its finally,
so a failed `git worktree add` (or a failure in the body) was masked by a second
SystemExit from the cleanup. Tear the worktree down best-effort, like the sibling
rmtree, so the original error propagates.

* fix(lint): ratchet fails loudly on an unresolvable base; drop dead checker state

Verify the merge-base ref resolves to a commit before trusting a missing-file
result from git show, so an invalid or empty BASE_SHA now turns the budget-ratchet
guard red instead of skipping every budget and passing vacuously

Also drop the unused Comments.by_line field and the phantom --changed-only usage
line from check_type_discipline's docstring, and cover the ref handling with tests

* fix(lint): degrade malformed source to LIT000 instead of crashing the checker

tokenize.generate_tokens raises IndentationError (a SyntaxError subclass) on a dedent
mismatch, which escaped scan_comments' tokenize.TokenError handler and crashed the whole
checker run, zeroing the gate for that invocation. Catch SyntaxError too so the file
falls through to ast.parse and is reported as LIT000, matching the checker's
graceful-degradation contract. Also add the trailing newline ruff-strict.toml lacked

* perf(lint): skip the base worktree scan when no rule is over its ceiling

cmd_check created a git worktree and re-scanned the base tree on every run, but a
rule can only breach when its head count is already over baseline + slack; when none
are, the base comparison cannot change the verdict. Short-circuit to OK in that case,
which is every green PR, roughly halving the gate's work. Extract over_ceiling and
cover it (and evaluate's drift-safety) with tests

* fix(lint): exempt .dict()/.list()/.set() method calls from LIT002

_construction_kind matched dict/list/set as constructors via func.attr too, flagging
common method calls like pydantic's model.dict() as mutable construction; 200 such
false positives existed in litellm. Recognize dict/list/set construction only when
unqualified while keeping the collections concretes (deque/defaultdict/...) matchable
as attributes, since those are rarely method names. Ratchet the LIT002 baseline down
25222 -> 25022 to reflect the removed false positives

* chore(lint): bump basedpyright ceilings to absorb staging base drift

The basedpyright gate added in #30379 is a total-count check against
basedpyright-code-budget.json and the linting workflow runs only on
pull_request, so pushes to litellm_internal_staging never re-baseline it.
Merging staging into this branch surfaced that drift: seven
reportAny/reportUnknown* rules sit 10-149 errors above their committed ceiling
even though this PR changes no files under litellm/, the only path basedpyright
scans (pyrightconfig include is litellm). The new baselines match the counts CI
measured on the merge commit, with the existing per-rule slack preserved

* fix(lint): ratchet guard watches every budget file, not just two

DEFAULT_BUDGETS only listed ruff-strict-budget.json and
type-discipline-budget.json, so mypy-code-budget.json and
basedpyright-code-budget.json were unguarded and their ceilings could rise with
no signal, which is exactly the failure mode this guard exists to prevent. The
gap became concrete when this PR bumped basedpyright-code-budget.json to absorb
staging drift. All four budgets are now watched, so the budget-ratchet job
surfaces that basedpyright bump for human review the same way it surfaces the
TID251 raise. A regression test pins that every *-budget.json on disk is in
DEFAULT_BUDGETS, failing loudly if a future budget escapes the ratchet

* fix: add a lot more slack

* fix(lint): restore LIT003 frozen slack to 0

The blanket slack bump set LIT003 (bare # noqa without codes or a reason) to a
slack of 50, which contradicts the documented zero-tolerance invariant: the gate
docstring and the PR description table both freeze LIT003/LIT004/LIT005 at slack
0 so any net-new unexplained suppression trips the gate. Slack 50 would let 50
new bare noqas through silently. The actual LIT003 count is 397, well under the
516 baseline, so restoring slack to 0 keeps the gate green while putting the
freeze back. LIT004/LIT005/LIT007 were already correct at 0

* fix(lint): restore documented slack 10 for the buffered LIT rules

The slack bump left LIT001/LIT002/LIT006/LIT008 at 2000/2500/100/100, 10-250x the
"/ 10" the PR description table and the gate docstring document. That buffer was
never needed: the gate already blames a rule only when its count exceeds the
ceiling and grew vs the merge-base, so the violations the staging merge added in
litellm/ sit in both head and base and are never charged to this PR. With slack
back at the documented 10 the gate stays green, and the ceiling is tight again
(LIT006 no longer waves through 99 net-new cast() calls). Baselines are
unchanged; only the slack returns to its documented value

* fix(lint): ratchet LIT003 baseline down to its actual count

The LIT003 baseline was 516 while the current bare-noqa count is 397, leaving
~119 units of headroom that undercut the documented zero-tolerance freeze: the
gate docstring claims any net-new bare noqa trips the gate, but with cap 516 a PR
could add over a hundred first. Drop the baseline to the measured 397 so the
freeze is exact (cap = 397 + slack 0), the same hard-zero-at-the-boundary shape
LIT005 and LIT007 already use and pass in CI. PR table row updated to 397 / 0

* fix: increase slack

* fix: increase slack

* docs(lint): align gate docstring with buffered LIT003/LIT004 slack

The budget now gives LIT003/LIT004 nonzero slack, so the gate's prose no
longer claims they are frozen at slack 0; LIT005 remains the reasonless-
suppression freeze and LIT007 the hard zero.
This commit is contained in:
Mateo Wang 2026-06-16 16:59:21 -07:00 • committed by GitHub
parent 5a62806fdc
commit be4fa702e7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 1290 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
.github/workflows/test-linting.yml vendored
View file

@ -77,6 +77,12 @@ jobs:
run: |
uv run --no-sync python scripts/ruff_strict_gate.py --base "$BASE_SHA"
- name: Check type-discipline budget (mutable collections / casts / type guards / kwargs / unexplained suppressions, delta vs base)
env:
BASE_SHA: ${{ github.event.pull_request.base.sha }}
run: |
uv run --no-sync python scripts/type_discipline_gate.py --base "$BASE_SHA"
- name: Print OpenAI version
run: |
uv run --no-sync python -c "import openai; print(f'OpenAI version: {openai.__version__}')"
@ -100,6 +106,33 @@ jobs:
run: |
uv run --no-sync python -c "from litellm import *" || (echo '🚨 import failed, this means you introduced unprotected imports! 🚨'; exit 1)
# Intentionally NON-GATING. This job turns red when a *-budget.json ceiling is
# raised (or a rule/budget is dropped) so a loosening is obvious in review, but it
# must be kept OUT of the branch-protection required-checks list so a justified
# bump can still be merged by a human who has seen and accepted the red.
budget-ratchet:
runs-on: ubuntu-latest
timeout-minutes: 5
permissions:
contents: read
steps:
- uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
fetch-depth: 0
persist-credentials: false
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: "3.12"
- name: Ratchet check (budgets may only decrease; non-gating)
env:
BASE_SHA: ${{ github.event.pull_request.base.sha }}
run: |
python scripts/budget_ratchet_check.py --base "$BASE_SHA"
any-discipline:
# Separate job: the first run cold-builds litellm's type cache (~2 min, ~3 GB),
# so keep it off the main lint job's time budget. Subsequent runs reuse the

2
CLAUDE.md
View file

@ -42,6 +42,8 @@ If you're trying to create a new function that relies on untyped stuff, instead
The Any-discipline gate (`make lint-any`, also a CI job) fails when a line you changed under `litellm/` holds a value typed `Any`, including the `X | Any`. Ideally `# any-ok: <reason>` is never used; treat it as a last resort for a genuine typed/untyped boundary that Pydantic truly can't model
If you get an LIT001 or LIT002 fail, refactor the code to follow functional programming best practices rather than introducing mutable data structures. For example, build values in one shot with comprehensions or generators wrapped in `tuple()` / `frozenset()` instead of seeding an empty `list`/`dict`/`set` and mutating it over time. Ideally `# mutable-ok` is never used; reach for it only as a genuine last resort when an immutable rewrite is truly impossible, and always pair it with a real reason
Ask to commit and push your work when you're done (or if you're confident that your code is good and works, just do it)
When you must use real LLM models to, for example, write e2e tests, write a QA runbook, etc., make sure to use the latest models (doesn't have to be smartest, can also be a modern small, fast one. No strong preference for smart vs fast here, just use something modern) as of the year and month of the current date. Do a web search as necessary to figure that out

82
ruff-strict-budget.json
View file

@ -1,27 +1,27 @@
{
"ANN001": {
"baseline": 2865,
"slack": 10
"slack": 50
},
"ANN002": {
"baseline": 64,
"slack": 3
"slack": 5
},
"ANN003": {
"baseline": 759,
"slack": 10
"slack": 30
},
"ANN201": {
"baseline": 1944,
"slack": 10
"slack": 50
},
"ANN202": {
"baseline": 858,
"slack": 10
"slack": 30
},
"ANN204": {
"baseline": 658,
"slack": 10
"slack": 20
},
"ANN205": {
"baseline": 117,
@ -33,7 +33,7 @@
},
"ANN401": {
"baseline": 1886,
"slack": 10
"slack": 50
},
"ASYNC230": {
"baseline": 11,
@ -45,15 +45,15 @@
},
"B006": {
"baseline": 180,
"slack": 3
"slack": 10
},
"B008": {
"baseline": 490,
"slack": 10
"slack": 15
},
"B009": {
"baseline": 79,
"slack": 10
"slack": 5
},
"B010": {
"baseline": 187,
@ -81,7 +81,7 @@
},
"BLE001": {
"baseline": 2854,
"slack": 10
"slack": 50
},
"C401": {
"baseline": 8,
@ -109,7 +109,7 @@
},
"C901": {
"baseline": 301,
"slack": 3
"slack": 15
},
"D419": {
"baseline": 6,
@ -125,7 +125,7 @@
},
"DTZ005": {
"baseline": 229,
"slack": 10
"slack": 15
},
"DTZ006": {
"baseline": 10,
@ -165,7 +165,7 @@
},
"I001": {
"baseline": 258,
"slack": 10
"slack": 15
},
"LOG015": {
"baseline": 5,
@ -189,11 +189,11 @@
},
"PERF403": {
"baseline": 69,
"slack": 10
"slack": 5
},
"PIE790": {
"baseline": 263,
"slack": 10
"slack": 15
},
"PIE800": {
"baseline": 1,
@ -233,7 +233,7 @@
},
"PLR0913": {
"baseline": 1813,
"slack": 3
"slack": 50
},
"PLR1704": {
"baseline": 3,
@ -245,7 +245,7 @@
},
"PLR1714": {
"baseline": 252,
"slack": 10
"slack": 15
},
"PLR1730": {
"baseline": 7,
@ -265,11 +265,11 @@
},
"PLW0602": {
"baseline": 215,
"slack": 10
"slack": 15
},
"PLW0603": {
"baseline": 183,
"slack": 3
"slack": 10
},
"PLW1508": {
"baseline": 188,
@ -301,15 +301,15 @@
},
"RET504": {
"baseline": 709,
"slack": 10
"slack": 20
},
"RUF010": {
"baseline": 844,
"slack": 10
"slack": 30
},
"RUF012": {
"baseline": 158,
"slack": 3
"slack": 10
},
"RUF015": {
"baseline": 8,
@ -321,7 +321,7 @@
},
"RUF022": {
"baseline": 80,
"slack": 10
"slack": 5
},
"RUF023": {
"baseline": 2,
@ -337,15 +337,15 @@
},
"RUF059": {
"baseline": 69,
"slack": 10
"slack": 5
},
"RUF100": {
"baseline": 465,
"slack": 10
"slack": 15
},
"S110": {
"baseline": 222,
"slack": 10
"slack": 15
},
"S112": {
"baseline": 21,
@ -353,11 +353,11 @@
},
"SIM101": {
"baseline": 58,
"slack": 10
"slack": 5
},
"SIM102": {
"baseline": 311,
"slack": 10
"slack": 15
},
"SIM103": {
"baseline": 119,
@ -412,20 +412,20 @@
"slack": 3
},
"TID251": {
"baseline": 2405,
"slack": 10
"baseline": 2664,
"slack": 50
},
"TRY002": {
"baseline": 528,
"slack": 10
"slack": 20
},
"TRY004": {
"baseline": 93,
"slack": 10
"slack": 5
},
"TRY201": {
"baseline": 409,
"slack": 10
"slack": 15
},
"TRY203": {
"baseline": 113,
@ -433,15 +433,15 @@
},
"TRY300": {
"baseline": 853,
"slack": 10
"slack": 30
},
"UP006": {
"baseline": 12941,
"slack": 10
"slack": 100
},
"UP007": {
"baseline": 2520,
"slack": 10
"slack": 50
},
"UP008": {
"baseline": 2,
@ -469,7 +469,7 @@
},
"UP032": {
"baseline": 609,
"slack": 10
"slack": 20
},
"UP034": {
"baseline": 1,
@ -477,7 +477,7 @@
},
"UP035": {
"baseline": 2250,
"slack": 10
"slack": 50
},
"UP036": {
"baseline": 1,
@ -485,10 +485,10 @@
},
"UP037": {
"baseline": 100,
"slack": 10
"slack": 5
},
"UP045": {
"baseline": 18417,
"slack": 10
"slack": 100
}
}

14
ruff-strict.toml
View file

@ -18,4 +18,16 @@ max-args = 5
"typing.Dict".msg = "Frozen dataclass / NamedTuple / ReadOnly TypedDict; create a Mapping alias with concrete value types if truly dynamic."
"typing.Set".msg = "frozenset[X] or AbstractSet[X]."
"typing.MutableSequence".msg = "Sequence[X]."
"typing.MutableMapping".msg = "See typing.Dict."
"typing.MutableMapping".msg = "See typing.Dict."
# Unchecked casts: cast() lies to the type checker with no runtime guarantee.
# Validate into a concrete frozen type at the boundary (pydantic) instead.
# Per-call-site coverage lives in check_type_discipline.py (LIT006); this freezes
# new cast imports. Suppress (with a reason) via `# noqa: TID251 # <reason>`.
"typing.cast".msg = "No unchecked casts: validate into a frozen dataclass/NamedTuple/ReadOnly TypedDict at the boundary (pydantic)."
"typing_extensions.cast".msg = "Same as typing.cast."
# Unverified narrowing predicates: the checker never validates the guard body, so a
# wrong guard silently corrupts types. Banned outright (there are none today).
"typing.TypeGuard".msg = "Unverified narrowing. Parse into a concrete type, or use isinstance for a runtime-checked narrowing."
"typing_extensions.TypeGuard".msg = "Same as typing.TypeGuard."
"typing.TypeIs".msg = "Unverified narrowing (the body is trusted). Parse into a concrete type instead."
"typing_extensions.TypeIs".msg = "Same as typing.TypeIs."

157
scripts/budget_ratchet_check.py Normal file
View file

@ -0,0 +1,157 @@
#!/usr/bin/env python3
"""Non-gating ratchet guard: budget ceilings may only fall, never rise.
Every `*-budget.json` file (ruff-strict, type-discipline, mypy-code, basedpyright-code) is a
one-way ratchet: each rule's ceiling is `baseline + slack`, and the whole point is
to drive that number DOWN over time. This check compares every budget file against
its own content at the merge-base with the target branch and fails (exits 1, red) if:
* a rule's ceiling went up,
* a rule was dropped from a budget (its ceiling effectively became infinite), or
* an entire budget file was deleted.
New rules and lowered/equal ceilings are fine.
This is deliberately NOT a gating check. It should turn the run red so that a
loosening is impossible to miss in review, but it must stay OUT of the
branch-protection required-checks list: a justified bump (e.g. banning a new API,
which mechanically raises a baseline) can then still be merged by a human who has
seen the red and accepted it.
Usage:
python scripts/budget_ratchet_check.py [--base REF] [budget.json ...]
Stdlib only.
"""
from __future__ import annotations
import argparse
import json
import subprocess
import sys
from pathlib import Path
from typing import NamedTuple
REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_BASE = "origin/litellm_internal_staging"
DEFAULT_BUDGETS: tuple[str, ...] = (
"ruff-strict-budget.json",
"type-discipline-budget.json",
"mypy-code-budget.json",
"basedpyright-code-budget.json",
)
class Regression(NamedTuple):
budget: str
rule: str
detail: str
def _run(cmd: list[str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(cmd, cwd=REPO_ROOT, capture_output=True, text=True)
def _merge_base(base: str) -> str:
"""The common ancestor of `base` and HEAD, so unrelated base drift is ignored."""
proc = _run(["git", "merge-base", base, "HEAD"])
return proc.stdout.strip() or base
def _load_head(rel: str) -> dict | None:
path = REPO_ROOT / rel
if not path.exists():
return None
return json.loads(path.read_text())
def _ref_is_commit(ref: str) -> bool:
return _run(["git", "rev-parse", "--verify", "--quiet", f"{ref}^{{commit}}"]).returncode == 0
def _load_base(rel: str, ref: str) -> dict | None:
"""Budget content at `ref`, or None when the file did not exist there.
`ref` is verified as a real commit by the caller, so a non-zero `git show` here means
the path was absent at that commit, not that the ref itself is unresolvable.
"""
proc = _run(["git", "show", f"{ref}:{rel}"])
if proc.returncode != 0:
return None
return json.loads(proc.stdout)
def _caps(budget: dict) -> dict[str, int]:
"""Map each rule to its ceiling (baseline + slack); skip malformed specs."""
caps: dict[str, int] = {}
for rule, spec in budget.items():
if isinstance(spec, dict):
caps[rule] = int(spec.get("baseline", 0)) + int(spec.get("slack", 0))
return caps
def regressions_for(rel: str, base: dict | None, head: dict | None) -> list[Regression]:
if base is None:
return [] # new budget file: nothing to ratchet against yet
if head is None:
return [Regression(rel, "*", "budget file was deleted (every ceiling removed)")]
base_caps = _caps(base)
head_caps = _caps(head)
out: list[Regression] = []
for rule, base_cap in sorted(base_caps.items()):
if rule not in head_caps:
out.append(Regression(rel, rule, f"rule dropped (ceiling {base_cap} -> removed)"))
elif head_caps[rule] > base_cap:
out.append(Regression(rel, rule, f"ceiling raised {base_cap} -> {head_caps[rule]}"))
return out
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base", default=DEFAULT_BASE)
parser.add_argument("budgets", nargs="*", help="budget files to check")
args = parser.parse_args()
budgets = args.budgets or list(DEFAULT_BUDGETS)
ref = _merge_base(args.base)
if not _ref_is_commit(ref):
print(
f"FAIL: base ref {ref!r} does not resolve to a commit, so the ratchet has nothing "
f"to compare against; refusing to pass vacuously (check the --base / BASE_SHA value)",
file=sys.stderr,
)
return 1
regressions: list[Regression] = []
checked: list[str] = []
for rel in budgets:
base = _load_base(rel, ref)
head = _load_head(rel)
if base is None and head is None:
continue
if base is None:
print(f"skip {rel}: new file (no base at {args.base} to ratchet against)")
continue
checked.append(rel)
regressions.extend(regressions_for(rel, base, head))
if regressions:
print(f"FAIL: budget ceiling(s) loosened vs base {args.base} (merge-base {ref[:12]}):")
for reg in regressions:
print(f" {reg.budget} {reg.rule}: {reg.detail}")
print(
"Budgets are one-way ratchets and may only go down or stay flat. This "
"check is non-gating: if the increase is justified (e.g. a newly banned "
"API), a human can merge over the red after acknowledging it."
)
return 1
suffix = f" ({', '.join(checked)})" if checked else ""
print(f"OK: no budget ceiling increased vs base {args.base}{suffix}")
return 0
if __name__ == "__main__":
raise SystemExit(main())

476
scripts/check_type_discipline.py Normal file
View file

@ -0,0 +1,476 @@
#!/usr/bin/env python3
"""Type-discipline checker: the rules ruff can't enforce.
Rules
-----
LIT001 Mutable collection in a type annotation, anywhere it appears: function
parameters, return types, class attributes, locals, and module globals.
Covers the builtins (dict/list/set, bare or parameterized), their typing
aliases (Dict/List/...), the collections concretes (deque/defaultdict/...),
and the mutable ABCs (MutableMapping/MutableSequence/MutableSet). A mutable
collection lets whoever holds it grow or rewrite it after the fact; annotate
a read-only view instead (Mapping/Sequence/AbstractSet/tuple[X, ...]/
frozenset[X], or a frozen dataclass / NamedTuple / ReadOnly TypedDict) and
build it functionally (comprehension / map, not append-in-a-loop).
Suppress with `# mutable-ok: <reason>` on the offending line.
LIT002 Mutable-collection *construction*: a list/dict/set literal or comprehension, or
a call to a mutable constructor (list/dict/set/deque/defaultdict/Counter/...).
Catches the unannotated seed-then-mutate pattern LIT001 cannot see (`acc = []`).
Build the value in one shot and freeze it: a `tuple`/`frozenset` wrapping a
generator (`tuple(f(x) for x in xs)`), a tuple literal, or a frozen dataclass /
NamedTuple / ReadOnly TypedDict. Generator expressions and `tuple`/`frozenset`
calls are not construction and pass. Annotation-internal lists (`Callable[[int],
str]`) are exempt. Suppress with `# mutable-ok: <reason>`.
LIT003 noqa suppression without rule codes or without a reason.
Required shape: `# noqa: TID251 # <reason>`
LIT004 type/pyright/mypy ignore without bracketed codes or without a reason.
Required shape: `# pyright: ignore[reportArgumentType] # <reason>`
LIT005 A `# mutable-ok` / `# cast-ok` / `# guard-ok` / `# kwargs-ok` / `# any-ok`
suppression without a reason. (`any-ok` belongs to check_any_discipline.py;
it is enumerated here so the reason requirement holds even when only this
stdlib checker runs.)
LIT006 `cast(...)` call. typing.cast is an unchecked assertion (the moral equivalent
of TypeScript's `as`); it lies to the type checker with zero runtime guarantee.
Validate into a concrete frozen type at the boundary instead.
Suppress with `# cast-ok: <reason>` on the call's first line.
LIT007 `TypeGuard[...]` / `TypeIs[...]` annotation. The narrowing predicate's body is
never verified by the checker, so a wrong guard silently corrupts types.
Prefer parsing into a concrete type. Suppress with `# guard-ok: <reason>`.
LIT008 `**kwargs` parameter. The keyword contract is erased and everything it carries
is effectively Any. ruff can force it to be typed (ANN003) but can't ban the
syntax. Declare explicit keyword params, or accept one frozen payload. `*args`,
by contrast, is fine when typed (it's just a tuple). Suppress: `# kwargs-ok: <reason>`.
LIT000 and LIT009 are the sibling Any gate's (check_any_discipline.py, #30379): a mypy
build/read failure and an Any-typed value. They share this LIT namespace but are emitted
by that checker, not this one.
Usage
-----
python check_type_discipline.py litellm/ tests/
Exit code 1 if any violation is found. Stdlib only.
"""
from __future__ import annotations
import ast
import io
import re
import sys
import tokenize
from dataclasses import dataclass
from pathlib import Path
from collections.abc import Iterable, Iterator, Sequence
from typing import NamedTuple
# Mutable collection types, banned in *every* annotation. Name-based, so `dict`,
# `typing.Dict`, `collections.deque`, and `collections.abc.MutableMapping` all match
# however they were imported. The read-only interfaces (Mapping, Sequence, the
# immutable AbstractSet / `abc.Set`, Collection) and the immutable concretes (tuple,
# frozenset) are the escape hatch and are deliberately absent -- as is the bare name
# `Set`, which collides with the read-only `collections.abc.Set`.
MUTABLE_COLLECTIONS = frozenset((
"dict", "list", "set",
"Dict", "List", "DefaultDict", "OrderedDict", "Counter", "Deque", "ChainMap",
"deque", "defaultdict",
"MutableMapping", "MutableSequence", "MutableSet",
))
# Callables whose result is a fresh *mutable* collection (LIT002). `tuple` and
# `frozenset` are deliberately absent -- they are the wrappers you reach for, and
# a generator expression fed to them is the blessed one-shot build.
MUTABLE_CONSTRUCTORS = frozenset((
"dict", "list", "set",
"deque", "defaultdict", "OrderedDict", "Counter", "ChainMap",
))
# A *qualified* call (`x.deque()`) counts as construction only for names that are rarely
# method names; `dict`/`list`/`set` are dropped here because `.dict()` / `.set()` / `.list()`
# are common methods (e.g. pydantic's `model.dict()`), not collection construction. A
# qualified `collections.deque(...)` still counts.
QUALIFIED_CONSTRUCTORS = MUTABLE_CONSTRUCTORS - frozenset(("dict", "list", "set"))
UNSAFE_GUARDS = frozenset(("TypeGuard", "TypeIs"))
MIN_REASON_LEN = 3
NOQA_RE = re.compile(
r"#\s*noqa"
r"(?P<colon>:\s*(?P<codes>[A-Z]+[0-9]+(?:\s*,\s*[A-Z]+[0-9]+)*))?"
r"(?P<rest>.*)",
re.IGNORECASE,
)
IGNORE_RE = re.compile(
r"#\s*(?:type|pyright|mypy):\s*ignore(?P<codes>\[[^\]]*\])?(?P<rest>.*)"
)
MUTABLE_OK_RE = re.compile(r"#\s*mutable-ok(?::\s*(?P<reason>.*))?")
CAST_OK_RE = re.compile(r"#\s*cast-ok(?::\s*(?P<reason>.*))?")
GUARD_OK_RE = re.compile(r"#\s*guard-ok(?::\s*(?P<reason>.*))?")
KWARGS_OK_RE = re.compile(r"#\s*kwargs-ok(?::\s*(?P<reason>.*))?")
ANY_OK_RE = re.compile(r"#\s*any-ok(?::\s*(?P<reason>.*))?")
# Suppression tokens that must each carry a reason (LIT005). `any-ok` is owned by
# check_any_discipline.py but listed here so the reason requirement is enforced even
# when only this stdlib checker runs.
OK_SUPPRESSIONS: tuple[tuple[str, re.Pattern[str]], ...] = (
("mutable-ok", MUTABLE_OK_RE),
("cast-ok", CAST_OK_RE),
("guard-ok", GUARD_OK_RE),
("kwargs-ok", KWARGS_OK_RE),
("any-ok", ANY_OK_RE),
)
class Violation(NamedTuple):
path: Path
line: int
code: str
message: str
def render(self) -> str:
return f"{self.path}:{self.line}: {self.code} {self.message}"
@dataclass(frozen=True, slots=True)
class Comments:
"""The lines carrying each valid `*-ok` suppression."""
mutable_ok_lines: frozenset[int]
cast_ok_lines: frozenset[int]
guard_ok_lines: frozenset[int]
kwargs_ok_lines: frozenset[int]
# --------------------------------------------------------------------------- #
# Comment scanning (LIT003 / LIT004 / LIT005)
# --------------------------------------------------------------------------- #
def _reason_of(rest: str) -> str:
return rest.strip().lstrip("#-").strip()
def _valid_ok(regex: re.Pattern[str], text: str) -> bool:
"""True iff `text` carries this suppression with a reason of usable length."""
m = regex.search(text)
return bool(m) and len((m.group("reason") or "").strip()) >= MIN_REASON_LEN
def _comment_violations(path: Path, line_no: int, text: str) -> Iterator[Violation]:
"""Pure: all LIT003/004/005 findings for one comment."""
for token, regex in OK_SUPPRESSIONS:
m = regex.search(text)
if m and len((m.group("reason") or "").strip()) < MIN_REASON_LEN:
yield Violation(path, line_no, "LIT005", f"{token} requires a reason: `# {token}: <reason>`")
m = NOQA_RE.search(text)
if m:
if not m.group("codes"):
yield Violation(path, line_no, "LIT003", "noqa requires rule codes: `# noqa: XXX123 # <reason>`")
elif len(_reason_of(m.group("rest"))) < MIN_REASON_LEN:
yield Violation(path, line_no, "LIT003", "noqa requires a reason: `# noqa: XXX123 # <reason>`")
m = IGNORE_RE.search(text)
if m:
codes = m.group("codes")
if not codes or codes == "[]":
yield Violation(path, line_no, "LIT004",
"ignore requires codes: `# pyright: ignore[ruleName] # <reason>`")
elif len(_reason_of(m.group("rest"))) < MIN_REASON_LEN:
yield Violation(path, line_no, "LIT004",
"ignore requires a reason: `# pyright: ignore[ruleName] # <reason>`")
def scan_comments(path: Path, source: str) -> tuple[Comments, tuple[Violation, ...]]:
try:
tokens = tokenize.generate_tokens(io.StringIO(source).readline)
comment_toks = tuple((t.start[0], t.string) for t in tokens if t.type == tokenize.COMMENT)
except (tokenize.TokenError, SyntaxError):
# tokenize raises TokenError (EOF mid-construct) or a SyntaxError subclass
# (IndentationError / TabError) on malformed source; defer to ast.parse below,
# which re-raises and is reported as LIT000 rather than crashing the run.
return Comments(frozenset(), frozenset(), frozenset(), frozenset()), ()
def _lines_with(regex: re.Pattern[str]) -> frozenset[int]:
return frozenset(line for line, text in comment_toks if _valid_ok(regex, text))
return (
Comments(
mutable_ok_lines=_lines_with(MUTABLE_OK_RE),
cast_ok_lines=_lines_with(CAST_OK_RE),
guard_ok_lines=_lines_with(GUARD_OK_RE),
kwargs_ok_lines=_lines_with(KWARGS_OK_RE),
),
tuple(v for line, text in comment_toks for v in _comment_violations(path, line, text)),
)
# --------------------------------------------------------------------------- #
def mutable_names_in(annotation: ast.expr) -> Iterator[str]:
"""Yield mutable-collection names anywhere inside an annotation expression.
Matches bare names (`dict`, `MutableMapping`) and dotted access (`typing.Dict`,
`collections.deque`, `collections.abc.MutableMapping`), descends through nesting
(`Mapping[str, list[int]]`, `tuple[set[int], ...]`) and string forward references.
"""
for node in ast.walk(annotation):
if isinstance(node, ast.Name) and node.id in MUTABLE_COLLECTIONS:
yield node.id
elif isinstance(node, ast.Attribute) and node.attr in MUTABLE_COLLECTIONS:
yield node.attr
elif isinstance(node, ast.Constant):
value: object = node.value # forward references arrive as string constants
if isinstance(value, str):
try:
inner = ast.parse(value, mode="eval").body
except SyntaxError:
continue
yield from mutable_names_in(inner)
def _mutable_ann(path: Path, line: int, name: str, where: str) -> Violation:
return Violation(
path, line, "LIT001",
f"mutable `{name}` in {where}: a mutable collection can be grown or rewritten "
f"by whoever holds it. Annotate a read-only view -- Mapping[...], Sequence[...], "
f"AbstractSet[...], tuple[X, ...], frozenset[X], or a frozen dataclass / "
f"NamedTuple / ReadOnly TypedDict -- and build it functionally, not by "
f"append-in-a-loop (suppress: `# mutable-ok: <reason>`)",
)
def _annotation_violations(
path: Path, annotation: ast.expr | None, line: int, where: str, ok_lines: frozenset[int]
) -> Iterator[Violation]:
if annotation is None or line in ok_lines:
return
yield from (_mutable_ann(path, line, name, where) for name in mutable_names_in(annotation))
def _function_violations(
path: Path, node: ast.FunctionDef | ast.AsyncFunctionDef, comments: Comments
) -> Iterator[Violation]:
mutable_ok = comments.mutable_ok_lines
args = node.args
for arg in (*args.posonlyargs, *args.args, *args.kwonlyargs):
yield from _annotation_violations(
path, arg.annotation, arg.lineno, f"parameter `{arg.arg}` of `{node.name}`", mutable_ok
)
# *args is allowed when typed (it's just a tuple); ruff ANN002 forces the
# annotation, so here we only add the LIT001 mutable-collection check on the element type.
if args.vararg is not None:
yield from _annotation_violations(
path, args.vararg.annotation, args.vararg.lineno, f"`*args` of `{node.name}`", mutable_ok
)
# **kwargs is banned outright (LIT008): it erases the keyword contract and forces
# Any-typing on everything it carries. ruff can require it be typed (ANN003) but
# cannot ban the syntax, so this rule does.
if args.kwarg is not None and args.kwarg.lineno not in comments.kwargs_ok_lines:
yield Violation(
path, args.kwarg.lineno, "LIT008",
f"`**{args.kwarg.arg}` is banned: it erases the keyword contract and forces "
f"Any-typing; declare explicit keyword parameters, or accept one frozen payload "
f"(frozen dataclass / NamedTuple / ReadOnly TypedDict) "
f"(suppress: `# kwargs-ok: <reason>`)",
)
if node.returns is not None:
yield from _annotation_violations(
path, node.returns, node.returns.lineno, f"return type of `{node.name}`", mutable_ok
)
def iter_annotation_violations(path: Path, tree: ast.AST, comments: Comments) -> Iterator[Violation]:
# Every annotation is in scope: signatures (params / *args / return) plus every
# `x: T` -- class attribute, local, or module global. The latter three are all
# ast.AnnAssign, so one walk covers them; only the signature annotations (which
# are not AnnAssign) need the dedicated helper.
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
yield from _function_violations(path, node, comments)
elif isinstance(node, ast.AnnAssign):
target = node.target.id if isinstance(node.target, ast.Name) else "<target>"
yield from _annotation_violations(
path, node.annotation, node.lineno,
f"the type of `{target}`", comments.mutable_ok_lines,
)
# --------------------------------------------------------------------------- #
# Unchecked casts (LIT006) and unverified narrowing predicates (LIT007)
# --------------------------------------------------------------------------- #
def _is_cast_call(node: ast.Call) -> bool:
"""`cast(...)` or `typing.cast(...)`, however the name was imported/aliased.
Name-based like MUTABLE_COLLECTIONS: a stray method called `.cast()` is a rare
false positive, suppressible with `# cast-ok: <reason>`.
"""
func = node.func
return (isinstance(func, ast.Name) and func.id == "cast") or (
isinstance(func, ast.Attribute) and func.attr == "cast"
)
def iter_cast_violations(path: Path, tree: ast.AST, comments: Comments) -> Iterator[Violation]:
for node in ast.walk(tree):
if isinstance(node, ast.Call) and _is_cast_call(node) and node.lineno not in comments.cast_ok_lines:
yield Violation(
path, node.lineno, "LIT006",
"cast() is an unchecked assertion (the type checker takes it on faith); "
"validate into a frozen dataclass/NamedTuple/ReadOnly TypedDict at the "
"boundary instead (suppress: `# cast-ok: <reason>`)",
)
def iter_guard_violations(path: Path, tree: ast.AST, comments: Comments) -> Iterator[Violation]:
# TypeGuard/TypeIs are legal only as a function's return annotation (`-> TypeGuard[int]`),
# so the walk is confined to `node.returns`; a runtime name that merely happens to read
# `TypeGuard` is not a narrowing predicate. ruff bans the import; this flags the use.
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) or node.returns is None:
continue
for sub in ast.walk(node.returns):
name = (
sub.id if isinstance(sub, ast.Name)
else sub.attr if isinstance(sub, ast.Attribute)
else None
)
if name in UNSAFE_GUARDS and sub.lineno not in comments.guard_ok_lines:
yield Violation(
path, sub.lineno, "LIT007",
f"`{name}` narrowing predicate: the checker never verifies the body, so a "
f"wrong guard silently corrupts types; parse into a concrete type instead "
f"(suppress: `# guard-ok: <reason>`)",
)
# --------------------------------------------------------------------------- #
# Mutable-collection construction (LIT002)
# --------------------------------------------------------------------------- #
def _annotations_of(node: ast.AST) -> tuple[ast.expr | None, ...]:
"""The annotation expressions a node carries (signatures and `x: T`)."""
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
a = node.args
params = (*a.posonlyargs, *a.args, *a.kwonlyargs, a.vararg, a.kwarg)
return (*(p.annotation for p in params if p is not None), node.returns)
if isinstance(node, ast.AnnAssign):
return (node.annotation,)
return ()
def _annotation_node_ids(tree: ast.AST) -> frozenset[int]:
"""ids() of every node living inside an annotation.
A list display inside an annotation (`Callable[[int], str]`) is type syntax,
not construction, so the LIT002 walk must skip those subtrees.
"""
return frozenset(
id(sub)
for node in ast.walk(tree)
for ann in _annotations_of(node)
if ann is not None
for sub in ast.walk(ann)
)
def _construction_kind(node: ast.expr) -> str | None:
"""Human label if `node` builds a mutable collection, else None."""
if isinstance(node, ast.List):
return "list literal"
if isinstance(node, ast.ListComp):
return "list comprehension"
if isinstance(node, ast.Set):
return "set literal"
if isinstance(node, ast.SetComp):
return "set comprehension"
if isinstance(node, ast.Dict):
return "dict literal"
if isinstance(node, ast.DictComp):
return "dict comprehension"
if isinstance(node, ast.Call):
func = node.func
if isinstance(func, ast.Name) and func.id in MUTABLE_CONSTRUCTORS:
return f"`{func.id}()` constructor"
if isinstance(func, ast.Attribute) and func.attr in QUALIFIED_CONSTRUCTORS:
return f"`{func.attr}()` constructor"
return None
def iter_construction_violations(path: Path, tree: ast.AST, comments: Comments) -> Iterator[Violation]:
in_annotation = _annotation_node_ids(tree)
for node in ast.walk(tree):
if not isinstance(node, ast.expr) or id(node) in in_annotation:
continue
kind = _construction_kind(node)
if kind is None or node.lineno in comments.mutable_ok_lines:
continue
yield Violation(
path, node.lineno, "LIT002",
f"mutable {kind}: this builds a collection that can be grown or rewritten. "
f"Build it in one shot and freeze it -- a tuple/frozenset wrapping a generator "
f"(`tuple(f(x) for x in xs)`), a tuple literal, or a frozen dataclass / NamedTuple "
f"/ ReadOnly TypedDict (suppress: `# mutable-ok: <reason>`)",
)
# --------------------------------------------------------------------------- #
# Driver
# --------------------------------------------------------------------------- #
def check_file(path: Path) -> tuple[Violation, ...]:
try:
source = path.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError) as exc:
return (Violation(path, 0, "LIT000", f"could not read file: {exc}"),)
comments, violations = scan_comments(path, source)
try:
tree = ast.parse(source, filename=str(path))
except SyntaxError as exc:
return (*violations, Violation(path, exc.lineno or 0, "LIT000", f"syntax error: {exc.msg}"))
return (
*violations,
*iter_annotation_violations(path, tree, comments),
*iter_cast_violations(path, tree, comments),
*iter_guard_violations(path, tree, comments),
*iter_construction_violations(path, tree, comments),
)
def collect_paths(raw: Iterable[str]) -> Iterator[Path]:
for item in raw:
p = Path(item)
if p.is_dir():
yield from sorted(p.rglob("*.py"))
elif p.suffix == ".py":
yield p
def main(argv: Sequence[str]) -> int:
paths = tuple(a for a in argv if not a.startswith("-"))
if not paths:
print("usage: check_type_discipline.py <files-or-dirs>...", file=sys.stderr)
return 2
violations = sorted(v for path in collect_paths(paths) for v in check_file(path))
for v in violations:
print(v.render())
if violations:
print(f"\n{len(violations)} violation(s).", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))

198
scripts/type_discipline_gate.py Normal file
View file

@ -0,0 +1,198 @@
#!/usr/bin/env python3
"""Total-count gate for the LIT* rules in scripts/check_type_discipline.py.
Sibling of scripts/ruff_strict_gate.py. Each rule listed in
type-discipline-budget.json has a hard ceiling (baseline + slack). The gate counts
each rule across the whole `litellm` tree and fails when a rule is both over its
ceiling and higher than the base it merges into, so a change is blamed for the
violations it adds, never for drift that already exists in the base.
Rules not present in the budget are ignored, but today every rule the checker
emits is gated: LIT001 (mutable collection in any annotation), LIT002
(mutable-collection construction), LIT003/LIT004 (noqa / ignore without codes or
reason), LIT006 (cast), and LIT008 (`**kwargs`) carry slack-buffered ceilings to
ratchet down; LIT005 (`*-ok` suppression without a reason) is frozen at slack 0
so any net-new reasonless suppression trips the gate; and LIT007 (TypeGuard/TypeIs)
is a hard zero. Re-baseline with `--update` to ratchet a ceiling down.
"""
import argparse
import json
import re
import shutil
import subprocess
import sys
import tempfile
from collections import Counter
from pathlib import Path
from typing import NamedTuple
REPO_ROOT = Path(__file__).resolve().parent.parent
CHECKER = REPO_ROOT / "scripts" / "check_type_discipline.py"
BUDGET_PATH = REPO_ROOT / "type-discipline-budget.json"
TARGET = "litellm"
DEFAULT_BASE = "origin/litellm_internal_staging"
_HUNK = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@")
_LINE = re.compile(r"^(?P<file>.+?):(?P<line>\d+): (?P<code>LIT\d+) ")
class Violation(NamedTuple):
file: str
line: int
code: str
class Breach(NamedTuple):
rule: str
total: int
cap: int
added: int
def _run(cmd: list, cwd: Path = REPO_ROOT) -> str:
proc = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True)
if proc.returncode not in (0, 1):
sys.stderr.write(proc.stderr)
raise SystemExit(f"{cmd[0]} exited {proc.returncode}")
return proc.stdout
def _check(root: Path, checker: Path) -> list:
# Resolve root first: on macOS tempfile dirs (/var/...) resolve to /private/var/...,
# and the checker prints already-resolved absolute paths, so relative_to would fail.
root = root.resolve()
out = _run([sys.executable, str(checker), str(root / TARGET)], cwd=root)
found = []
for line in out.splitlines():
m = _LINE.match(line)
if m is None:
continue
name = Path(m.group("file"))
full = name if name.is_absolute() else root / name
rel = full.resolve().relative_to(root).as_posix()
found.append(Violation(rel, int(m.group("line")), m.group("code")))
return found
def head_violations() -> list:
return _check(REPO_ROOT, CHECKER)
def count_by_rule(violations: list) -> dict:
return dict(Counter(v.code for v in violations))
def base_counts(ref: str) -> dict:
parent = Path(tempfile.mkdtemp(prefix="lit_base_"))
worktree = parent / "wt"
try:
_run(["git", "worktree", "add", "--detach", str(worktree), ref])
# Measure the base with the *current* rule logic, not whatever shipped at base.
(worktree / "scripts").mkdir(parents=True, exist_ok=True)
checker = worktree / "scripts" / "check_type_discipline.py"
shutil.copy(CHECKER, checker)
return count_by_rule(_check(worktree, checker))
finally:
# Best-effort teardown: cleanup must never raise, or it masks the real error when
# the body (or the `worktree add` itself) failed. rmtree is already best-effort.
subprocess.run(
["git", "worktree", "remove", "--force", str(worktree)],
cwd=REPO_ROOT, capture_output=True, text=True,
)
shutil.rmtree(parent, ignore_errors=True)
def over_ceiling(head: dict, budget: dict) -> frozenset:
"""Rules whose head count already exceeds baseline + slack.
A rule can only breach when it is over its ceiling, so when none are the base
comparison cannot change the verdict and the base worktree scan can be skipped.
"""
return frozenset(
rule for rule, spec in budget.items()
if head.get(rule, 0) > spec["baseline"] + spec["slack"]
)
def evaluate(head: dict, base: dict, budget: dict) -> list:
breaches = []
for rule, spec in budget.items():
cap = spec["baseline"] + spec["slack"]
total = head.get(rule, 0)
if total > cap and total > base.get(rule, 0):
breaches.append(Breach(rule, total, cap, total - base.get(rule, 0)))
return sorted(breaches)
def parse_changed_lines(diff_text: str) -> dict:
changed: dict = {}
path = None
for line in diff_text.splitlines():
if line.startswith("+++ b/"):
path = line[6:]
elif path and (match := _HUNK.match(line)):
start = int(match.group(1))
count = int(match.group(2)) if match.group(2) is not None else 1
changed.setdefault(path, set()).update(range(start, start + count))
return changed
def introduced(violations: list, changed: dict) -> list:
return [v for v in violations if v.line in changed.get(v.file, set())]
def cmd_check(base: str) -> None:
budget = json.loads(BUDGET_PATH.read_text())
head = head_violations()
head_counts = count_by_rule(head)
if not over_ceiling(head_counts, budget):
print(f"OK: every LIT rule is within its codebase ceiling (base {base})")
return
base_point = _run(["git", "merge-base", base, "HEAD"]).strip() or base
breaches = evaluate(head_counts, base_counts(base_point), budget)
if not breaches:
print(f"OK: every LIT rule is within its codebase ceiling (base {base})")
return
new = introduced(
head,
parse_changed_lines(
_run(["git", "diff", base_point, "--unified=0", "--no-color", "--", TARGET])
),
)
print(f"FAIL: LIT-rule totals exceed their ceiling (base {base}):")
for breach in breaches:
print(
f" {breach.rule}: total {breach.total} over cap {breach.cap} (this change added {breach.added})"
)
for violation in sorted(v for v in new if v.code == breach.rule):
print(f" {violation.file}:{violation.line}")
print(
"Remove the new violations, give each a reason (`# noqa: XXX # <reason>`, "
"`# pyright: ignore[rule] # <reason>`, `# mutable-ok: <reason>`, "
"`# cast-ok: <reason>`, `# guard-ok: <reason>`, `# kwargs-ok: <reason>`), or "
"remove an equal number elsewhere; the ceiling is baseline + slack in "
"type-discipline-budget.json."
)
raise SystemExit(1)
def cmd_update() -> None:
budget = json.loads(BUDGET_PATH.read_text())
head = count_by_rule(head_violations())
for rule in budget:
budget[rule]["baseline"] = head.get(rule, 0)
BUDGET_PATH.write_text(json.dumps(budget, indent=2, sort_keys=True) + "\n")
print("Re-captured per-rule baselines from the current tree")
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base", default=DEFAULT_BASE)
parser.add_argument("--update", action="store_true")
args = parser.parse_args()
cmd_update() if args.update else cmd_check(args.base)
if __name__ == "__main__":
main()

97
tests/test_litellm/test_budget_ratchet_check.py Normal file
View file

@ -0,0 +1,97 @@
"""Tests for scripts/budget_ratchet_check.py.
The guard's whole contract is "ceilings may only fall": a raised ceiling, a dropped
rule, or a deleted file is a regression, while a lowered/equal ceiling, a brand-new
rule, or a brand-new budget file is fine. Each branch is pinned here.
"""
import importlib.util
import subprocess
import sys
from pathlib import Path
_MODULE_PATH = Path(__file__).resolve().parents[2] / "scripts" / "budget_ratchet_check.py"
_spec = importlib.util.spec_from_file_location("budget_ratchet_check", _MODULE_PATH)
ratchet = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(ratchet)
def _spec_of(baseline, slack):
return {"baseline": baseline, "slack": slack}
def test_caps_sum_baseline_and_slack_and_skip_malformed():
caps = ratchet._caps({"LIT006": _spec_of(1013, 10), "junk": 5})
assert caps == {"LIT006": 1023} # malformed (non-dict) spec ignored
def test_raised_ceiling_is_a_regression():
base = {"LIT006": _spec_of(1013, 10)}
head = {"LIT006": _spec_of(1013, 11)} # cap 1023 -> 1024
regs = ratchet.regressions_for("b.json", base, head)
assert [r.rule for r in regs] == ["LIT006"]
assert "1023 -> 1024" in regs[0].detail
def test_lowered_or_equal_ceiling_is_clean():
base = {"LIT006": _spec_of(1013, 10)}
assert ratchet.regressions_for("b.json", base, {"LIT006": _spec_of(1000, 10)}) == []
assert ratchet.regressions_for("b.json", base, {"LIT006": _spec_of(1013, 10)}) == []
# slack traded for baseline at the same ceiling is fine
assert ratchet.regressions_for("b.json", base, {"LIT006": _spec_of(1023, 0)}) == []
def test_dropped_rule_is_a_regression():
regs = ratchet.regressions_for("b.json", {"LIT007": _spec_of(0, 0)}, {})
assert [r.rule for r in regs] == ["LIT007"]
assert "dropped" in regs[0].detail
def test_new_rule_in_head_is_clean():
assert ratchet.regressions_for("b.json", {}, {"LIT009": _spec_of(5, 0)}) == []
def test_deleted_budget_file_is_a_regression():
regs = ratchet.regressions_for("b.json", {"LIT006": _spec_of(1, 0)}, None)
assert [r.rule for r in regs] == ["*"]
assert "deleted" in regs[0].detail
def test_new_budget_file_has_nothing_to_ratchet():
assert ratchet.regressions_for("b.json", None, {"LIT006": _spec_of(1, 0)}) == []
def test_default_budgets_watch_every_budget_file_in_the_repo():
# This job is the repo's only ceiling-raise alarm, so every *-budget.json on disk must be
# watched; a budget left out of DEFAULT_BUDGETS (e.g. basedpyright-code-budget.json) can be
# loosened with no signal. Equality also catches a phantom entry that no longer exists.
repo_root = _MODULE_PATH.parents[1]
on_disk = frozenset(p.name for p in repo_root.glob("*budget*.json"))
assert on_disk == frozenset(ratchet.DEFAULT_BUDGETS)
# --------------------------------------------------------------------------- #
# Base-ref resolution: a bad ref must fail loudly, never pass vacuously
# --------------------------------------------------------------------------- #
def test_ref_is_commit_distinguishes_real_from_bogus():
assert ratchet._ref_is_commit("HEAD") is True
assert ratchet._ref_is_commit("definitely-not-a-real-ref-zzz") is False
def test_load_base_reads_a_present_file_and_none_for_an_absent_one():
# A real budget file exists at HEAD; a made-up path is absent at the same (valid) ref.
assert ratchet._load_base("type-discipline-budget.json", "HEAD") is not None
assert ratchet._load_base("scripts/no-such-budget-xyz.json", "HEAD") is None
def test_unresolvable_base_ref_exits_nonzero_instead_of_skipping():
proc = subprocess.run(
[sys.executable, str(_MODULE_PATH), "--base", "definitely-not-a-real-ref-zzz"],
cwd=_MODULE_PATH.parents[1],
capture_output=True,
text=True,
)
assert proc.returncode == 1
assert "does not resolve to a commit" in proc.stderr

199
tests/test_litellm/test_check_type_discipline.py Normal file
View file

@ -0,0 +1,199 @@
"""Tests for scripts/check_type_discipline.py.
Each rule is exercised on a snippet that violates it and on one that does not, so a
mutation that drops a rule, inverts a suppression, or breaks the comment scanner makes
a test fail. The comment-scanner cases are the regression for the readline path: if
`scan_comments` ever stops tokenizing comments, the LIT003/LIT005 assertions go red.
"""
import importlib.util
import json
import sys
from pathlib import Path
_REPO_ROOT = Path(__file__).resolve().parents[2]
_MODULE_PATH = _REPO_ROOT / "scripts" / "check_type_discipline.py"
_spec = importlib.util.spec_from_file_location("check_type_discipline", _MODULE_PATH)
checker = importlib.util.module_from_spec(_spec)
sys.modules[_spec.name] = checker # let the frozen dataclass resolve its own module
_spec.loader.exec_module(checker)
def _codes(tmp_path, source):
f = tmp_path / "snippet.py"
f.write_text(source, encoding="utf-8")
return [v.code for v in checker.check_file(f)]
# --------------------------------------------------------------------------- #
# Comment scanning (the readline path) — LIT003 / LIT004 / LIT005
# --------------------------------------------------------------------------- #
def test_scan_comments_tokenizes_every_comment():
# Direct regression for scan_comments: a bare noqa (LIT003) only surfaces if the comment
# was tokenized, and the valid cast-ok suppression line must be captured. A crash in the
# readline path would leave both empty.
source = "x = 1 # noqa\ny = 2 # cast-ok: validated upstream by the caller\n"
comments, violations = checker.scan_comments(Path("snippet.py"), source)
assert [v.code for v in violations] == ["LIT003"]
assert comments.cast_ok_lines == frozenset({2})
def test_scan_comments_does_not_crash_on_malformed_source():
# A dedent mismatch makes tokenize raise IndentationError (a SyntaxError subclass);
# scan_comments must swallow it, not propagate and crash the whole run.
comments, violations = checker.scan_comments(Path("x.py"), "if True:\n a = 1\n b = 2\n")
assert violations == ()
assert comments.cast_ok_lines == frozenset()
def test_malformed_source_degrades_to_lit000(tmp_path):
# The checker's contract is "bad file -> LIT000, never crash". An untokenizable file
# falls through scan_comments to ast.parse, which is reported as a single LIT000.
assert _codes(tmp_path, "if True:\n a = 1\n b = 2\n") == ["LIT000"]
def test_noqa_without_codes_is_flagged(tmp_path):
assert "LIT003" in _codes(tmp_path, "x = 1 # noqa\n")
def test_noqa_with_codes_and_reason_is_clean(tmp_path):
assert "LIT003" not in _codes(tmp_path, "x = 1 # noqa: TID251 # legacy import, removed in #123\n")
def test_ignore_without_reason_is_flagged(tmp_path):
assert "LIT004" in _codes(tmp_path, "x = 1 # type: ignore[arg-type]\n")
def test_ignore_with_codes_and_reason_is_clean(tmp_path):
assert "LIT004" not in _codes(tmp_path, "x = 1 # pyright: ignore[reportArgumentType] # upstream stub is wrong\n")
def test_ok_suppression_without_reason_is_flagged(tmp_path):
codes = _codes(tmp_path, "y = [] # mutable-ok\n")
assert "LIT005" in codes # reasonless suppression
assert "LIT002" in codes # and it does not suppress, so the construction still trips
# --------------------------------------------------------------------------- #
# Mutable annotations (LIT001) and construction (LIT002)
# --------------------------------------------------------------------------- #
def test_mutable_annotation_is_flagged(tmp_path):
assert "LIT001" in _codes(tmp_path, "x: dict[str, int]\n")
def test_typing_alias_and_forward_ref_annotations_are_flagged(tmp_path):
assert "LIT001" in _codes(tmp_path, "from typing import List\nx: List[int]\n")
assert "LIT001" in _codes(tmp_path, 'x: "dict[str, int]"\n')
def test_readonly_annotations_are_clean(tmp_path):
for ann in ("Mapping[str, int]", "Sequence[int]", "tuple[int, ...]", "frozenset[int]"):
assert "LIT001" not in _codes(tmp_path, f"from typing import Mapping, Sequence\nx: {ann}\n")
def test_mutable_construction_is_flagged(tmp_path):
assert "LIT002" in _codes(tmp_path, "y = []\n")
assert "LIT002" in _codes(tmp_path, "z = dict(a=1)\n")
def test_construction_inside_annotation_is_exempt(tmp_path):
# `Callable[[int], str]` carries a list display that is type syntax, not construction.
assert "LIT002" not in _codes(
tmp_path, "from typing import Callable\ndef f(cb: Callable[[int], str]) -> None:\n return None\n"
)
def test_generator_and_tuple_are_not_construction(tmp_path):
assert "LIT002" not in _codes(tmp_path, "g = tuple(i for i in range(3))\n")
assert "LIT002" not in _codes(tmp_path, "t = (1, 2, 3)\n")
def test_dict_list_set_method_calls_are_not_construction(tmp_path):
# `.dict()` / `.list()` / `.set()` are common method names (e.g. pydantic model.dict()),
# not collection construction; only the unqualified builtins count.
assert "LIT002" not in _codes(tmp_path, "d = model.dict()\n")
assert "LIT002" not in _codes(tmp_path, "s = obj.set()\n")
assert "LIT002" in _codes(tmp_path, "d = dict(a=1)\n") # unqualified still counts
def test_qualified_collections_constructors_still_count(tmp_path):
# collections concretes are rarely method names, so a qualified call still flags.
assert "LIT002" in _codes(tmp_path, "import collections\nq = collections.deque()\n")
assert "LIT002" in _codes(tmp_path, "import collections\nm = collections.defaultdict(list)\n")
def test_mutable_ok_with_reason_suppresses_both_rules(tmp_path):
codes = _codes(tmp_path, "x: dict[str, int] = {} # mutable-ok: in-place buffer mutated hot path\n")
assert "LIT001" not in codes
assert "LIT002" not in codes
# --------------------------------------------------------------------------- #
# Casts (LIT006)
# --------------------------------------------------------------------------- #
def test_cast_call_is_flagged(tmp_path):
assert "LIT006" in _codes(tmp_path, "from typing import cast\ny = cast(int, object())\n")
def test_cast_ok_with_reason_suppresses(tmp_path):
assert "LIT006" not in _codes(
tmp_path, "from typing import cast\ny = cast(int, object()) # cast-ok: validated by schema above\n"
)
# --------------------------------------------------------------------------- #
# Narrowing predicates (LIT007) — must fire only in return annotations
# --------------------------------------------------------------------------- #
def test_guard_in_return_annotation_is_flagged(tmp_path):
src = "from typing import TypeGuard\ndef is_int(v: object) -> TypeGuard[int]:\n return isinstance(v, int)\n"
assert "LIT007" in _codes(tmp_path, src)
def test_guard_name_outside_annotation_is_not_flagged(tmp_path):
# A runtime name or attribute that merely reads `TypeGuard`/`TypeIs` is not a predicate.
assert "LIT007" not in _codes(tmp_path, "TypeGuard = 1\nx = TypeGuard + 1\n")
assert "LIT007" not in _codes(tmp_path, "import obj\n_ = obj.TypeIs\n")
def test_guard_ok_with_reason_suppresses(tmp_path):
src = (
"from typing import TypeGuard\n"
"def is_int(v: object) -> TypeGuard[int]: # guard-ok: predicate proven by the assert below\n"
" assert isinstance(v, int)\n"
" return True\n"
)
assert "LIT007" not in _codes(tmp_path, src)
# --------------------------------------------------------------------------- #
# **kwargs (LIT008) — typed *args stays clean
# --------------------------------------------------------------------------- #
def test_kwargs_parameter_is_flagged(tmp_path):
assert "LIT008" in _codes(tmp_path, "def f(**kwargs) -> None:\n return None\n")
def test_typed_args_is_clean_but_kwargs_ok_suppresses(tmp_path):
assert "LIT008" not in _codes(tmp_path, "def f(*args: int) -> None:\n return None\n")
assert "LIT008" not in _codes(
tmp_path, "def f(**kwargs: int) -> None: # kwargs-ok: passthrough to a third-party sink\n return None\n"
)
# --------------------------------------------------------------------------- #
# Budget integrity: every emittable LIT rule (bar the LIT000 read/parse error) is gated
# --------------------------------------------------------------------------- #
def test_budget_covers_exactly_the_checker_rules():
budget = json.loads((_REPO_ROOT / "type-discipline-budget.json").read_text())
assert set(budget) == {f"LIT00{n}" for n in range(1, 9)}

40
tests/test_litellm/test_type_discipline_gate.py Normal file
View file

@ -0,0 +1,40 @@
"""Tests for scripts/type_discipline_gate.py.
The gate's correctness lives in two pure functions: `over_ceiling` (which decides
whether the expensive base worktree scan is even needed) and `evaluate` (the
drift-safe breach check). Both are pinned here.
"""
import importlib.util
from pathlib import Path
_MODULE_PATH = Path(__file__).resolve().parents[2] / "scripts" / "type_discipline_gate.py"
_spec = importlib.util.spec_from_file_location("type_discipline_gate", _MODULE_PATH)
gate = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(gate)
def _budget(baseline, slack):
return {"LIT006": {"baseline": baseline, "slack": slack}}
def test_over_ceiling_flags_only_counts_above_baseline_plus_slack():
budget = _budget(10, 2) # cap 12
assert gate.over_ceiling({"LIT006": 12}, budget) == frozenset() # at cap
assert gate.over_ceiling({"LIT006": 13}, budget) == frozenset({"LIT006"}) # over cap
assert gate.over_ceiling({}, budget) == frozenset() # missing rule counts as zero
def test_over_ceiling_is_independent_across_rules():
budget = {"LIT001": {"baseline": 5, "slack": 0}, "LIT006": {"baseline": 10, "slack": 0}}
assert gate.over_ceiling({"LIT001": 6, "LIT006": 10}, budget) == frozenset({"LIT001"})
def test_evaluate_blames_only_a_rule_over_cap_and_over_base():
budget = _budget(10, 0) # cap 10
# over cap and grown vs base -> breach
assert [b.rule for b in gate.evaluate({"LIT006": 12}, {"LIT006": 9}, budget)] == ["LIT006"]
# over cap but flat vs base (pre-existing drift) -> not blamed
assert gate.evaluate({"LIT006": 12}, {"LIT006": 12}, budget) == []
# within cap -> not blamed regardless of base
assert gate.evaluate({"LIT006": 10}, {"LIT006": 0}, budget) == []

34
type-discipline-budget.json Normal file
View file

@ -0,0 +1,34 @@
{
"LIT001": {
"baseline": 21452,
"slack": 2000
},
"LIT002": {
"baseline": 25022,
"slack": 2500
},
"LIT003": {
"baseline": 397,
"slack": 25
},
"LIT004": {
"baseline": 2515,
"slack": 50
},
"LIT005": {
"baseline": 0,
"slack": 0
},
"LIT006": {
"baseline": 1013,
"slack": 100
},
"LIT007": {
"baseline": 0,
"slack": 0
},
"LIT008": {
"baseline": 914,
"slack": 90
}
}