ci: fail PRs that change the public litellm API without declaring it

Adds a griffe-based CI gate over the importable litellm SDK surface. Breaking
changes (removed objects, changed signatures, changed defaults) need a
Conventional Commits declaration, either a bang after the type or a
BREAKING CHANGE: footer. Newly exported top-level names need a feat: or fix:
title so a chore/refactor PR cannot quietly widen the public API.

Scope is deliberately narrower than raw griffe output: litellm.proxy.* is
excluded because the proxy's contract is HTTP rather than Python, and
re-exported stdlib names are excluded because their canonical home is another
package. On a 17-day sample window that took the finding count from 46 to 11,
with all 11 genuine.
This commit is contained in:
ryan-crabbe-berri 2026-08-07 13:36:31 -07:00
parent 48cb89dba7
commit 8f36f040bc
3 changed files with 574 additions and 0 deletions

View file

@ -0,0 +1,260 @@
#!/usr/bin/env python3
"""Gate PRs that change the importable `litellm` API without declaring it.
Scope is the pip package's SDK surface: what `import litellm` exposes. The
proxy's contract is HTTP, not Python, so `litellm.proxy.*` is out of scope,
as are re-exported stdlib names (`from typing import Union`) whose canonical
home is another package. Over a sample 17-day window that scoping took the
finding count from 46 to 11, and all 11 were real.
Two layers, both computed with griffe against the PR's base ref:
1. Breaking changes (removed objects, changed signatures, changed defaults).
Allowed only when the PR declares a breaking change the Conventional
Commits way: a `!` in the title type, or a `BREAKING CHANGE:` footer.
2. Newly exported top-level names (`litellm.<name>`). Allowed only under a
`feat` or `fix` title, so a `chore`/`refactor` PR cannot quietly widen
the public surface.
Attribute *value* changes are advisory: `Union[A, B] -> A | B` rewrites and
f-string conversions trip that check constantly without breaking anyone.
Usage:
check_api_breaking_changes.py --base-ref origin/main --pr-title "feat: x"
"""
from __future__ import annotations
import argparse
import os
import re
import sys
from collections.abc import Iterator, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Final, assert_never
if TYPE_CHECKING:
from griffe import Alias, Breakage, ExplanationStyle, Module, Object
DEFAULT_PACKAGE: Final = "litellm"
OUT_OF_SCOPE_PREFIXES: Final = ("litellm.proxy.",)
ADVISORY_KINDS: Final = frozenset({"ATTRIBUTE_CHANGED_VALUE"})
SURFACE_WIDENING_TYPES: Final = frozenset({"feat", "fix"})
CONVENTIONAL_TITLE_RE: Final = re.compile(r"^(?P<type>[a-z]+)(?:\([^)]*\))?(?P<bang>!)?:\s*\S")
BREAKING_FOOTER_RE: Final = re.compile(r"^BREAKING[ -]CHANGE:", re.MULTILINE)
ANSI_RE: Final = re.compile(r"\x1b\[[0-9;]*m")
@dataclass(frozen=True, slots=True)
class ApiFinding:
kind: str
path: str
detail: str
file: str | None
line: int | None
@dataclass(frozen=True, slots=True)
class ApiDelta:
blocking: tuple[ApiFinding, ...]
advisory: tuple[ApiFinding, ...]
added_names: tuple[str, ...]
@dataclass(frozen=True, slots=True)
class Declaration:
commit_type: str | None
breaking: bool
@dataclass(frozen=True, slots=True)
class Approved:
pass
@dataclass(frozen=True, slots=True)
class UndeclaredBreakingChanges:
findings: tuple[ApiFinding, ...]
@dataclass(frozen=True, slots=True)
class UndeclaredSurfaceWidening:
names: tuple[str, ...]
commit_type: str | None
Verdict = Approved | UndeclaredBreakingChanges | UndeclaredSurfaceWidening
def parse_declaration(pr_title: str, pr_body: str) -> Declaration:
match: Final = CONVENTIONAL_TITLE_RE.match(pr_title.strip())
bang: Final = bool(match and match.group("bang"))
footer: Final = bool(BREAKING_FOOTER_RE.search(pr_body))
return Declaration(
commit_type=match.group("type") if match else None,
breaking=bang or footer,
)
def decide(delta: ApiDelta, declaration: Declaration) -> Verdict:
if delta.blocking and not declaration.breaking:
return UndeclaredBreakingChanges(delta.blocking)
if delta.added_names and declaration.commit_type not in SURFACE_WIDENING_TYPES:
return UndeclaredSurfaceWidening(delta.added_names, declaration.commit_type)
return Approved()
def source_location(obj: Object | Alias) -> tuple[str | None, int | None]:
try:
return str(obj.filepath), obj.lineno
except Exception:
return None, None
def home_path(obj: Object | Alias) -> str:
try:
return obj.canonical_path
except Exception:
return str(getattr(obj, "target_path", obj.path))
def is_in_scope(obj: Object | Alias, package: str) -> bool:
if obj.path.startswith(OUT_OF_SCOPE_PREFIXES):
return False
return home_path(obj).startswith(f"{package}.")
def to_finding(breakage: Breakage, style: ExplanationStyle) -> ApiFinding:
file, line = source_location(breakage.obj)
return ApiFinding(
kind=breakage.kind.name,
path=breakage.obj.path,
detail=ANSI_RE.sub("", breakage.explain(style=style)).strip(),
file=file,
line=line,
)
def top_level_names(module: Module) -> frozenset[str]:
return frozenset(name for name in module.members if not name.startswith("_"))
def build_delta(
old: Module,
new: Module,
breakages: Iterator[Breakage],
style: ExplanationStyle,
package: str = DEFAULT_PACKAGE,
) -> ApiDelta:
findings: Final = tuple(
dict.fromkeys(to_finding(breakage, style) for breakage in breakages if is_in_scope(breakage.obj, package))
)
return ApiDelta(
blocking=tuple(f for f in findings if f.kind not in ADVISORY_KINDS),
advisory=tuple(f for f in findings if f.kind in ADVISORY_KINDS),
added_names=tuple(sorted(top_level_names(new) - top_level_names(old))),
)
def collect_delta(package: str, repo: Path, base_ref: str, head_ref: str | None) -> ApiDelta:
import griffe
old: Final = griffe.load_git(package, ref=base_ref, repo=repo, allow_inspection=False)
new: Final = (
griffe.load_git(package, ref=head_ref, repo=repo, allow_inspection=False)
if head_ref
else griffe.load(package, search_paths=[repo], allow_inspection=False)
)
return build_delta(old, new, griffe.find_breaking_changes(old, new), griffe.ExplanationStyle.ONE_LINE, package)
def render_annotations(findings: Sequence[ApiFinding]) -> str:
return "\n".join(
f"::error file={f.file},line={f.line}::{f.detail}" if f.file else f"::error::{f.detail}" for f in findings
)
def render_summary(delta: ApiDelta, verdict: Verdict) -> str:
header: Final = _verdict_header(verdict)
blocking: Final = "\n".join(f"- `{f.path}` {f.detail}" for f in delta.blocking)
added: Final = "\n".join(f"- `{name}`" for name in delta.added_names)
advisory: Final = "\n".join(f"- `{f.path}` {f.detail}" for f in delta.advisory)
sections: Final = (
("Breaking changes", blocking),
("New public names", added),
("Advisory (value changes, not gated)", advisory),
)
body: Final = "\n\n".join(f"### {title}\n{content}" for title, content in sections if content)
return f"## Public API check\n\n{header}\n\n{body}".rstrip() + "\n"
def _verdict_header(verdict: Verdict) -> str:
match verdict:
case Approved():
return "No undeclared public API changes."
case UndeclaredBreakingChanges(findings):
return (
f"{len(findings)} breaking change(s) to the public `litellm` API are not declared. "
"Add `!` after the type in the PR title (`feat!: ...`) or a `BREAKING CHANGE:` "
"footer in the PR body, and document the migration."
)
case UndeclaredSurfaceWidening(names, commit_type):
found: Final = f"`{commit_type}`" if commit_type else "an unparseable title"
return (
f"{len(names)} new public name(s) added under {found}. Widening the public API "
"needs a `feat:` or `fix:` PR title."
)
case _:
assert_never(verdict)
def _write(path_env: str, content: str) -> None:
destination: Final = os.environ.get(path_env)
if not destination:
return
with open(destination, "a", encoding="utf-8") as handle:
handle.write(content + "\n")
def _parse_args(argv: Sequence[str] | None) -> argparse.Namespace:
parser: Final = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--package", default=DEFAULT_PACKAGE)
parser.add_argument("--repo", default=".", type=Path)
parser.add_argument("--base-ref", required=True)
parser.add_argument("--head-ref", default=None)
parser.add_argument("--pr-title", default="")
parser.add_argument("--pr-body", default="")
return parser.parse_args(argv)
def main(argv: Sequence[str] | None = None) -> int:
args: Final = _parse_args(argv)
delta: Final = collect_delta(args.package, args.repo, args.base_ref, args.head_ref)
declaration: Final = parse_declaration(args.pr_title, args.pr_body)
verdict: Final = decide(delta, declaration)
summary: Final = render_summary(delta, verdict)
print(summary)
_write("GITHUB_STEP_SUMMARY", summary)
match verdict:
case Approved():
return 0
case UndeclaredBreakingChanges(findings):
print(render_annotations(findings))
return 1
case UndeclaredSurfaceWidening():
return 1
case _:
assert_never(verdict)
if __name__ == "__main__":
sys.exit(main())

