fix(cost_map): mark fireworks_ai/minimax-m3 as vision and pin capabilities verified against live calls

ci_cd/cost_map_guard.py

@@ -1,11 +1,14 @@
 """Guard the cost map on pull requests.
 
-Every pull request whose diff against its merge base touches one of the three cost map files gets the file
-checks: the files parse, the backup copy matches the root file, and the JSON schema is in sync and validates the
-map. A pull request that leaves all three untouched skips them, since merging it keeps the base branch's copies
-and its head tree only carries whatever state the branch was cut from. Pull requests from the cost map sync bot
-(branches named litellm_cost_map_sync_*) always get the file checks and additionally may only touch those three
-files and may only add or update models.
+Every pull request whose diff against its merge base touches one of the three cost map files or the pins file gets
+the file checks: the files parse, the backup copy matches the root file, the JSON schema is in sync and validates the
+map, and every capability pinned in ci_cd/cost_map_pins.json (a value verified against a live provider call where the
+provider's model listing says otherwise) still holds in the map, so a pull request that deletes the pins file fails
+too. A pull request that leaves all of them untouched
+skips the checks, since merging it keeps the base branch's copies and its head tree only carries whatever state the
+branch was cut from. Pull requests from the cost map sync bot (branches named litellm_cost_map_sync_*) always get the
+file checks and additionally may only touch the three cost map files and may only add or update models, so a sync
+can never lower a pinned capability or edit the pins.
 """
 
 from __future__ import annotations
@@ -14,7 +17,7 @@
 import json
 import subprocess
 import sys
-from collections.abc import Sequence
+from collections.abc import Iterator, Sequence
 from dataclasses import dataclass
 from typing import Final
 
@@ -23,17 +26,30 @@
 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"
+PINS_PATH: Final = "ci_cd/cost_map_pins.json"
 GUARDED_PATHS: Final = (COST_MAP_PATH, BACKUP_PATH, SCHEMA_PATH)
+CHECKED_PATHS: Final = (*GUARDED_PATHS, PINS_PATH)
 BOT_BRANCH_PREFIX: Final = "litellm_cost_map_sync_"
 
 CostMap = dict[str, object]
+_MISSING: Final = object()
 
 
 @dataclass(frozen=True, slots=True)
 class Snapshot:
     cost_map: str
     backup: str
     schema: str
+    pins: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class Pin:
+    key: str
+    field: str
+    value: object
+    checked: str
+    source: str
 
 
 def _parse_object(text: str, path: str) -> CostMap | str:
@@ -82,6 +98,48 @@ 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 _pin(key: str, field: str, spec: object) -> Pin | str:
+    malformed: Final = f"{PINS_PATH}: {key}.{field} needs value, checked, and source"
+    if not isinstance(spec, dict) or "value" not in spec:
+        return malformed
+    checked: Final = spec.get("checked")
+    source: Final = spec.get("source")
+    if not isinstance(checked, str) or not isinstance(source, str) or not checked or not source:
+        return malformed
+    return Pin(key, field, spec["value"], checked, source)
+
+
+def _pins(parsed: CostMap) -> Iterator[Pin | str]:
+    for key, fields in parsed.items():
+        if not isinstance(fields, dict):
+            yield f"{PINS_PATH}: {key} must map field names to pins"
+            continue
+        for field, spec in fields.items():
+            yield _pin(key, field, spec)
+
+
+def _pin_violations(parsed: CostMap, entries: dict[str, dict[str, object]]) -> Iterator[str]:
+    for pin in _pins(parsed):
+        if isinstance(pin, str):
+            yield pin
+        elif pin.key not in entries:
+            yield f"{PINS_PATH} pins {pin.key}.{pin.field} but {COST_MAP_PATH} has no {pin.key} entry"
+        elif entries[pin.key].get(pin.field, _MISSING) != pin.value:
+            yield (
+                f"{COST_MAP_PATH}: {pin.key}.{pin.field} must stay {json.dumps(pin.value)} "
+                f"(verified {pin.checked}: {pin.source}); change {PINS_PATH} with new evidence first"
+            )
+
+
+def pin_failures(pins_text: str, head_map: CostMap) -> tuple[str, ...]:
+    if not pins_text:
+        return (f"{PINS_PATH} is missing; restore it, its pins were verified against live provider calls",)
+    parsed: Final = _parse_object(pins_text, PINS_PATH)
+    if isinstance(parsed, str):
+        return (parsed,)
+    return tuple(_pin_violations(parsed, _entries(head_map)))
+
+
 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):
