feat(ci): freeze the conftest save/restore inventory so it can only shrink (#37621)

* feat(ci): freeze the conftest save/restore inventory so it can only shrink

* fix(ci): resolve the named constant a conftest save loop iterates

* fix(ci): match the snapshot shape instead of a list of blessed dict names

* feat(ci): fail a branch that clears TQ violations without lowering the ceiling

A limit that only ever falls is not the same as one that falls when it can.
Clearing violations and leaving the ceiling above the new count let the same
violations return later under a limit nobody moved, so the gate now fails on
that and names `make lint-budget-update` as the fix. It needs both head below
base and head below limit, so headroom already in the base is never blamed on
the branch that happens to run next.

Drops the seeded-rule exemption from the ratchet along with it. Its stated
reason was that the base tree predates a rule introduced on this branch, but
base counts are measured with the current checker, so such a rule is counted at
the base too and its grandfathered total was never at risk of reading as fixed.
Removing the exemption is what lets a newly seeded rule ratchet like the six
that came before it.

The base scan is skipped when the branch touches neither the test tree nor the
checker, since neither count can have moved.

.github/workflows/test-linting.yml

@@ -137,7 +137,7 @@ jobs:
         run: |
           uv run --no-sync python scripts/type_discipline_gate.py --base "$GATE_BASE_SHA"
 
-      - name: Check test-quality budget (zero-assert / mock-echo tests, sys.path.insert, raw env writes, litellm global mutation, credential-gated skips, delta vs base)
+      - name: Check test-quality budget (zero-assert / mock-echo tests, sys.path.insert, raw env writes, litellm global mutation, credential-gated skips, conftest snapshot inventory, delta vs base)
         if: steps.changes.outputs.decision != 'skip'
         run: |
           uv run --no-sync python scripts/test_quality_gate.py --base "$GATE_BASE_SHA"

Makefile

@@ -206,8 +206,8 @@ lint-type-discipline: $(LINT_DEP_INSTALL) $(LINT_DEP_BASE)
 	$(UV_RUN) python scripts/type_discipline_gate.py --base origin/litellm_internal_staging
 
 # Test-quality budget (zero-assert / mock-echo tests, sys.path.insert, raw env writes,
-# litellm module-global mutation, credential-gated skips), counted across tests/ the
-# same delta-vs-base way.
+# litellm module-global mutation, credential-gated skips, conftest snapshot
+# inventory), counted across tests/ the same delta-vs-base way.
 lint-test-quality: $(LINT_DEP_INSTALL) $(LINT_DEP_BASE)
 	$(UV_RUN) python scripts/test_quality_gate.py --base origin/litellm_internal_staging
 

scripts/check_test_quality.py

@@ -39,15 +39,23 @@ TQ005   `litellm.<attr> = ...` module-global mutation. The SDK's module globals
         what the 491-line save/restore conftest exists to paper over. Inject the
         dependency or use a fixture that restores it.
 TQ006   A `pytest.skip` reached only when a credential-shaped environment variable is
-        absent. Absence is what the condition has to say: `not key`, `key is None`,
-        `"KEY" not in os.environ`. A skip taken when the credential is present is
-        somebody's deliberate branch and is left alone. On a runner that does not hold that credential the guard fires every
+        absent. On a runner that does not hold that credential the guard fires every
         time, so the test reports green having executed nothing and is indistinguishable
         from coverage that exists. Fake the provider at the HTTP boundary, or fail
-        loudly, so a missing credential shows up as a missing credential. The gate is
-        followed through one local or module-level binding, which is the
-        `key = os.getenv(...)` then `if not key: pytest.skip(...)` shape most of these
-        use.
+        loudly, so a missing credential shows up as a missing credential. Absence is
+        what the condition has to say -- `not key`, `key is None`, `"KEY" not in
+        os.environ` -- since a skip taken when the credential is present is somebody's
+        deliberate branch. The gate follows one local or module-level binding, which is
+        the `key = os.getenv(...)` then `if not key: pytest.skip(...)` shape most of
+        these use.
+TQ007   A module global that a conftest saves before every test and restores after it.
+        The save/restore list is a hand-maintained inventory of the leaks the suite
+        already knows about, so it is allowed to shrink and never to grow: a new entry
+        means one more global whose lifetime the tests manage instead of the code owning
+        it. Give the consumers an injection seam rather than another snapshot line. The
+        names are read from the keys the conftest assigns directly and from whatever the
+        save loop iterates, including a module-level tuple or dict it names rather than
+        spells out.
 
 Every rule is suppressible with `# test-quality-ok: <reason>` on the reported
 line, following the repo's `*-ok: <reason>` convention. A suppression without a
@@ -123,6 +131,9 @@ PATCH_MEMBERS: Final = frozenset(("object", "dict", "multiple"))
 ENVIRON_READERS: Final = frozenset(("os.environ.get", "environ.get", "os.getenv", "getenv"))
 ENVIRON_MAPPINGS: Final = frozenset(("os.environ", "environ"))
 SKIP_CALLS: Final = frozenset(("pytest.skip", "skip"))
+CONFTEST_NAME: Final = "conftest.py"
+SDK_MODULE: Final = "litellm"
+
 CREDENTIAL_NAME_RE: Final = re.compile(
     r"(?:API_KEY|_KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|DATABASE_URL|ACCESS_KEY_ID)$"
 )
@@ -547,6 +558,105 @@ def iter_credential_skip_violations(path: Path, tree: ast.Module) -> Iterator[Vi
                     )
 
 
+def _reads_sdk_attribute(node: ast.AST) -> bool:
+    return any(
+        (
+            isinstance(inner, ast.Call)
+            and _dotted_name(inner.func) == "getattr"
+            and bool(inner.args)
+            and _dotted_name(inner.args[0]) == SDK_MODULE
+        )
+        or (isinstance(inner, ast.Attribute) and _dotted_name(inner.value) == SDK_MODULE)
+        for inner in ast.walk(node)
+    )
+
+
+def _subscript_targets(node: ast.AST) -> Iterator[ast.Subscript]:
+    for inner in ast.walk(node):
+        if isinstance(inner, ast.Assign):
+            yield from (target for target in inner.targets if isinstance(target, ast.Subscript))
+
+
+def _saves_sdk_attribute_by_key(node: ast.AST) -> Iterator[ast.Subscript]:
+    """Every `<dict>["name"] = <something read off litellm>`, whatever the dict is called.
+
+    Matching on the shape rather than on a list of blessed dict names is what reaches
+    the conftest that builds its snapshot inside a helper and calls the dict `state`.
+    """
+    for inner in ast.walk(node):
+        if isinstance(inner, ast.Assign) and _reads_sdk_attribute(inner.value):
+            yield from (target for target in inner.targets if isinstance(target, ast.Subscript))
+
+
+def _saves_sdk_attributes_in_loop(node: ast.For) -> bool:
+    """A save loop reads the SDK and stores under the loop variable, in either order.
+
+    The read is often bound to a local first (`val = getattr(litellm, attr)`) and only
+    then stored, so the read and the store are separate statements and cannot be
+    required of the same assignment.
+    """
+    if not isinstance(node.target, ast.Name):
+        return False
+    stores_by_key: Final = any(
+        isinstance(subscript.slice, ast.Name) and subscript.slice.id == node.target.id
+        for statement in node.body
+        for subscript in _subscript_targets(statement)
+    )
+    return stores_by_key and any(_reads_sdk_attribute(statement) for statement in node.body)
+
+
+def _module_constants(tree: ast.Module) -> Mapping[str, ast.expr]:
+    return MappingProxyType({
+        target.id: node.value
+        for node in tree.body
+        if isinstance(node, ast.Assign)
+        for target in node.targets
+        if isinstance(target, ast.Name)
+    })
+
+
+def _string_members(node: ast.expr) -> Iterator[tuple[str, int]]:
+    """The string names a collection literal holds: a tuple/list's items, a dict's keys."""
+    elements: Final = (
+        node.elts if isinstance(node, (ast.Tuple, ast.List)) else node.keys if isinstance(node, ast.Dict) else ()
+    )
+    yield from (
+        (element.value, element.lineno)
+        for element in elements
+        if isinstance(element, ast.Constant) and isinstance(element.value, str)
+    )
+
+
+def _snapshotted_names(tree: ast.Module) -> Iterator[tuple[str, int]]:
+    constants: Final = _module_constants(tree)
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Assign):
+            yield from (
+                (subscript.slice.value, subscript.lineno)
+                for subscript in _saves_sdk_attribute_by_key(node)
+                if isinstance(subscript.slice, ast.Constant) and isinstance(subscript.slice.value, str)
+            )
+        elif isinstance(node, ast.For) and _saves_sdk_attributes_in_loop(node):
+            iterable: Final = constants.get(node.iter.id) if isinstance(node.iter, ast.Name) else node.iter
+            if iterable is not None:
+                yield from _string_members(iterable)
+
+
+def iter_conftest_inventory_violations(path: Path, tree: ast.Module) -> Iterator[Violation]:
+    if path.name != CONFTEST_NAME:
+        return
+    seen: Final = dict(reversed(tuple(_snapshotted_names(tree))))
+    for name, line in sorted(seen.items(), key=lambda item: item[1]):
+        yield Violation(
+            path,
+            line,
+            "TQ007",
+            f"`litellm.{name}` is saved and restored around every test in this tree; the list is an "
+            "inventory of known leaks and may only shrink, so give the consumers an injection seam "
+            f"instead of adding to it (suppress: `# {SUPPRESSION_TOKEN}: <reason>`)",
+        )
+
+
 def check_file(path: Path) -> tuple[Violation, ...]:
     try:
         source: Final = path.read_text(encoding="utf-8")
