feat(ci): add the cost map guard check

Replace test-model-map.yml with a pull_request_target guard that validates the
cost map, its backup, and its generated schema on every PR, and additionally
enforces the sync bot contract on litellm_cost_map_sync_* branches: only the
three cost map files may change, no model or field is removed, and the special
root keys stay untouched.

.github/workflows/cost-map-guard.yml

@@ -0,0 +1,45 @@
+name: Cost map guard
+
+on: # zizmor: ignore[dangerous-triggers] runs the base branch's code only; the PR's cost map files are read as data and never executed
+  pull_request_target:
+    branches:
+      - main
+      - litellm_internal_staging
+      - litellm_oss_staging
+      - "litellm_**"
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number }}
+  cancel-in-progress: true
+
+jobs:
+  cost-map-guard:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          persist-credentials: false
+      - name: Fetch the pull request head and its merge base
+        id: revisions
+        env:
+          GH_TOKEN: ${{ github.token }}
+          BASE_SHA: ${{ github.event.pull_request.base.sha }}
+          HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+        run: |
+          merge_base="$(gh api "repos/${GITHUB_REPOSITORY}/compare/${BASE_SHA}...${HEAD_SHA}" --jq '.merge_base_commit.sha')"
+          git fetch --no-tags --depth=1 origin "$merge_base" "$HEAD_SHA"
+          echo "merge_base=$merge_base" >> "$GITHUB_OUTPUT"
+      - name: Set up uv
+        uses: ./.github/actions/setup-uv-with-retries
+        with:
+          version: "0.10.9"
+      - name: Run the guard
+        env:
+          MERGE_BASE: ${{ steps.revisions.outputs.merge_base }}
+          HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+          HEAD_REF: ${{ github.event.pull_request.head.ref }}
+        run: |
+          uv run --frozen python ci_cd/cost_map_guard.py --base "$MERGE_BASE" --head "$HEAD_SHA" --head-ref "$HEAD_REF"

.github/workflows/test-model-map.yml

@@ -1,37 +0,0 @@
-name: Validate model_prices_and_context_window.json
-
-on:
-  pull_request:
-    branches:
-      - main
-      - litellm_internal_staging
-      - litellm_oss_staging
-      - "litellm_**"
-
-permissions:
-  contents: read
-
-concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
-  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
-
-jobs:
-  validate-model-prices-json:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
-        with:
-          persist-credentials: false
-
-      - name: Validate model_prices_and_context_window.json
-        run: |
-          jq empty model_prices_and_context_window.json
-
-      - name: Set up uv
-        uses: ./.github/actions/setup-uv-with-retries
-        with:
-          version: "0.10.9"
-
-      - name: Check model_prices_and_context_window.schema.json is in sync
-        run: |
-          uv run --frozen python ci_cd/generate_model_prices_schema.py --check

ci_cd/cost_map_guard.py