View file

@ -0,0 +1,50 @@
name: Check Public API for Breaking Changes
on:
pull_request:
branches:
- main
- litellm_internal_staging
- litellm_oss_staging
- "litellm_**"
paths:
- "litellm/**/*.py"
- ".github/scripts/check_api_breaking_changes.py"
- ".github/workflows/check-api-breaking-changes.yml"
permissions:
contents: read
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
jobs:
check-public-api:
name: Diff the public API against the base branch
runs-on: ubuntu-latest
timeout-minutes: 15
steps:
- name: Checkout PR
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
fetch-depth: 0
persist-credentials: false
- name: Set up uv
uses: ./.github/actions/setup-uv-with-retries
with:
version: "0.10.9"
- name: Diff public API
env:
PR_TITLE: ${{ github.event.pull_request.title }}
PR_BODY: ${{ github.event.pull_request.body }}
BASE_SHA: ${{ github.event.pull_request.base.sha }}
run: |
uv run --no-project --with griffe==2.1.0 \
python .github/scripts/check_api_breaking_changes.py \
--repo . \
--base-ref "$BASE_SHA" \
--pr-title "$PR_TITLE" \
--pr-body "$PR_BODY"

View file

@ -0,0 +1,264 @@
"""Unit tests for the public-API gate (`.github/scripts/check_api_breaking_changes.py`).
The griffe-loading half needs a git repo, so these cover the decision half: what
counts as a declaration, what is in scope, and which combinations of findings and
PR metadata are allowed through.
"""
from __future__ import annotations
import importlib.util
import sys
from pathlib import Path
import pytest
SCRIPT_PATH = Path(__file__).resolve().parents[2] / ".github" / "scripts" / "check_api_breaking_changes.py"
def _load_module():
spec = importlib.util.spec_from_file_location("check_api_breaking_changes", SCRIPT_PATH)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
gate = _load_module()
def finding(kind: str = "OBJECT_REMOVED", path: str = "litellm.BedrockLLM"):
return gate.ApiFinding(kind=kind, path=path, detail=f"{path}: {kind}", file="litellm/x.py", line=1)
def delta(blocking=(), added=(), advisory=()):
return gate.ApiDelta(blocking=tuple(blocking), advisory=tuple(advisory), added_names=tuple(added))
class FakeObject:
"""Stands in for a griffe Object/Alias: `path` is where it is exported from,
`canonical` is where it actually lives (None mimics an unresolvable alias)."""
def __init__(self, path: str, canonical: str | None, target: str | None = None, lineno: int = 7):
self.path = path
self.filepath = Path("litellm/x.py")
self.lineno = lineno
self._canonical = canonical
self._target = target
@property
def canonical_path(self) -> str:
if self._canonical is None:
raise RuntimeError("alias does not resolve")
return self._canonical
@property
def target_path(self) -> str:
if self._target is None:
raise AttributeError("target_path")
return self._target
class TestParseDeclaration:
def test_bang_after_type_declares_breaking(self):
assert gate.parse_declaration("feat!: drop BedrockLLM", "").breaking is True
def test_bang_after_scope_declares_breaking(self):
declaration = gate.parse_declaration("refactor(proxy)!: rename hook", "")
assert declaration.breaking is True
assert declaration.commit_type == "refactor"
def test_plain_type_does_not_declare_breaking(self):
declaration = gate.parse_declaration("feat: add provider", "")
assert declaration.breaking is False
assert declaration.commit_type == "feat"
def test_footer_declares_breaking_without_bang(self):
body = "Some context\n\nBREAKING CHANGE: `BedrockLLM` is gone, use `BedrockConverse`\n"
assert gate.parse_declaration("fix: tidy", body).breaking is True
def test_hyphenated_footer_declares_breaking(self):
assert gate.parse_declaration("fix: tidy", "BREAKING-CHANGE: gone\n").breaking is True
def test_footer_must_start_a_line(self):
body = "We considered whether this is a BREAKING CHANGE: it is not.\n"
assert gate.parse_declaration("fix: tidy", body).breaking is False
def test_bang_in_subject_does_not_count(self):
assert gate.parse_declaration("fix: this is urgent!: really", "").breaking is False
def test_unparseable_title_has_no_type(self):
declaration = gate.parse_declaration("Drop BedrockLLM", "")
assert declaration.commit_type is None
assert declaration.breaking is False
class TestDecide:
def test_clean_delta_is_approved(self):
verdict = gate.decide(delta(), gate.parse_declaration("chore: tidy", ""))
assert isinstance(verdict, gate.Approved)
def test_undeclared_breaking_change_is_rejected(self):
verdict = gate.decide(delta(blocking=[finding()]), gate.parse_declaration("fix: tidy", ""))
assert isinstance(verdict, gate.UndeclaredBreakingChanges)
assert verdict.findings[0].path == "litellm.BedrockLLM"
def test_declared_breaking_change_is_approved(self):
verdict = gate.decide(delta(blocking=[finding()]), gate.parse_declaration("feat!: drop it", ""))
assert isinstance(verdict, gate.Approved)
def test_footer_alone_clears_a_breaking_change(self):
declaration = gate.parse_declaration("fix: tidy", "BREAKING CHANGE: gone\n")
assert isinstance(gate.decide(delta(blocking=[finding()]), declaration), gate.Approved)
def test_advisory_findings_never_block(self):
only_advisory = delta(advisory=[finding(kind="ATTRIBUTE_CHANGED_VALUE")])
verdict = gate.decide(only_advisory, gate.parse_declaration("chore: tidy", ""))
assert isinstance(verdict, gate.Approved)
@pytest.mark.parametrize("commit_type", ["feat", "fix"])
def test_new_names_allowed_under_feature_types(self, commit_type: str):
verdict = gate.decide(delta(added=["new_flag"]), gate.parse_declaration(f"{commit_type}: x", ""))
assert isinstance(verdict, gate.Approved)
@pytest.mark.parametrize("commit_type", ["chore", "refactor", "docs", "test", "ci"])
def test_new_names_rejected_under_non_feature_types(self, commit_type: str):
verdict = gate.decide(delta(added=["new_flag"]), gate.parse_declaration(f"{commit_type}: x", ""))
assert isinstance(verdict, gate.UndeclaredSurfaceWidening)
assert verdict.names == ("new_flag",)
assert verdict.commit_type == commit_type
def test_new_names_rejected_when_title_is_unparseable(self):
verdict = gate.decide(delta(added=["new_flag"]), gate.parse_declaration("Add a flag", ""))
assert isinstance(verdict, gate.UndeclaredSurfaceWidening)
assert verdict.commit_type is None
def test_breaking_change_is_reported_before_surface_widening(self):
both = delta(blocking=[finding()], added=["new_flag"])
verdict = gate.decide(both, gate.parse_declaration("chore: tidy", ""))
assert isinstance(verdict, gate.UndeclaredBreakingChanges)
def test_breaking_bang_does_not_excuse_surface_widening_under_chore(self):
both = delta(blocking=[finding()], added=["new_flag"])
verdict = gate.decide(both, gate.parse_declaration("chore!: tidy", ""))
assert isinstance(verdict, gate.UndeclaredSurfaceWidening)
class TestScope:
def test_sdk_object_is_in_scope(self):
obj = FakeObject("litellm.BedrockLLM", "litellm.llms.bedrock.chat.handler.BedrockLLM")
assert gate.is_in_scope(obj, "litellm") is True
def test_proxy_internals_are_out_of_scope(self):
obj = FakeObject(
"litellm.proxy.hooks.parallel_request_limiter_v3.TPM_RESERVED_TOKENS_KEY",
"litellm.proxy.hooks.parallel_request_limiter_v3.TPM_RESERVED_TOKENS_KEY",
)
assert gate.is_in_scope(obj, "litellm") is False
def test_reexported_stdlib_name_is_out_of_scope(self):
assert gate.is_in_scope(FakeObject("litellm.utils.Union", "typing.Union"), "litellm") is False
def test_unresolvable_stdlib_alias_is_out_of_scope(self):
obj = FakeObject("litellm.utils.List", canonical=None, target="typing.List")
assert gate.is_in_scope(obj, "litellm") is False
def test_unresolvable_internal_alias_stays_in_scope(self):
obj = FakeObject("litellm.BedrockLLM", canonical=None, target="litellm.llms.bedrock.BedrockLLM")
assert gate.is_in_scope(obj, "litellm") is True
def test_alias_without_target_falls_back_to_its_own_path(self):
assert gate.is_in_scope(FakeObject("litellm.Router", canonical=None), "litellm") is True
class FakeKind:
def __init__(self, name: str):
self.name = name
class FakeBreakage:
def __init__(self, obj: FakeObject, kind: str, explanation: str):
self.obj = obj
self.kind = FakeKind(kind)
self._explanation = explanation
def explain(self, style: object) -> str:
return self._explanation
class FakeModule:
def __init__(self, *names: str):
self.members = {name: object() for name in names}
class TestBuildDelta:
def test_griffe_color_codes_are_stripped(self):
breakage = FakeBreakage(
FakeObject("litellm.BedrockLLM", "litellm.llms.bedrock.BedrockLLM"),
"OBJECT_REMOVED",
"\x1b[1mlitellm/x.py\x1b[0m:0: BedrockLLM: \x1b[33mPublic object was removed\x1b[39m",
)
built = gate.build_delta(FakeModule(), FakeModule(), iter([breakage]), style=None)
assert built.blocking[0].detail == "litellm/x.py:0: BedrockLLM: Public object was removed"
assert built.blocking[0].line == 7
def test_the_same_breakage_seen_through_two_aliases_is_reported_once(self):
duplicate = [
FakeBreakage(FakeObject("litellm.BedrockLLM", "litellm.llms.bedrock.BedrockLLM"), "OBJECT_REMOVED", "gone"),
FakeBreakage(FakeObject("litellm.BedrockLLM", "litellm.llms.bedrock.BedrockLLM"), "OBJECT_REMOVED", "gone"),
]
built = gate.build_delta(FakeModule(), FakeModule(), iter(duplicate), style=None)
assert len(built.blocking) == 1
def test_out_of_scope_breakages_are_dropped(self):
noise = [
FakeBreakage(FakeObject("litellm.utils.Union", "typing.Union"), "OBJECT_REMOVED", "typing gone"),
FakeBreakage(FakeObject("litellm.proxy.utils.KEY", "litellm.proxy.utils.KEY"), "OBJECT_REMOVED", "k gone"),
]
built = gate.build_delta(FakeModule(), FakeModule(), iter(noise), style=None)
assert built.blocking == ()
def test_value_changes_land_in_advisory_not_blocking(self):
breakage = FakeBreakage(
FakeObject("litellm.router.Span", "litellm.router.Span"), "ATTRIBUTE_CHANGED_VALUE", "Union[A, B] -> A | B"
)
built = gate.build_delta(FakeModule(), FakeModule(), iter([breakage]), style=None)
assert built.blocking == ()
assert len(built.advisory) == 1
def test_added_top_level_names_are_detected_and_private_ones_ignored(self):
built = gate.build_delta(
FakeModule("completion"), FakeModule("completion", "new_flag", "_private"), iter([]), style=None
)
assert built.added_names == ("new_flag",)
def test_removed_top_level_names_are_not_counted_as_additions(self):
built = gate.build_delta(FakeModule("completion", "old_flag"), FakeModule("completion"), iter([]), style=None)
assert built.added_names == ()
class TestRendering:
def test_summary_names_the_declaration_escape_hatch(self):
blocking = delta(blocking=[finding()])
summary = gate.render_summary(blocking, gate.decide(blocking, gate.parse_declaration("fix: x", "")))
assert "BREAKING CHANGE:" in summary
assert "litellm.BedrockLLM" in summary
def test_summary_separates_advisory_from_blocking(self):
mixed = delta(blocking=[finding()], advisory=[finding(kind="ATTRIBUTE_CHANGED_VALUE")])
summary = gate.render_summary(mixed, gate.decide(mixed, gate.parse_declaration("feat!: x", "")))
assert "### Breaking changes" in summary
assert "### Advisory" in summary
def test_clean_summary_has_no_finding_sections(self):
summary = gate.render_summary(delta(), gate.Approved())
assert "###" not in summary
def test_annotations_point_at_the_source_line(self):
rendered = gate.render_annotations([finding()])
assert rendered == "::error file=litellm/x.py,line=1::litellm.BedrockLLM: OBJECT_REMOVED"
def test_annotation_without_a_location_still_renders(self):
located = gate.ApiFinding(kind="OBJECT_REMOVED", path="litellm.X", detail="gone", file=None, line=None)
assert gate.render_annotations([located]) == "::error::gone"