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

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

Fireworks' serverless minimax-m3 answers image inputs, but its model
listings say supports_image_input false, so the last two registry syncs
flipped supports_vision back to false and the Fireworks pre-flight turned
every image request into a 400 without ever calling the provider.

Flip both minimax-m3 rows back to supports_vision true and record the
live-call evidence in ci_cd/cost_map_pins.json. The cost map guard now
fails any PR that lowers a pinned capability, quoting the evidence, and
the sync bot may not touch the pins file.

* fix(cost_map): fail the guard when the pins file is missing

* fix(cost_map_guard): skip the pin checks on a branch cut before the pins file existed

* fix(ci): check cost map pins against the merge result and keep price sync PRs off the pins file

* fix(ci): run the cost map guard against the live base branch tip and re-run it when main's pins change

* fix(ci): pass the live base tip to the cost map guard through env and re-run every guard run on a pin change

main now runs the guard on pull_request with the base branch's copy of the script, so the workflow passes the
live tip in COST_MAP_GUARD_TARGET, which the older script ignores. The recheck reads past GraphQL's 100-file
cap, re-runs every guard run for a shared head, and fails when a re-run cannot start. The guard tells a merge
conflict apart from a failed merge-tree call.

* fix(ci): run the cost map pins recheck on litellm_ base branches too

* fix(ci): bar bot-opened PRs from the pins, recheck pin PRs on cost map changes, and skip the merge for untouched PRs

* fix(ci): read the pushed files from the compare API in the cost map pins recheck

The push event GitHub hands to Actions leaves each commit's added, modified and
removed lists null, so the recheck never saw a pins change and re-ran only the PRs
that touch the pins. Compare the push's before and after commits instead, and
re-check every cost map PR when the branch is new, the push was forced, the
compare fails, or it lists 300 files, the most it returns.

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>

.github/CODEOWNERS

@@ -1,5 +1,6 @@
 /model_prices_and_context_window.json @mateo-berri @ryan-crabbe-berri @kerry-berri @joshua-berri
 /litellm/model_prices_and_context_window_backup.json @mateo-berri @ryan-crabbe-berri @kerry-berri @joshua-berri
+/ci_cd/cost_map_pins.json @mateo-berri @ryan-crabbe-berri @kerry-berri @joshua-berri
 /helm/ @yassin-berriai
 /terraform/ @yassin-berriai
 /.github/workflows/zizmor.yml @yuneng-berri

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

@@ -21,16 +21,18 @@ jobs:
         with:
           ref: ${{ github.event.pull_request.base.sha }}
           persist-credentials: false
-      - name: Fetch the pull request head and its merge base
+      - name: Fetch the pull request head, its merge base, and the base branch tip
         id: revisions
         env:
           GH_TOKEN: ${{ github.token }}
-          BASE_SHA: ${{ github.event.pull_request.base.sha }}
+          BASE_REF: ${{ github.event.pull_request.base.ref }}
           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"
+          target="$(gh api "repos/${GITHUB_REPOSITORY}/branches/${BASE_REF}" --jq '.commit.sha')"
+          merge_base="$(gh api "repos/${GITHUB_REPOSITORY}/compare/${target}...${HEAD_SHA}" --jq '.merge_base_commit.sha')"
+          git fetch --no-tags --depth=1 origin "$merge_base" "$HEAD_SHA" "$target"
           echo "merge_base=$merge_base" >> "$GITHUB_OUTPUT"
+          echo "target=$target" >> "$GITHUB_OUTPUT"
       - name: Set up uv
         uses: ./.github/actions/setup-uv-with-retries
         with:
@@ -38,6 +40,8 @@ jobs:
       - name: Run the guard
         env:
           MERGE_BASE: ${{ steps.revisions.outputs.merge_base }}
+          COST_MAP_GUARD_TARGET: ${{ steps.revisions.outputs.target }}
+          COST_MAP_GUARD_AUTHOR_TYPE: ${{ github.event.pull_request.user.type }}
           HEAD_SHA: ${{ github.event.pull_request.head.sha }}
           HEAD_REF: ${{ github.event.pull_request.head.ref }}
         run: |