@@ -112,7 +170,7 @@ def _bot_failures(base: Snapshot, head_map: CostMap, changed_files: Sequence[str
 
 
 def touches_cost_map(changed_files: Sequence[str]) -> bool:
-    return any(path in GUARDED_PATHS for path in changed_files)
+    return any(path in CHECKED_PATHS for path in changed_files)
 
 
 def contract_for(bot: bool, changed_files: Sequence[str]) -> str:
@@ -127,7 +185,11 @@ def guard_failures(base: Snapshot, head: Snapshot, changed_files: Sequence[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 ()))
+    return (
+        *_file_failures(head, head_map),
+        *pin_failures(head.pins, head_map),
+        *(_bot_failures(base, head_map, changed_files) if bot else ()),
+    )
 
 
 def _git(*args: str) -> str | None:
@@ -136,7 +198,7 @@ def _git(*args: str) -> str | None:
 
 
 def snapshot(revision: str) -> Snapshot:
-    return Snapshot(*(_git("show", f"{revision}:{path}") or "" for path in GUARDED_PATHS))
+    return Snapshot(*(_git("show", f"{revision}:{path}") or "" for path in CHECKED_PATHS))
 
 
 def changed_files(base: str, head: str) -> tuple[str, ...] | None:

ci_cd/cost_map_pins.json

@@ -0,0 +1,16 @@
+{
+    "fireworks_ai/accounts/fireworks/models/minimax-m3": {
+        "supports_vision": {
+            "value": true,
+            "checked": "2026-09-26",
+            "source": "POST https://api.fireworks.ai/inference/v1/chat/completions read a 512x512 PNG back as digit=7, shape=circle, color=red while GET https://api.fireworks.ai/inference/v1/models still said supports_image_input false and GET https://api.fireworks.ai/v1/serverless/models?format=nested carried no supportsImageInput on the row"
+        }
+    },
+    "fireworks_ai/minimax-m3": {
+        "supports_vision": {
+            "value": true,
+            "checked": "2026-09-26",
+            "source": "POST https://api.fireworks.ai/inference/v1/chat/completions read a 512x512 PNG back as digit=7, shape=circle, color=red while GET https://api.fireworks.ai/inference/v1/models still said supports_image_input false and GET https://api.fireworks.ai/v1/serverless/models?format=nested carried no supportsImageInput on the row"
+        }
+    }
+}

litellm/model_prices_and_context_window_backup.json

@@ -25828,7 +25828,7 @@
         "supports_reasoning": true,
         "supports_response_schema": true,
         "supports_tool_choice": true,
-        "supports_vision": false
+        "supports_vision": true
     },
     "fireworks_ai/accounts/fireworks/models/mixtral-8x22b-instruct-hf": {
         "input_cost_per_token": 1.2e-06,
@@ -26119,7 +26119,7 @@
         "supports_reasoning": true,
         "supports_response_schema": true,
         "supports_tool_choice": true,
-        "supports_vision": false
+        "supports_vision": true
     },
     "fireworks_ai/qwen3p7-plus": {
         "cache_read_input_token_cost": 8e-08,

model_prices_and_context_window.json

@@ -25828,7 +25828,7 @@
         "supports_reasoning": true,
         "supports_response_schema": true,
         "supports_tool_choice": true,
-        "supports_vision": false
+        "supports_vision": true
     },
     "fireworks_ai/accounts/fireworks/models/mixtral-8x22b-instruct-hf": {
         "input_cost_per_token": 1.2e-06,
@@ -26119,7 +26119,7 @@
         "supports_reasoning": true,
         "supports_response_schema": true,
         "supports_tool_choice": true,
-        "supports_vision": false
+        "supports_vision": true
     },
     "fireworks_ai/qwen3p7-plus": {
         "cache_read_input_token_cost": 8e-08,

tests/unit/test_cost_map_guard.py

@@ -51,11 +51,19 @@ 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:
+NO_PINS: Final = "{}\n"
+
+
+def _snapshot(
+    cost_map: dict[str, object], backup: str | None = None, schema: str | None = None, pins: str = NO_PINS
+) -> 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
+        cost_map=text,
+        backup=text if backup is None else backup,
+        schema=rendered if schema is None else schema,
+        pins=pins,
     )
 
 
@@ -109,7 +117,7 @@ def test_schema_validation_errors_are_reported() -> None:
 
 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)