@@ -567,6 +677,7 @@ def check_file(path: Path) -> tuple[Violation, ...]:
             *iter_environ_violations(path, tree),
             *iter_global_mutation_violations(path, tree),
             *iter_credential_skip_violations(path, tree),
+            *iter_conftest_inventory_violations(path, tree),
         )
         if violation.line not in skip
     )

scripts/test_quality_gate.py

@@ -12,9 +12,15 @@ Every rule is seeded at exactly its count on the day the gate landed, so the
 suite's existing debt is grandfathered and any net-new violation trips the gate
 immediately. ``--update`` ratchets a limit down by the violations this branch
 fixed relative to its branch point (the merge-base), so the ceilings only ever
-fall. A rule absent from the budget at the merge-base was seeded on this branch;
-``--update`` leaves its limit untouched, because the base tree predates the rule
-and its whole grandfathered count would otherwise be misread as "fixed".
+fall. Base counts are measured with the *current* checker, so a rule introduced
+on this branch is counted at the base too and ratchets like every other one.
+
+Only ever falling is not the same as always falling, so the gate enforces the
+second half: a branch that clears violations and leaves the ceiling above its
+new count fails, naming the rules and telling the author to run
+``make lint-budget-update``. Without that, a removed violation could come back
+later under a ceiling nobody lowered. Drift already in the base is never
+blamed, so this fires only on the branch that did the clearing.
 
 The deliberate difference from its sibling: this gate has no headroom anywhere.
 Type discipline seeded LIT010/LIT011 at 1.5x to leave room for an in-flight