.github/workflows/cost-map-pins-recheck.yml

@@ -0,0 +1,91 @@
+name: Cost map pins recheck
+
+on:
+  push:
+    branches:
+      - main
+      - "litellm_**"
+    paths:
+      - model_prices_and_context_window.json
+      - litellm/model_prices_and_context_window_backup.json
+      - ci_cd/cost_map_pins.json
+
+permissions: {}
+
+jobs:
+  rerun-cost-map-guard:
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    permissions:
+      actions: write
+      contents: read
+      pull-requests: read
+    steps:
+      - name: Re-run the cost map guard on open pull requests the push could break
+        env:
+          GH_TOKEN: ${{ github.token }}
+          BASE_REF: ${{ github.ref_name }}
+          BEFORE_SHA: ${{ github.event.before }}
+          AFTER_SHA: ${{ github.event.after }}
+          FORCED: ${{ github.event.forced }}
+          COST_MAP_PATHS: '["model_prices_and_context_window.json", "litellm/model_prices_and_context_window_backup.json", "ci_cd/cost_map_pins.json"]'
+          PINS_PATHS: '["ci_cd/cost_map_pins.json"]'
+        run: |
+          if [[ "$BEFORE_SHA" =~ ^0+$ || "$FORCED" == true ]] \
+            || ! gh api "repos/${GITHUB_REPOSITORY}/compare/${BEFORE_SHA}...${AFTER_SHA}?per_page=1" \
+              --jq '(env.PINS_PATHS | fromjson) as $pins | (.files | length) < 300 and all(.files[]; [.filename, .previous_filename] | any(IN($pins[])) | not)' \
+              | grep -qx true; then
+            export RECHECK_PATHS="$COST_MAP_PATHS"
+          else
+            export RECHECK_PATHS="$PINS_PATHS"
+          fi
+          rerun_cutoff="$(date -u -d '30 days ago' +%Y-%m-%dT%H:%M:%SZ)"
+          candidates="$(gh api graphql --paginate -F owner="${GITHUB_REPOSITORY_OWNER}" -F name="${GITHUB_REPOSITORY#*/}" -F base="$BASE_REF" \
+            -f query='query($owner: String!, $name: String!, $base: String!, $endCursor: String) {
+              repository(owner: $owner, name: $name) {
+                pullRequests(states: OPEN, baseRefName: $base, first: 25, after: $endCursor) {
+                  nodes { number headRefOid changedFiles files(first: 100) { nodes { path } } }
+                  pageInfo { hasNextPage endCursor }
+                }
+              }
+            }' \
+            --jq '(env.RECHECK_PATHS | fromjson) as $paths | .data.repository.pullRequests.nodes[]
+              | {number, headRefOid, listed: any(.files.nodes[].path; IN($paths[])), unlisted: (.changedFiles > 100)}
+              | select(.listed or .unlisted) | "\(.number) \(.headRefOid) \(.listed)"')"
+          failed=0
+          declare -A rerun_heads
+          while read -r number head_sha listed; do
+            [ -n "$number" ] || continue
+            if [ "$listed" = false ]; then
+              if ! changed="$(gh api --paginate "repos/${GITHUB_REPOSITORY}/pulls/${number}/files?per_page=100" \
+                --jq '(env.RECHECK_PATHS | fromjson) as $paths | .[].filename | select(IN($paths[]))')"; then
+                echo "::error::could not list the files of #$number"
+                failed=1
+                continue
+              fi
+              [ -n "$changed" ] || continue
+            fi
+            [ -z "${rerun_heads[$head_sha]:-}" ] || continue
+            rerun_heads[$head_sha]=1
+            if ! runs="$(gh run list --repo "$GITHUB_REPOSITORY" --workflow cost-map-guard.yml --commit "$head_sha" --limit 100 --json databaseId,createdAt --jq '.[] | "\(.databaseId) \(.createdAt)"')"; then
+              echo "::error::could not list the cost map guard runs of #$number"
+              failed=1
+              continue
+            fi
+            if [ -z "$runs" ]; then
+              echo "::warning::no cost map guard run found for #$number"
+              continue
+            fi
+            while read -r run_id created_at; do
+              if [[ "$created_at" < "$rerun_cutoff" ]]; then
+                echo "::warning::cost map guard run $run_id for #$number is over 30 days old, so GitHub cannot re-run it; #$number gets the pin checks on its next push"
+                continue
+              fi
+              gh run watch "$run_id" --repo "$GITHUB_REPOSITORY" > /dev/null || true
+              if ! gh run rerun "$run_id" --repo "$GITHUB_REPOSITORY"; then
+                echo "::error::could not re-run cost map guard run $run_id for #$number"
+                failed=1
+              fi
+            done <<< "$runs"
+          done <<< "$candidates"
+          exit "$failed"