+    head = guard.Snapshot(cost_map=text, backup=text, schema=BASE.schema, pins=NO_PINS)
     (failure,) = _failures(head, bot=False)
     assert "Unclassified keys" in failure and "weird_thing" in failure
 
@@ -129,9 +137,9 @@ def test_human_pr_that_leaves_the_cost_map_alone_skips_the_file_checks() -> None
     assert _failures(unparseable, changed_files=CODE_ONLY, bot=False) == ()
 
 
-@pytest.mark.parametrize("guarded_path", guard.GUARDED_PATHS)
-def test_touching_any_cost_map_file_keeps_the_file_checks(guarded_path: str) -> None:
-    failures: Final = _failures(STALE_HEAD, changed_files=(*CODE_ONLY, guarded_path), bot=False)
+@pytest.mark.parametrize("checked_path", guard.CHECKED_PATHS)
+def test_touching_any_cost_map_file_keeps_the_file_checks(checked_path: str) -> None:
+    failures: Final = _failures(STALE_HEAD, changed_files=(*CODE_ONLY, checked_path), bot=False)
     assert [failure for failure in failures if failure.startswith(guard.BACKUP_PATH)]
     assert [failure for failure in failures if failure.startswith(guard.SCHEMA_PATH)]
 
@@ -175,8 +183,84 @@ def test_bot_may_not_change_special_root_keys() -> None:
     assert _failures(head) == ("bot PRs may not change fallback_generalizations",)
 
 
-def _commit(repo: Path, cost_map: dict[str, object], message: str) -> str:
+PINNED_SOURCE: Final = "a 512x512 probe image was read back over the provider's chat endpoint"
+PINS: Final = _serialize(
+    {"openrouter/a": {"supports_vision": {"value": True, "checked": "2026-09-26", "source": PINNED_SOURCE}}}
+)
+
+
+def test_a_pinned_capability_that_holds_passes_for_humans_and_bots() -> None:
+    head = _snapshot(BASE_MAP, pins=PINS)
+    assert _failures(head, bot=False) == ()
+    assert _failures(head, bot=True) == ()
+
+
+@pytest.mark.parametrize("bot", [False, True])
+def test_lowering_a_pinned_capability_fails_with_its_evidence(bot: bool) -> None:
+    head = _snapshot({**BASE_MAP, "openrouter/a": _entry(supports_vision=False)}, pins=PINS)
+    assert _failures(head, bot=bot) == (
+        f"{guard.COST_MAP_PATH}: openrouter/a.supports_vision must stay true (verified 2026-09-26: {PINNED_SOURCE}); "
+        f"change {guard.PINS_PATH} with new evidence first",
+    )
+
+
+def test_dropping_a_pinned_field_or_model_is_reported() -> None:
+    no_field = _snapshot({**BASE_MAP, "openrouter/a": _entry()}, pins=PINS)
+    assert [failure for failure in _failures(no_field, bot=False) if "must stay true" in failure]
+    no_model = _snapshot({key: value for key, value in BASE_MAP.items() if key != "openrouter/a"}, pins=PINS)
+    assert _failures(no_model, bot=False) == (
+        f"{guard.PINS_PATH} pins openrouter/a.supports_vision but {guard.COST_MAP_PATH} has no openrouter/a entry",
+    )
+
+
+@pytest.mark.parametrize(
+    ("pins", "expected"),
+    [
+        (
+            "{not json",
+            f"{guard.PINS_PATH} is not valid JSON: Expecting property name enclosed in double quotes: "
+            "line 1 column 2 (char 1)",
+        ),
+        (
+            _serialize({"openrouter/a": {"supports_vision": {"value": True}}}),
+            f"{guard.PINS_PATH}: openrouter/a.supports_vision needs value, checked, and source",
+        ),
+        (_serialize({"openrouter/a": True}), f"{guard.PINS_PATH}: openrouter/a must map field names to pins"),
+    ],
+)
+def test_malformed_pins_are_reported(pins: str, expected: str) -> None:
+    assert _failures(_snapshot(BASE_MAP, pins=pins), bot=False) == (expected,)
+
+
+@pytest.mark.parametrize("bot", [False, True])
+def test_a_missing_pins_file_fails(bot: bool) -> None:
+    assert _failures(_snapshot(BASE_MAP, pins=""), bot=bot) == (
+        f"{guard.PINS_PATH} is missing; restore it, its pins were verified against live provider calls",
+    )
+
+
+def test_bot_may_not_touch_the_pins_file() -> None:
+    head = _snapshot(BASE_MAP, pins=PINS)
+    changed = (*guard.GUARDED_PATHS, guard.PINS_PATH)
+    assert _failures(head, changed_files=changed, bot=False) == ()
+    assert _failures(head, changed_files=changed) == (
+        f"bot PRs may only change the cost map files, not {guard.PINS_PATH}",
+    )
+
+
+def test_checked_in_pins_hold_in_the_checked_in_cost_map() -> None:
+    cost_map = json.loads((ROOT / guard.COST_MAP_PATH).read_text())
+    pins = (ROOT / guard.PINS_PATH).read_text()
+    assert json.loads(pins), "the pins file must pin at least one capability"
+    assert guard.pin_failures(pins, cost_map) == ()
+
+
+def _commit(repo: Path, cost_map: dict[str, object], message: str, pins: str | None = None) -> str:
     text = _serialize(cost_map)