@@ -138,6 +144,21 @@ def over_ceiling(head: Mapping[str, int], budget: Mapping[str, Mapping[str, int]
     )
 
 
+def unratcheted(
+    head: Mapping[str, int],
+    base: Mapping[str, int],
+    budget: Mapping[str, Mapping[str, int]],
+) -> tuple[Breach, ...]:
+    """Rules this branch cleared without lowering the ceiling behind them. Requires
+    both `head < base`, so drift already in the base is never blamed on this change,
+    and `head < limit`, so a ceiling already at the count is left alone."""
+    return tuple(sorted(
+        Breach(rule, head.get(rule, 0), spec["limit"], head.get(rule, 0) - base.get(rule, 0))
+        for rule, spec in budget.items()
+        if head.get(rule, 0) < base.get(rule, 0) and head.get(rule, 0) < spec["limit"]
+    ))
+
+
 def evaluate(
     head: Mapping[str, int],
     base: Mapping[str, int],
@@ -177,15 +198,39 @@ def introduced(
     return tuple(v for v in violations if v.line in changed.get(v.file, frozenset()))
 
 
+def touches_measured_tree(base_point: str) -> bool:
+    """Whether this branch changed anything that can move a count. A branch that
+    touches neither the test tree nor the checker cannot have cleared a violation,
+    so the base scan is skipped and the gate stays cheap on the common change."""
+    changed: Final = _run(
+        ["git", "diff", "--name-only", base_point, "--", TARGET, str(CHECKER.relative_to(REPO_ROOT))]
+    )
+    return bool(changed.strip())
+
+
 def cmd_check(base: str) -> None:
     budget: Final = json.loads(BUDGET_PATH.read_text())
     head: Final = head_violations()
     head_counts: Final = count_by_rule(head)
-    if not over_ceiling(head_counts, budget):
+    base_point: Final = resolve_base_point(base)
+    if not over_ceiling(head_counts, budget) and not touches_measured_tree(base_point):
         print(f"OK: every TQ rule is within its test-suite ceiling (base {base})")
         return
-    base_point: Final = resolve_base_point(base)
-    breaches: Final = evaluate(head_counts, base_counts(base_point), budget)
+    base_at_point: Final = base_counts(base_point)
+    stale: Final = unratcheted(head_counts, base_at_point, budget)
+    if stale:
+        print(f"FAIL: TQ-rule limits were left above the count this branch reached (base {base}):")
+        for breach in stale:
+            print(
+                f"  {breach.rule}: this branch cleared {-breach.added} down to {breach.total}, "
+                f"but the limit is still {breach.cap}"
+            )
+        print(
+            "Run `make lint-budget-update` and commit the lowered limits, so the "
+            "violations you cleared cannot come back under a ceiling nobody moved."
+        )
+        raise SystemExit(1)
+    breaches: Final = evaluate(head_counts, base_at_point, budget)
     if not breaches:
         print(f"OK: every TQ rule is within its test-suite ceiling (base {base})")
         return
@@ -216,46 +261,25 @@ def ratcheted_budget(
     budget: Mapping[str, Mapping[str, int]],
     current: Mapping[str, int],
     base: Mapping[str, int],
-    seeded: frozenset[str] = frozenset(),
 ) -> Mapping[str, Mapping[str, int]]:
     """Each rule's limit lowered by the violations `current` fixed vs `base`. The drop
-    is clamped to what was actually cleared, so a limit only ever falls. Rules in
-    `seeded` were introduced on this branch and pass through untouched."""
+    is clamped to what was actually cleared, so a limit only ever falls."""
     return MappingProxyType({
-        rule: {
-            "limit": spec["limit"] if rule in seeded
-            else max(0, spec["limit"] - max(0, base.get(rule, 0) - current.get(rule, 0)))
-        }
+        rule: {"limit": max(0, spec["limit"] - max(0, base.get(rule, 0) - current.get(rule, 0)))}
         for rule, spec in sorted(budget.items())
     })
 
 
-def _base_budget_rules(base_point: str) -> frozenset[str]:
-    proc: Final = subprocess.run(
-        ["git", "show", f"{base_point}:{BUDGET_PATH.name}"],
-        cwd=REPO_ROOT, capture_output=True, text=True,
-    )
-    if proc.returncode != 0:
-        return frozenset()
-    return frozenset(json.loads(proc.stdout))
-
-
 def cmd_update(base_ref: str = DEFAULT_BASE) -> None:
     """Ratchet each rule's limit down by the violations this branch fixed."""
     budget: Final = json.loads(BUDGET_PATH.read_text())
     base_point: Final = resolve_base_point(base_ref)
-    seeded: Final = frozenset(budget) - _base_budget_rules(base_point)
     updated: Final = ratcheted_budget(
-        budget, count_by_rule(head_violations()), base_counts(base_point), seeded
+        budget, count_by_rule(head_violations()), base_counts(base_point)
     )
     BUDGET_PATH.write_text(json.dumps(dict(updated), indent=2, sort_keys=True) + "\n")
     cleared: Final = sum(budget[rule]["limit"] - updated[rule]["limit"] for rule in updated)
     print(f"Ratcheted TQ-rule limits down by {cleared} violations this branch fixed")
-    if seeded:
-        print(
-            "Left untouched (seeded on this branch, absent from the base budget): "
-            + ", ".join(sorted(seeded))
-        )
 
 
 def cmd_seed() -> None:

test-quality-budget.json

@@ -16,5 +16,8 @@
   },
   "TQ006": {
     "limit": 34
+  },
+  "TQ007": {
+    "limit": 117
   }
 }

tests/test_litellm/test_check_test_quality.py

@@ -397,3 +397,154 @@ def test_a_none_comparison_reads_as_absence(tmp_path):
 def test_a_membership_test_without_the_negation_is_left_alone(tmp_path):
     source = _MEMBERSHIP_GATE.replace('"ACME_API_KEY" not in os.environ', '"ACME_API_KEY" in os.environ')
     assert _codes(tmp_path, source) == []
+
+
+_SNAPSHOT_CONFTEST = """import litellm
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def restore_globals():
+    original_state = {}
+    original_state["drop_params"] = litellm.drop_params
+    for attr in ("api_base", "num_retries"):
+        original_state[attr] = getattr(litellm, attr)
+    yield
+    for attr, value in original_state.items():
+        setattr(litellm, attr, value)
+"""
+
+
+def _conftest_codes(tmp_path, source, name="conftest.py"):
+    snippet = tmp_path / name
+    snippet.write_text(source, encoding="utf-8")
+    return [v.code for v in checker.check_file(snippet)]
+
+
+def test_every_snapshotted_global_is_counted_once(tmp_path):
+    assert _conftest_codes(tmp_path, _SNAPSHOT_CONFTEST) == ["TQ007", "TQ007", "TQ007"]
+
+
+def test_the_names_come_from_the_loop_tuple_as_well_as_the_direct_keys(tmp_path):
+    snippet = tmp_path / "conftest.py"
+    snippet.write_text(_SNAPSHOT_CONFTEST, encoding="utf-8")
+    reported = [v.message.split("`")[1] for v in checker.check_file(snippet)]
+    assert sorted(reported) == ["litellm.api_base", "litellm.drop_params", "litellm.num_retries"]
+
+
+def test_the_same_global_saved_twice_counts_once(tmp_path):
+    source = _SNAPSHOT_CONFTEST.replace(
+        '("api_base", "num_retries")', '("api_base", "num_retries", "drop_params")'
+    )
+    assert _conftest_codes(tmp_path, source) == ["TQ007", "TQ007", "TQ007"]
+
+
+def test_the_rule_only_looks_at_conftest_files(tmp_path):
+    assert _conftest_codes(tmp_path, _SNAPSHOT_CONFTEST, name="test_snapshot.py") == []
+
+
+def test_a_conftest_that_snapshots_nothing_is_clean(tmp_path):
+    source = "import pytest\n\n\n@pytest.fixture\ndef client():\n    return object()\n"
+    assert _conftest_codes(tmp_path, source) == []
+
+
+def test_a_snapshot_entry_is_suppressible_with_a_reason(tmp_path):
+    source = _SNAPSHOT_CONFTEST.replace(
+        'original_state["drop_params"] = litellm.drop_params',
+        'original_state["drop_params"] = litellm.drop_params  # test-quality-ok: owned by the SDK config surface',
+    )
+    assert _conftest_codes(tmp_path, source) == ["TQ007", "TQ007"]
+
+
+_NAMED_MAPPING_CONFTEST = """import litellm
+import pytest
+
+_SCALAR_DEFAULTS = {
+    "num_retries": None,
+    "set_verbose": False,
+}
+_EXTRA_ATTRS = ("api_base", "drop_params")
+
+
+@pytest.fixture(autouse=True)
+def restore_globals():
+    original_state = {}
+    for attr in _SCALAR_DEFAULTS:
+        original_state[attr] = getattr(litellm, attr)
+    for attr in _EXTRA_ATTRS:
+        original_state[attr] = getattr(litellm, attr)
+    yield
+    for attr, value in original_state.items():
+        setattr(litellm, attr, value)
+"""
+
+
+def test_a_save_loop_over_a_module_level_dict_counts_its_keys(tmp_path):
+    # The two largest inventories in the repo name their list instead of spelling it
+    # out, so a rule that only reads literal iterables sees neither.
+    reported = [v.message.split("`")[1] for v in checker.check_file(_written(tmp_path, _NAMED_MAPPING_CONFTEST))]
+    assert sorted(reported) == [
+        "litellm.api_base",
+        "litellm.drop_params",
+        "litellm.num_retries",
+        "litellm.set_verbose",
+    ]
+
+
+def test_a_named_iterable_that_is_not_a_module_constant_is_skipped_quietly(tmp_path):
+    source = _NAMED_MAPPING_CONFTEST.replace("for attr in _EXTRA_ATTRS:", "for attr in dir(litellm):")
+    reported = [v.message.split("`")[1] for v in checker.check_file(_written(tmp_path, source))]
+    assert sorted(reported) == ["litellm.num_retries", "litellm.set_verbose"]
+
+
+def _written(tmp_path, source, name="conftest.py"):
+    path = tmp_path / name
+    path.write_text(source, encoding="utf-8")
+    return path
+
+
+_HELPER_DICT_CONFTEST = """import litellm
+import pytest
+
+_CALLBACK_ATTRS = ("callbacks", "success_callback")
+
+
+def _copy_litellm_state():
+    state = {}
+    for attr in _CALLBACK_ATTRS:
+        if hasattr(litellm, attr):
+            value = getattr(litellm, attr)
+            state[attr] = value.copy() if isinstance(value, list) else value
+    return state
+
+
+@pytest.fixture(autouse=True)
+def restore_globals():
+    saved = _copy_litellm_state()
+    yield
+    for attr, value in saved.items():
+        setattr(litellm, attr, value)
+"""
+
+
+def test_a_snapshot_built_in_a_helper_under_any_dict_name_is_counted(tmp_path):
+    # Two conftests build their inventory inside a helper and call the dict `state`,
+    # so a rule keyed on blessed dict names sees neither.
+    reported = [v.message.split("`")[1] for v in checker.check_file(_written(tmp_path, _HELPER_DICT_CONFTEST))]
+    assert sorted(reported) == ["litellm.callbacks", "litellm.success_callback"]
+
+
+def test_the_read_may_sit_a_statement_above_the_store(tmp_path):
+    # `val = getattr(litellm, attr)` then `state[attr] = val.copy()` is the common
+    # shape; requiring the store itself to read litellm loses every one of them.
+    source = _HELPER_DICT_CONFTEST.replace(
+        "            state[attr] = value.copy() if isinstance(value, list) else value",
+        "            state[attr] = list(value)",
+    )
+    reported = [v.message.split("`")[1] for v in checker.check_file(_written(tmp_path, source))]
+    assert sorted(reported) == ["litellm.callbacks", "litellm.success_callback"]
+
+
+def test_a_loop_storing_under_a_key_that_is_not_the_loop_variable_is_not_an_inventory(tmp_path):
+    source = _HELPER_DICT_CONFTEST.replace("state[attr] =", 'state["fixed"] =')
+    assert [v.code for v in checker.check_file(_written(tmp_path, source))] == []

tests/test_litellm/test_test_quality_gate.py

@@ -1,9 +1,11 @@
 """Tests for scripts/test_quality_gate.py.
 
-The gate's whole value is that it blames a change only for what it adds, and that a
-limit can never rise. Both properties live in pure functions, so they are tested
-directly: `evaluate` for the blame rule, `ratcheted_budget` for the one-way ratchet,
-and `parse_changed_lines` for the diff scan that turns a breach into file:line.
+The gate's whole value is that it blames a change only for what it adds, that a limit
+can never rise, and that a limit cannot stay above a count the branch pushed below it.
+All three live in pure functions, so they are tested directly: `evaluate` for the blame
+rule, `ratcheted_budget` for the one-way ratchet, `unratcheted` for the ceiling a branch
+left behind, and `parse_changed_lines` for the diff scan that turns a breach into
+file:line.
 """
 
 import importlib.util
@@ -66,11 +68,32 @@ def test_ratchet_never_goes_below_zero():
     assert updated["TQ001"]["limit"] == 0
 
 
-def test_ratchet_leaves_a_rule_seeded_on_this_branch_untouched():
-    updated = gate.ratcheted_budget(
-        _BUDGET, {"TQ001": 0}, {"TQ001": 10}, seeded=frozenset({"TQ001"})
-    )
-    assert updated["TQ001"]["limit"] == 10
+def test_ratchet_lowers_a_rule_introduced_on_this_branch_like_any_other():
+    updated = gate.ratcheted_budget(_BUDGET, {"TQ001": 4}, {"TQ001": 10})
+    assert updated["TQ001"]["limit"] == 4
+
+
+def test_a_branch_that_cleared_violations_must_lower_the_ceiling():
+    stale = gate.unratcheted({"TQ001": 6}, {"TQ001": 10}, _BUDGET)
+    assert [(b.rule, b.total, b.cap, b.added) for b in stale] == [("TQ001", 6, 10, -4)]
+
+
+def test_headroom_already_in_the_base_is_not_blamed_on_this_branch():
+    assert gate.unratcheted({"TQ001": 6}, {"TQ001": 6}, _BUDGET) == ()
+
+
+def test_a_branch_that_cleared_down_to_the_ceiling_exactly_is_clean():
+    assert gate.unratcheted({"TQ001": 10}, {"TQ001": 12}, _BUDGET) == ()
+
+
+def test_a_branch_that_added_violations_is_not_a_ratchet_finding():
+    assert gate.unratcheted({"TQ001": 14}, {"TQ001": 10}, _BUDGET) == ()
+
+
+def test_the_ratchet_finding_survives_the_update_that_answers_it():
+    cleared = {"TQ001": 6}
+    updated = gate.ratcheted_budget(_BUDGET, cleared, {"TQ001": 10})
+    assert gate.unratcheted(cleared, {"TQ001": 10}, updated) == ()
 
 
 def test_parse_changed_lines_groups_hunks_under_their_own_file():
@@ -121,5 +144,5 @@ def test_the_shipped_budget_covers_every_rule_the_checker_can_emit():
     import json
 
     budget = json.loads((_REPO_ROOT / "test-quality-budget.json").read_text())
-    assert set(budget) == {"TQ001", "TQ002", "TQ003", "TQ004", "TQ005", "TQ006"}
+    assert set(budget) == {"TQ001", "TQ002", "TQ003", "TQ004", "TQ005", "TQ006", "TQ007"}
     assert all(spec["limit"] >= 0 for spec in budget.values())