ci_cd/cost_map_guard.py

@@ -1,37 +1,67 @@
 """Guard the cost map on pull requests.
 
-Every pull request whose diff against its merge base touches one of the two cost map files gets the file
-checks: the files parse and the backup copy matches the root file. A pull request that leaves both 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 two files and may only add or update models.
+Every pull request whose diff against its merge base touches one of the two cost map files or the pins file gets the
+file checks: the files parse, the backup copy matches the root file, 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 tree that merging the pull request into its base branch produces, so a branch cut before
+a pin was added cannot lower it, and a pull request that deletes the pins file fails too. 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 the two cost map files and may
+only add or update models, so a sync can never lower a pinned capability or edit the pins. Price sync pull requests
+(branches named litellm-providers/price-sync*) and pull requests a bot account opened (COST_MAP_GUARD_AUTHOR_TYPE is
+Bot) get the human checks and may not touch the pins file either.
 """
 
 from __future__ import annotations
 
 import argparse
 import json
+import os
 import subprocess
 import sys
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterator, Mapping, Sequence
 from dataclasses import dataclass
 from types import MappingProxyType
 from typing import Final
 
 COST_MAP_PATH: Final = "model_prices_and_context_window.json"
 BACKUP_PATH: Final = "litellm/model_prices_and_context_window_backup.json"
+PINS_PATH: Final = "ci_cd/cost_map_pins.json"
 GUARDED_PATHS: Final = (COST_MAP_PATH, BACKUP_PATH)
+CHECKED_PATHS: Final = (*GUARDED_PATHS, PINS_PATH)
 BOT_BRANCH_PREFIX: Final = "litellm_cost_map_sync_"
+PRICE_SYNC_BRANCH_PREFIX: Final = "litellm-providers/price-sync"
+TARGET_ENV: Final = "COST_MAP_GUARD_TARGET"
+AUTHOR_TYPE_ENV: Final = "COST_MAP_GUARD_AUTHOR_TYPE"
+BOT_AUTHOR_TYPE: Final = "Bot"
+MERGE_CONFLICT_EXIT_CODE: Final = 1
 SPECIAL_ROOT_KEYS: Final = frozenset({"sample_spec", "fallback_generalizations"})
 EVALUATION_OUTPUT_PRICE_SOURCES: Final[Mapping[str, str]] = MappingProxyType({})
 
 CostMap = dict[str, object]
+_MISSING: Final = object()
 
 
 @dataclass(frozen=True, slots=True)
 class Snapshot:
     cost_map: str
     backup: str