+    pins_file = repo / guard.PINS_PATH
+    if pins is not None or not pins_file.exists():
+        pins_file.parent.mkdir(exist_ok=True)
+        pins_file.write_text(NO_PINS if pins is None else pins)
     (repo / guard.COST_MAP_PATH).write_text(text)
     (repo / guard.BACKUP_PATH).parent.mkdir(exist_ok=True)
     (repo / guard.BACKUP_PATH).write_text(text)
@@ -271,3 +355,36 @@ def test_main_rejects_a_bot_pr_that_edits_code(tmp_path: Path) -> None:
     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
+
+
+@pytest.mark.parametrize("head_ref", [BOT_REF, "litellm_fix_pricing"])
+def test_main_reads_the_pins_from_the_head_revision(tmp_path: Path, head_ref: str) -> None:
+    subprocess.run(("git", "init", "-q", str(tmp_path)), check=True)
+    base: Final = _commit(tmp_path, BASE_MAP, "base", pins=PINS)
+    head: Final = _commit(tmp_path, {**BASE_MAP, "openrouter/a": _entry(supports_vision=False)}, "sync lowers it")
+    result: Final = _run_guard(tmp_path, base, head, head_ref)
+    assert result.returncode == 1, result.stdout + result.stderr
+    pin_line: Final = f"- {guard.COST_MAP_PATH}: openrouter/a.supports_vision must stay true (verified 2026-09-26: "
+    assert [line for line in result.stdout.splitlines() if line.startswith(pin_line)]
+
+
+def test_main_fails_a_pr_that_deletes_the_pins_file(tmp_path: Path) -> None:
+    subprocess.run(("git", "init", "-q", str(tmp_path)), check=True)
+    base: Final = _commit(tmp_path, BASE_MAP, "base", pins=PINS)
+    subprocess.run(("git", "rm", "-q", guard.PINS_PATH), cwd=tmp_path, check=True)
+    head: Final = _git_commit(tmp_path, "delete the pins")
+    result: Final = _run_guard(tmp_path, base, head, "litellm_fix_pricing")
+    assert result.returncode == 1, result.stdout + result.stderr
+    missing: Final = f"- {guard.PINS_PATH} is missing; restore it, its pins were verified against live provider calls"
+    assert missing in result.stdout.splitlines()
+
+
+def test_main_checks_a_pins_only_pr_against_the_head_map(tmp_path: Path) -> None:
+    subprocess.run(("git", "init", "-q", str(tmp_path)), check=True)
+    base: Final = _commit(tmp_path, {**BASE_MAP, "openrouter/a": _entry(supports_vision=False)}, "base")
+    (tmp_path / guard.PINS_PATH).parent.mkdir(exist_ok=True)
+    (tmp_path / guard.PINS_PATH).write_text(PINS)
+    head: Final = _git_commit(tmp_path, "pin a capability the map disagrees with")
+    result: Final = _run_guard(tmp_path, base, head, "litellm_fix_pricing")
+    assert result.returncode == 1, result.stdout + result.stderr
+    assert "cost map guard failed (human PR, file checks only):" in result.stdout.splitlines()