@@ -0,0 +1,146 @@
+"""Guard the cost map on pull requests.
+
+Every pull request gets the file checks: the three cost map files parse, the backup copy matches the root file,
+and the JSON schema is in sync and validates the map. Pull requests from the cost map sync bot (branches named
+litellm_cost_map_sync_*) additionally may only touch those three files and may only add or update models.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+import sys
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Final
+
+from generate_model_prices_schema import SPECIAL_ROOT_KEYS, build_schema, render, validation_errors
+
+COST_MAP_PATH: Final = "model_prices_and_context_window.json"
+BACKUP_PATH: Final = "litellm/model_prices_and_context_window_backup.json"
+SCHEMA_PATH: Final = "model_prices_and_context_window.schema.json"
+GUARDED_PATHS: Final = (COST_MAP_PATH, BACKUP_PATH, SCHEMA_PATH)
+BOT_BRANCH_PREFIX: Final = "litellm_cost_map_sync_"
+
+CostMap = dict[str, object]
+
+
+@dataclass(frozen=True, slots=True)
+class Snapshot:
+    cost_map: str
+    backup: str
+    schema: str
+
+
+def _parse_object(text: str, path: str) -> CostMap | str:
+    try:
+        parsed: Final = json.loads(text)
+    except json.JSONDecodeError as error:
+        return f"{path} is not valid JSON: {error}"
+    return parsed if isinstance(parsed, dict) else f"{path} must be a JSON object at the root"
+
+
+def _rendered_schema(cost_map: CostMap) -> str:
+    try:
+        return render(build_schema(cost_map))
+    except SystemExit as error:
+        return str(error)
+
+
+def _file_failures(head: Snapshot, head_map: CostMap) -> tuple[str, ...]:
+    schema_text: Final = _rendered_schema(head_map)
+    if not schema_text.startswith("{"):
+        return (schema_text,)
+    backup_failure: Final = (
+        ()
+        if head.backup == head.cost_map
+        else (f"{BACKUP_PATH} differs from {COST_MAP_PATH}; copy the root file over it",)
+    )
+    schema_failure: Final = (
+        ()
+        if head.schema == schema_text
+        else (
+            f"{SCHEMA_PATH} is out of sync with {COST_MAP_PATH}; "
+            "run `python ci_cd/generate_model_prices_schema.py` and commit the result",
+        )
+    )
+    return (
+        *backup_failure,
+        *schema_failure,
+        *(
+            f"{COST_MAP_PATH} does not validate against its schema: {error}"
+            for error in validation_errors(head_map, json.loads(schema_text))[:20]
+        ),
+    )
+
+
+def _entries(cost_map: CostMap) -> dict[str, dict[str, object]]:
+    return {key: entry for key, entry in cost_map.items() if isinstance(entry, dict)}
+
+
+def _bot_failures(base: Snapshot, head_map: CostMap, changed_files: Sequence[str]) -> tuple[str, ...]:
+    base_map: Final = _parse_object(base.cost_map, COST_MAP_PATH)
+    if isinstance(base_map, str):
+        return (f"merge base: {base_map}",)
+    base_entries: Final = _entries(base_map)
+    head_entries: Final = _entries(head_map)
+    removed_fields: Final = tuple(
+        f"{key}.{field}"
+        for key, entry in base_entries.items()
+        if key in head_entries
+        for field in entry
+        if field not in head_entries[key]
+    )
+    return (
+        *(
+            f"bot PRs may only change the cost map files, not {path}"
+            for path in changed_files
+            if path not in GUARDED_PATHS
+        ),
+        *(f"bot PRs may not remove models: {key}" for key in base_map if key not in head_map),
+        *(f"bot PRs may not remove fields: {ref}" for ref in removed_fields),
+        *(
+            f"bot PRs may not change {key}"
+            for key in sorted(SPECIAL_ROOT_KEYS)
+            if base_map.get(key) != head_map.get(key)
+        ),
+    )
+
+
+def guard_failures(base: Snapshot, head: Snapshot, changed_files: Sequence[str], bot: bool) -> tuple[str, ...]:
+    head_map: Final = _parse_object(head.cost_map, COST_MAP_PATH)
+    if isinstance(head_map, str):
+        return (head_map,)
+    return (*_file_failures(head, head_map), *(_bot_failures(base, head_map, changed_files) if bot else ()))
+
+
+def _git(*args: str) -> str:
+    result: Final = subprocess.run(("git", *args), check=False, capture_output=True, text=True)
+    return result.stdout if result.returncode == 0 else ""
+
+
+def snapshot(revision: str) -> Snapshot:
+    return Snapshot(*(_git("show", f"{revision}:{path}") for path in GUARDED_PATHS))
+
+
+def main(argv: Sequence[str]) -> int:
+    parser: Final = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base", required=True, help="merge base of the pull request")
+    parser.add_argument("--head", required=True, help="head commit of the pull request")
+    parser.add_argument("--head-ref", required=True, help="head branch name of the pull request")
+    args: Final = parser.parse_args(argv)
+    bot: Final = args.head_ref.startswith(BOT_BRANCH_PREFIX)
+    changed_files: Final = tuple(_git("diff", "--name-only", args.base, args.head).splitlines())
+    failures: Final = guard_failures(snapshot(args.base), snapshot(args.head), changed_files, bot)
+    contract: Final = "bot contract enforced" if bot else "human PR, file checks only"
+    if failures:
+        print(f"cost map guard failed ({contract}):")
+        print("\n".join(f"- {failure}" for failure in failures))
+        return 1
+    print(f"cost map guard passed ({contract})")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

tests/test_litellm/test_cost_map_guard.py

@@ -0,0 +1,195 @@
+import importlib.util
+import json
+import subprocess
+import sys
+from pathlib import Path
+from types import ModuleType
+from typing import Final
+
+import pytest
+
+ROOT: Final = Path(__file__).resolve().parents[2]
+CI_CD: Final = ROOT / "ci_cd"
+
+
+def _load(name: str) -> ModuleType:
+    spec = importlib.util.spec_from_file_location(name, CI_CD / f"{name}.py")
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+schema_module: Final = _load("generate_model_prices_schema")
+guard: Final = _load("cost_map_guard")
+
+MAP_FILES: Final = (guard.COST_MAP_PATH,)
+BOT_REF: Final = "litellm_cost_map_sync_2026-09-04T12-00Z"
+
+
+def _entry(price: float = 1e-06, **extra: object) -> dict[str, object]:
+    return {
+        "input_cost_per_token": price,
+        "output_cost_per_token": price * 2,
+        "litellm_provider": "openrouter",
+        "mode": "chat",
+        "max_tokens": 4096,
+        **extra,
+    }
+
+
+BASE_MAP: Final = {
+    "sample_spec": {"input_cost_per_token": "USD per prompt token"},
+    "fallback_generalizations": {"rules": [{"name": "r", "pattern": "^x"}]},
+    "openrouter/a": _entry(supports_vision=True),
+    "openrouter/b": _entry(2e-06),
+}
+
+
+def _serialize(cost_map: dict[str, object]) -> str:
+    return json.dumps(cost_map, indent=4, ensure_ascii=False) + "\n"
+
+
+def _snapshot(cost_map: dict[str, object], backup: str | None = None, schema: str | None = None) -> object:
+    text = _serialize(cost_map)
+    rendered = schema_module.render(schema_module.build_schema(cost_map))
+    return guard.Snapshot(
+        cost_map=text, backup=text if backup is None else backup, schema=rendered if schema is None else schema
+    )
+
+
+BASE: Final = _snapshot(BASE_MAP)
+
+
+def _failures(head: object, changed_files: tuple[str, ...] = MAP_FILES, bot: bool = True) -> tuple[str, ...]:
+    return guard.guard_failures(BASE, head, changed_files, bot)
+
+
+def test_in_sync_files_pass_for_humans_and_bots() -> None:
+    assert _failures(BASE, bot=False) == ()
+    assert _failures(BASE, bot=True) == ()
+
+
+def test_bot_may_add_and_reprice_models() -> None:
+    head = _snapshot({**BASE_MAP, "openrouter/a": _entry(9e-06, supports_vision=True), "openrouter/c": _entry()})
+    assert _failures(head) == ()
+
+
+def test_broken_json_is_reported() -> None:
+    head = guard.Snapshot(cost_map="{not json", backup="{not json", schema="{}")
+    assert _failures(head, bot=False) == (
+        f"{guard.COST_MAP_PATH} is not valid JSON: Expecting property name enclosed in double quotes: line 1 column 2 (char 1)",
+    )
+
+
+def test_non_object_root_is_reported() -> None:
+    head = guard.Snapshot(cost_map="[]", backup="[]", schema="{}")
+    assert _failures(head, bot=False) == (f"{guard.COST_MAP_PATH} must be a JSON object at the root",)
+
+
+def test_backup_drift_is_reported() -> None:
+    head = _snapshot(BASE_MAP, backup=_serialize({**BASE_MAP, "openrouter/b": _entry(3e-06)}))
+    assert [failure for failure in _failures(head, bot=False) if failure.startswith(guard.BACKUP_PATH)]
+
+
+def test_schema_out_of_sync_is_reported() -> None:
+    head = _snapshot({**BASE_MAP, "openrouter/c": _entry(supports_audio_input=True)}, schema=BASE.schema)
+    assert [failure for failure in _failures(head, bot=False) if failure.startswith(guard.SCHEMA_PATH)]
+
+
+def test_schema_validation_errors_are_reported() -> None:
+    head = _snapshot({**BASE_MAP, "openrouter/c": _entry(-1e-06)})
+    prefix = f"{guard.COST_MAP_PATH} does not validate against its schema: openrouter/c."
+    assert [failure.removeprefix(prefix).split(":")[0] for failure in _failures(head, bot=False)] == [
+        "input_cost_per_token",
+        "output_cost_per_token",
+    ]
+
+
+def test_unclassified_entry_key_is_reported() -> None:
+    text = _serialize({**BASE_MAP, "openrouter/c": _entry(weird_thing=1)})
+    head = guard.Snapshot(cost_map=text, backup=text, schema=BASE.schema)
+    (failure,) = _failures(head, bot=False)
+    assert "Unclassified keys" in failure and "weird_thing" in failure
+
+
+def test_bot_may_only_touch_the_cost_map_files() -> None:
+    changed = (*guard.GUARDED_PATHS, "litellm/utils.py", ".github/workflows/cost-map-guard.yml")
+    assert _failures(BASE, changed_files=changed, bot=False) == ()
+    assert _failures(BASE, changed_files=changed) == (
+        "bot PRs may only change the cost map files, not litellm/utils.py",
+        "bot PRs may only change the cost map files, not .github/workflows/cost-map-guard.yml",
+    )
+
+
+def test_bot_may_not_remove_models() -> None:
+    head = _snapshot({key: value for key, value in BASE_MAP.items() if key != "openrouter/b"})
+    assert _failures(head, bot=False) == ()
+    assert _failures(head) == ("bot PRs may not remove models: openrouter/b",)
+
+
+def test_bot_may_not_remove_fields() -> None:
+    head = _snapshot({**BASE_MAP, "openrouter/a": _entry()})
+    assert _failures(head, bot=False) == ()
+    assert _failures(head) == ("bot PRs may not remove fields: openrouter/a.supports_vision",)
+
+
+def test_bot_may_not_change_special_root_keys() -> None:
+    head = _snapshot({**BASE_MAP, "fallback_generalizations": {"rules": []}})
+    assert _failures(head, bot=False) == ()
+    assert _failures(head) == ("bot PRs may not change fallback_generalizations",)
+
+
+def _commit(repo: Path, cost_map: dict[str, object], message: str) -> str:
+    text = _serialize(cost_map)
+    (repo / guard.COST_MAP_PATH).write_text(text)
+    (repo / guard.BACKUP_PATH).parent.mkdir(exist_ok=True)
+    (repo / guard.BACKUP_PATH).write_text(text)
+    (repo / guard.SCHEMA_PATH).write_text(schema_module.render(schema_module.build_schema(cost_map)))
+    subprocess.run(("git", "add", "-A"), cwd=repo, check=True)
+    subprocess.run(
+        ("git", "-c", "user.name=t", "-c", "user.email=t@example.com", "commit", "-q", "-m", message),
+        cwd=repo,
+        check=True,
+    )
+    return subprocess.run(
+        ("git", "rev-parse", "HEAD"), cwd=repo, check=True, capture_output=True, text=True
+    ).stdout.strip()
+
+
+def _run_guard(repo: Path, base: str, head: str, head_ref: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        (sys.executable, str(CI_CD / "cost_map_guard.py"), "--base", base, "--head", head, "--head-ref", head_ref),
+        cwd=repo,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+@pytest.mark.parametrize(
+    ("head_ref", "expected_code", "expected_line"),
+    [
+        (BOT_REF, 1, "- bot PRs may not remove models: openrouter/b"),
+        ("litellm_fix_pricing", 0, "cost map guard passed (human PR, file checks only)"),
+    ],
+)
+def test_main_reads_both_revisions_from_git(
+    tmp_path: Path, head_ref: str, expected_code: int, expected_line: str
+) -> None:
+    subprocess.run(("git", "init", "-q", str(tmp_path)), check=True)
+    base = _commit(tmp_path, BASE_MAP, "base")
+    head = _commit(tmp_path, {key: value for key, value in BASE_MAP.items() if key != "openrouter/b"}, "head")
+    result = _run_guard(tmp_path, base, head, head_ref)
+    assert result.returncode == expected_code, result.stdout + result.stderr
+    assert expected_line in result.stdout.splitlines()
+
+
+def test_main_rejects_a_bot_pr_that_edits_code(tmp_path: Path) -> None:
+    subprocess.run(("git", "init", "-q", str(tmp_path)), check=True)
+    base = _commit(tmp_path, BASE_MAP, "base")
+    (tmp_path / "litellm" / "utils.py").write_text("print('hi')\n")
+    head = _commit(tmp_path, {**BASE_MAP, "openrouter/c": _entry()}, "head")
+    assert _run_guard(tmp_path, base, head, BOT_REF).returncode == 1
+    assert _run_guard(tmp_path, base, head, "litellm_fix_pricing").returncode == 0