+    pins: str
+
+
+@dataclass(frozen=True, slots=True)
+class MergeFailed:
+    stderr: 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:
@@ -91,6 +121,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: Mapping[str, Mapping[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):
@@ -120,8 +192,18 @@ def _bot_failures(base: Snapshot, head_map: CostMap, changed_files: Sequence[str
     )
 
 
+def pin_author_failures(head_ref: str, author_type: str, changed_files: Sequence[str]) -> tuple[str, ...]:
+    if PINS_PATH not in changed_files:
+        return ()
+    if head_ref.startswith(PRICE_SYNC_BRANCH_PREFIX):
+        return (f"price sync PRs may not change {PINS_PATH}; a pin moves only in a human PR with new evidence",)
+    if author_type == BOT_AUTHOR_TYPE:
+        return (f"PRs opened by a bot may not change {PINS_PATH}; a pin moves only in a human PR with new evidence",)
+    return ()
+
+
 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:
@@ -130,9 +212,21 @@ def contract_for(bot: bool, changed_files: Sequence[str]) -> str:
     return "human PR, file checks only" if touches_cost_map(changed_files) else "human PR, cost map untouched"
 
 
+def _merged_pin_failures(base: Snapshot, merged: Snapshot | None) -> tuple[str, ...]:
+    if merged is None:
+        return ("the pull request does not merge cleanly into its base branch; merge the base branch in first",)
+    if not merged.pins and not base.pins:
+        return ()
+    merged_map: Final = _parse_object(merged.cost_map, COST_MAP_PATH)
+    if isinstance(merged_map, str):
+        return (f"after merging into the base branch, {merged_map}",)
+    return pin_failures(merged.pins, merged_map)
+
+
 def guard_failures(
     base: Snapshot,
     head: Snapshot,
+    merged: Snapshot | None,
     changed_files: Sequence[str],
     bot: bool,
     *,
@@ -147,6 +241,7 @@ def guard_failures(
         *_file_failures(
             head, head_map, evaluation_output_price_sources=evaluation_output_price_sources
         ),
+        *_merged_pin_failures(base, merged),
         *(_bot_failures(base, head_map, changed_files) if bot else ()),
     )
 
@@ -157,7 +252,22 @@ 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 merged_tree(merge_base: str, target: str, head: str) -> str | MergeFailed | None:
+    result: Final = subprocess.run(
+        ("git", "merge-tree", "--write-tree", f"--merge-base={merge_base}", target, head),
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    tree: Final = result.stdout.split("\n", 1)[0]
+    if result.returncode == 0:
+        return tree
+    if result.returncode == MERGE_CONFLICT_EXIT_CODE and tree:
+        return None
+    return MergeFailed(result.stderr.strip())
 
 
 def changed_files(base: str, head: str) -> tuple[str, ...] | None:
@@ -169,6 +279,13 @@ 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")
+    env_target: Final = os.environ.get(TARGET_ENV)
+    parser.add_argument(
+        "--target",
+        default=env_target,
+        required=env_target is None,
+        help=f"tip of the base branch the pull request merges into, defaulting to ${TARGET_ENV}",
+    )
     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)
@@ -176,7 +293,19 @@ def main(argv: Sequence[str]) -> int:
     if changed is None:
         print(f"cost map guard failed: git diff {args.base} {args.head} failed, so the changed files are unknown")
         return 1
-    failures: Final = guard_failures(snapshot(args.base), snapshot(args.head), changed, bot)
+    if not bot and not touches_cost_map(changed):
+        print(f"cost map guard passed ({contract_for(bot, changed)})")
+        return 0
+    merged: Final = merged_tree(args.base, args.target, args.head)
+    if isinstance(merged, MergeFailed):
+        print(f"cost map guard failed: git merge-tree into {args.target} failed: {merged.stderr}")
+        return 1
+    failures: Final = (
+        *guard_failures(
+            snapshot(args.base), snapshot(args.head), None if merged is None else snapshot(merged), changed, bot
+        ),
+        *pin_author_failures(args.head_ref, os.environ.get(AUTHOR_TYPE_ENV, ""), changed),
+    )
     contract: Final = contract_for(bot, changed)
     if failures:
         print(f"cost map guard failed ({contract}):")

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

@@ -28071,7 +28071,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,
@@ -28345,7 +28345,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

@@ -28071,7 +28071,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,
@@ -28345,7 +28345,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

@@ -1,5 +1,6 @@
 import importlib.util
 import json
+import os
 import subprocess
 import sys
 from collections.abc import Mapping
@@ -51,12 +52,16 @@ 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) -> object:
+NO_PINS: Final = "{}\n"
+
+
+def _snapshot(cost_map: dict[str, object], backup: str | None = None, pins: str = NO_PINS) -> object:
     text = _serialize(cost_map)
-    return guard.Snapshot(cost_map=text, backup=text if backup is None else backup)
+    return guard.Snapshot(cost_map=text, backup=text if backup is None else backup, pins=pins)
 
 
 BASE: Final = _snapshot(BASE_MAP)
+UP_TO_DATE: Final = object()
 
 
 def _failures(
@@ -64,10 +69,13 @@ def _failures(
     changed_files: tuple[str, ...] = MAP_FILES,
     bot: bool = True,
     evaluation_output_price_sources: Mapping[str, str] = guard.EVALUATION_OUTPUT_PRICE_SOURCES,
+    base: object = BASE,
+    merged: object = UP_TO_DATE,
 ) -> tuple[str, ...]:
     return guard.guard_failures(
-        BASE,
+        base,
         head,
+        head if merged is UP_TO_DATE else merged,
         changed_files,
         bot,
         evaluation_output_price_sources=evaluation_output_price_sources,
@@ -129,14 +137,14 @@ def test_bot_may_add_and_reprice_models() -> None:
 
 
 def test_broken_json_is_reported() -> None:
-    head = guard.Snapshot(cost_map="{not json", backup="{not json")
+    head = guard.Snapshot(cost_map="{not json", backup="{not json", pins=NO_PINS)
     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="[]")
+    head = guard.Snapshot(cost_map="[]", backup="[]", pins=NO_PINS)
     assert _failures(head, bot=False) == (f"{guard.COST_MAP_PATH} must be a JSON object at the root",)
 
 
@@ -154,15 +162,15 @@ CODE_ONLY: Final = (
 
 
 def test_human_pr_that_leaves_the_cost_map_alone_skips_the_file_checks() -> None:
-    unparseable: Final = guard.Snapshot(cost_map="{not json", backup="")
+    unparseable: Final = guard.Snapshot(cost_map="{not json", backup="", pins=NO_PINS)
     assert _failures(STALE_HEAD, changed_files=CODE_ONLY, bot=False) == ()
     assert _failures(STALE_HEAD, changed_files=(), bot=False) == ()
     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)]
 
 
@@ -204,8 +212,155 @@ 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) == ()
+
+
+LOWERED: Final = (
+    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"
+)
+NO_VISION_MAP: Final = {**BASE_MAP, "openrouter/a": _entry(supports_vision=False)}
+
+
+@pytest.mark.parametrize("bot", [False, True])
+def test_lowering_a_pinned_capability_fails_with_its_evidence(bot: bool) -> None:
+    assert _failures(_snapshot(NO_VISION_MAP, pins=PINS), bot=bot) == (LOWERED,)
+
+
+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,)
+
+
+MISSING_PINS: Final = f"{guard.PINS_PATH} is missing; restore it, its pins were verified against live provider calls"
+
+
+@pytest.mark.parametrize("bot", [False, True])
+def test_a_pr_that_deletes_the_pins_file_fails(bot: bool) -> None:
+    head: Final = _snapshot(BASE_MAP, pins="")
+    assert _failures(head, bot=bot, base=_snapshot(BASE_MAP, pins=PINS)) == (MISSING_PINS,)
+
+
+@pytest.mark.parametrize("bot", [False, True])
+def test_pin_checks_are_skipped_when_no_side_of_the_merge_has_the_pins_file(bot: bool) -> None:
+    assert _failures(_snapshot(NO_VISION_MAP, pins=""), bot=bot, base=_snapshot(NO_VISION_MAP, pins="")) == ()
+
+
+@pytest.mark.parametrize("bot", [False, True])
+def test_a_branch_cut_before_a_pin_was_added_cannot_lower_it(bot: bool) -> None:
+    head: Final = _snapshot(NO_VISION_MAP, pins="")
+    merged: Final = _snapshot(NO_VISION_MAP, pins=PINS)
+    assert _failures(head, bot=bot, base=_snapshot(BASE_MAP, pins=""), merged=merged) == (LOWERED,)
+
+
+@pytest.mark.parametrize("bot", [False, True])
+def test_a_branch_cut_before_a_pin_was_added_passes_when_merging_keeps_the_pinned_value(bot: bool) -> None:
+    repriced: Final = {"openrouter/b": _entry(3e-06)}
+    head: Final = _snapshot({**NO_VISION_MAP, **repriced}, pins="")
+    merged: Final = _snapshot({**BASE_MAP, **repriced}, pins=PINS)
+    assert _failures(head, bot=bot, base=_snapshot(NO_VISION_MAP, pins=""), merged=merged) == ()
+
+
+def test_a_pr_that_does_not_merge_cleanly_is_told_to_merge_the_base_branch_in() -> None:
+    assert _failures(BASE, bot=False, merged=None) == (
+        "the pull request does not merge cleanly into its base branch; merge the base branch in first",
+    )
+
+
+def test_a_merge_result_that_is_not_valid_json_is_reported() -> None:
+    merged: Final = guard.Snapshot(cost_map="{broken", backup="{broken", pins=PINS)
+    assert _failures(BASE, bot=False, merged=merged) == (
+        f"after merging into the base branch, {guard.COST_MAP_PATH} is not valid JSON: "
+        "Expecting property name enclosed in double quotes: line 1 column 2 (char 1)",
+    )
+
+
+def test_a_pr_that_adds_the_pins_file_gets_its_pins_checked() -> None:
+    head: Final = _snapshot({**BASE_MAP, "openrouter/a": _entry(supports_vision=False)}, pins=PINS)
+    changed: Final = (*MAP_FILES, guard.PINS_PATH)
+    failures: Final = _failures(head, changed_files=changed, bot=False, base=_snapshot(BASE_MAP, pins=""))
+    assert [failure for failure in failures if "openrouter/a.supports_vision must stay true" in failure]
+
+
+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}",
+    )
+
+
+PRICE_SYNC_PIN_FAILURE: Final = (
+    f"price sync PRs may not change {guard.PINS_PATH}; a pin moves only in a human PR with new evidence"
+)
+BOT_AUTHOR_PIN_FAILURE: Final = (
+    f"PRs opened by a bot may not change {guard.PINS_PATH}; a pin moves only in a human PR with new evidence"
+)
+
+
+@pytest.mark.parametrize(
+    ("head_ref", "author_type", "changed", "expected"),
+    [
+        ("litellm-providers/price-sync-fireworks-ai", "User", (*MAP_FILES, guard.PINS_PATH), (PRICE_SYNC_PIN_FAILURE,)),
+        ("litellm-providers/price-sync-fireworks-ai", "Bot", MAP_FILES, ()),
+        ("litellm_azure_ai_mai_image_2_5_pro", "Bot", (*MAP_FILES, guard.PINS_PATH), (BOT_AUTHOR_PIN_FAILURE,)),
+        ("litellm_azure_ai_mai_image_2_5_pro", "Bot", MAP_FILES, ()),
+        ("fix-openrouter-price-sync", "User", (*MAP_FILES, guard.PINS_PATH), ()),
+        ("fix-openrouter-price-sync", "", (*MAP_FILES, guard.PINS_PATH), ()),
+    ],
+)
+def test_only_human_opened_non_price_sync_prs_may_touch_the_pins_file(
+    head_ref: str, author_type: str, changed: tuple[str, ...], expected: tuple[str, ...]
+) -> None:
+    assert guard.pin_author_failures(head_ref, author_type, changed) == expected
+
+
+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)
@@ -230,13 +385,38 @@ def _git_commit(repo: Path, message: str) -> str:
     ).stdout.strip()
 
 
-def _run_guard(repo: Path, base: str, head: str, head_ref: str) -> subprocess.CompletedProcess[str]:
+def _run_guard(
+    repo: Path,
+    base: str,
+    head: str,
+    head_ref: str,
+    *,
+    target: str | None = None,
+    target_env: str | None = None,
+    author_type: str | None = None,
+) -> subprocess.CompletedProcess[str]:
+    target_args: Final = () if target_env is not None else ("--target", target or base)
+    env_overrides: Final = {
+        **({} if target_env is None else {guard.TARGET_ENV: target_env}),
+        **({} if author_type is None else {guard.AUTHOR_TYPE_ENV: author_type}),
+    }
     return subprocess.run(
-        (sys.executable, str(CI_CD / "cost_map_guard.py"), "--base", base, "--head", head, "--head-ref", head_ref),
+        (
+            sys.executable,
+            str(CI_CD / "cost_map_guard.py"),
+            "--base",
+            base,
+            "--head",
+            head,
+            *target_args,
+            "--head-ref",
+            head_ref,
+        ),
         cwd=repo,
         capture_output=True,
         text=True,
         check=False,
+        env={**os.environ, **env_overrides},
     )
 
 
@@ -298,3 +478,69 @@ 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)]
+
+
+@pytest.mark.parametrize("head_ref", [BOT_REF, "litellm_fix_pricing"])
+def test_main_holds_a_pin_the_base_branch_added_after_the_pr_branched(tmp_path: Path, head_ref: str) -> None:
+    subprocess.run(("git", "init", "-q", str(tmp_path)), check=True)
+    merge_base: Final = _commit(tmp_path, BASE_MAP, "merge base")
+    target: Final = _commit(tmp_path, BASE_MAP, "pin on the base branch", pins=PINS)
+    subprocess.run(("git", "checkout", "-q", merge_base), cwd=tmp_path, check=True)
+    head: Final = _commit(tmp_path, NO_VISION_MAP, "branch cut before the pin lowers it", pins=NO_PINS)
+    result: Final = _run_guard(tmp_path, merge_base, head, head_ref, target_env=target)
+    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_reports_a_failed_merge_apart_from_a_conflict(tmp_path: Path) -> None:
+    subprocess.run(("git", "init", "-q", str(tmp_path)), check=True)
+    base: Final = _commit(tmp_path, BASE_MAP, "base")
+    head: Final = _commit(tmp_path, {**BASE_MAP, "openrouter/c": _entry()}, "head")
+    result: Final = _run_guard(tmp_path, base, head, "litellm_fix_pricing", target="0" * 40)
+    assert result.returncode == 1, result.stdout + result.stderr
+    assert result.stdout.startswith(f"cost map guard failed: git merge-tree into {'0' * 40} failed: ")
+
+
+def test_main_tells_a_conflicting_pr_to_merge_the_base_branch_in(tmp_path: Path) -> None:
+    subprocess.run(("git", "init", "-q", str(tmp_path)), check=True)
+    merge_base: Final = _commit(tmp_path, BASE_MAP, "merge base")
+    target: Final = _commit(tmp_path, {**BASE_MAP, "openrouter/b": _entry(3e-06)}, "base branch reprices b")
+    subprocess.run(("git", "checkout", "-q", merge_base), cwd=tmp_path, check=True)
+    head: Final = _commit(tmp_path, {**BASE_MAP, "openrouter/b": _entry(4e-06)}, "pr reprices b differently")
+    result: Final = _run_guard(tmp_path, merge_base, head, "litellm_fix_pricing", target=target)
+    assert result.returncode == 1, result.stdout + result.stderr
+    conflict_line: Final = "- the pull request does not merge cleanly into its base branch; merge the base branch in first"
+    assert conflict_line in result.stdout.splitlines()
+
+
+def test_main_skips_the_merge_for_a_pr_that_leaves_the_cost_map_alone(tmp_path: Path) -> None:
+    subprocess.run(("git", "init", "-q", str(tmp_path)), check=True)
+    base: Final = _commit(tmp_path, BASE_MAP, "base")
+    head: Final = _commit_code_only(tmp_path, "code change")
+    result: Final = _run_guard(tmp_path, base, head, "litellm_fix_pricing", target="0" * 40)
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert result.stdout.splitlines() == ["cost map guard passed (human PR, cost map untouched)"]
+
+
+@pytest.mark.parametrize(("author_type", "expected_code"), [("Bot", 1), ("User", 0)])
+def test_main_lets_only_a_human_opened_pr_change_the_pins(tmp_path: Path, author_type: str, expected_code: int) -> 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, NO_VISION_MAP, "lower the flag and drop its pin", pins=NO_PINS)
+    result: Final = _run_guard(
+        tmp_path, base, head, "litellm_azure_ai_mai_image_2_5_pro", author_type=author_type
+    )
+    assert result.returncode == expected_code, result.stdout + result.stderr
+    assert (f"- {BOT_AUTHOR_PIN_FAILURE}" in result.stdout.splitlines()) == (author_type == "Bot")