diff --git a/reviewbot/config.py b/reviewbot/config.py
index 31dae7f..d542e83 100644
--- a/reviewbot/config.py
+++ b/reviewbot/config.py
@@ -390,6 +390,12 @@ class Config:
     # feedback loop (validate once, accept whatever the model produced). The
     # model gets up to ``task_normalize_max_retries + 1`` patch attempts.
     task_normalize_max_retries: int = 2
+    # Reject (and send back for correction) a patch whose test-expectation
+    # change does not have the shape maintainers accept: only the CI runner's
+    # Expectations key may change, no tolerance may loosen
+    # (reviewbot/expectation_keys.py). Rides the same retry budget as the
+    # normalizer; a violation that survives it is flagged in the PR body.
+    task_expectation_key_check: bool = True
     # Optional free-text guidance injected into the task system prompt and the
     # normalize-failure feedback, alongside the normalize command. Use it to
     # encode policy the command itself can't express — e.g. "prefer root-cause
@@ -735,6 +741,7 @@ def from_env(
             ).strip()
             or None,
             task_normalize_max_retries=_int_env("TASK_NORMALIZE_MAX_RETRIES", 2),
+            task_expectation_key_check=_bool_env("TASK_EXPECTATION_KEY_CHECK", True),
             task_preflight_command=(
                 shlex.split(os.environ.get("TASK_PREFLIGHT_COMMAND") or "") or None
             ),
diff --git a/reviewbot/expectation_keys.py b/reviewbot/expectation_keys.py
new file mode 100644
index 0000000..79cf16d
--- /dev/null
+++ b/reviewbot/expectation_keys.py
@@ -0,0 +1,281 @@
+"""Check that a test-expectation change has the shape maintainers accept.
+
+transformers keeps hardware-specific expected values in ``Expectations({key:
+value})`` blocks, keyed ``(device, compute_capability)``. The way maintainers
+re-record a value that drifted on the CI runner is narrow, and the evidence is
+in their merges: on 2026-10-11, of the serge PRs opened in September and October,
+3 of 3 value updates that touched only the runner's ``("cuda", 8)`` entry merged
+cleanly (#48515, #49068, #49076). The one that overwrote the broad ``("cuda",
+None)`` key merged with a wrong value (#48580). The rest of the open queue
+overwrote ``(None, None)``, rewrote XPU values with no XPU evidence, or edited a
+plain literal in place.
+
+The rule, as this module enforces it:
+
+* only the CI runner's key may change or be added: ``("cuda", 8)``, or
+  ``("cuda", (8, 6))`` where the block keys by ``(major, minor)``
+  (:data:`RUNNER_KEYS`; the runners are NVIDIA A10G, compute capability 8.6);
+* every other key keeps its value. The one move allowed is *demotion*: an
+  existing value may move to ``(None, None)`` unchanged, which is how #48198 did
+  it;
+* a new ``Expectations`` block (converting a plain literal) may hold only
+  ``(None, None)`` and a runner key;
+* in an expectation-only patch, nothing outside an ``Expectations`` block may
+  change except the glue a conversion needs (the import, the
+  ``get_expectation()`` call). That catches loosened tolerances and plain
+  literals overwritten in place.
+
+This only reads the two versions of a test file. It cannot tell whether a new
+value is *right* (see :mod:`reviewbot.expectation_guard` for degenerate values);
+it checks that the change is shaped so a reviewer can see what the new hardware
+changed and the old value still holds where it held.
+"""
+
+from __future__ import annotations
+
+import ast
+import difflib
+import re
+from dataclasses import dataclass
+
+__all__ = ["RUNNER_KEYS", "expectation_key_problems", "RULE_TEXT"]
+
+RUNNER_KEYS = frozenset({("cuda", 8), ("cuda", (8, 6))})
+GENERIC_KEY = (None, None)
+
+RULE_TEXT = (
+    'The CI runner is an NVIDIA A10G: its key is `("cuda", 8)`, or '
+    '`("cuda", (8, 6))` where the block already keys by (major, minor). Change or '
+    "add ONLY that key. Every other key keeps its value: never overwrite "
+    '`(None, None)`, `("cuda", None)`, another CUDA generation, `"xpu"` or '
+    '`"rocm"`. You may move an existing value, unchanged, to `(None, None)`. If '
+    "the expected value is a plain literal, convert it to `Expectations({...})` "
+    "with the old value under `(None, None)` and the new one under the runner "
+    "key. Do not change tolerances or anything else outside the `Expectations` "
+    "block."
+)
+
+# A runner-key value may move by this fraction of the block's largest
+# magnitude. Accepted drift is far below it: #48515 moved 0.4% of its scale,
+# #49068 (bf16 logits near zero) 3.6%. The rejected pvt groups moved 11% and 45%.
+MAX_RUNNER_DRIFT = 0.05
+
+_NUMBER = re.compile(r"-?\d+\.\d+(?:[eE][-+]?\d+)?|-?\d+(?:[eE][-+]?\d+)")
+_TOLERANCE = re.compile(r"\b(atol|rtol)\s*=\s*([0-9.]+(?:[eE][-+]?\d+)?)")
+# Changed lines outside a block that read as an expected value or a check: an
+# assertion, an EXPECTED_*/expected_* name, a tolerance, or a bare literal line
+# (a multi-line tensor or string continuation). Anything else is test code, and
+# a test-code fix is not this module's business (#47697 renamed a config key).
+_EXPECTED_LINE = re.compile(
+    r"expect|assert|\batol\b|\brtol\b|^\s*[\[\]\"'\d-]", re.IGNORECASE
+)
+# Lines a literal → Expectations conversion has to touch outside the block.
+_GLUE = re.compile(
+    r"get_expectation\(|^\s*Expectations,?\s*$|import\b.*\bExpectations\b"
+)
+
+
+@dataclass
+class _Block:
+    scope: tuple[str, int]  # (enclosing function, ordinal within it)
+    entries: dict  # key tuple -> normalized value source
+    first: int  # 1-based line span, inclusive
+    last: int
+
+
+def _key(node: ast.expr):
+    try:
+        key = ast.literal_eval(node)
+    except (ValueError, SyntaxError, TypeError):
+        return None
+    return key if isinstance(key, tuple) and len(key) == 2 else None
+
+
+def _is_expectations(call: ast.Call) -> bool:
+    func = call.func
+    name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
+    return (
+        name == "Expectations"
+        and bool(call.args)
+        and isinstance(call.args[0], ast.Dict)
+    )
+
+
+def _blocks(src: str) -> list[_Block] | None:
+    try:
+        tree = ast.parse(src)
+    except SyntaxError:
+        return None
+    out: list[_Block] = []
+    counts: dict[str, int] = {}
+
+    def visit(node: ast.AST, scope: str) -> None:
+        for child in ast.iter_child_nodes(node):
+            if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                visit(child, child.name)
+                continue
+            if isinstance(child, ast.Call) and _is_expectations(child):
+                d = child.args[0]
+                entries = {}
+                for k, v in zip(d.keys, d.values):
+                    key = _key(k) if k is not None else None
+                    if key is not None:
+                        entries[key] = ast.unparse(v)
+                ordinal = counts.get(scope, 0)
+                counts[scope] = ordinal + 1
+                out.append(
+                    _Block(
+                        (scope, ordinal),
+                        entries,
+                        child.lineno,
+                        child.end_lineno or child.lineno,
+                    )
+                )
+            visit(child, scope)
+
+    visit(tree, "<module>")
+    return out
+
+
+def _numbers(value_src: str) -> list[float]:
+    return [float(x) for x in _NUMBER.findall(value_src)]
+
+
+def _drift(old_src: str, new_src: str) -> float | None:
+    """Largest change between two same-shaped numeric values, as a fraction of
+    the old value's largest magnitude; ``None`` for text or a shape change."""
+    old, new = _numbers(old_src), _numbers(new_src)
+    if not old or len(old) != len(new):
+        return None
+    scale = max(abs(x) for x in old) or 1.0
+    return max(abs(a - b) for a, b in zip(old, new)) / scale
+
+
+def _loosened(old_lines: list[str], new_lines: list[str]) -> list[str]:
+    def tolerances(lines):
+        out: dict[str, float] = {}
+        for ln in lines:
+            for name, value in _TOLERANCE.findall(ln):
+                try:
+                    out[name] = max(out.get(name, 0.0), float(value))
+                except ValueError:
+                    pass
+        return out
+
+    old, new = tolerances(old_lines), tolerances(new_lines)
+    return [
+        f"{n} {old[n]:g} -> {new[n]:g}"
+        for n in sorted(new)
+        if n in old and new[n] > old[n]
+    ]
+
+
+def _fmt(key) -> str:
+    return repr(key).replace("'", '"')
+
+
+def _within(line: int, blocks: list[_Block]) -> bool:
+    return any(b.first <= line <= b.last for b in blocks)
+
+
+def expectation_key_problems(
+    path: str, old_src: str, new_src: str, *, expectation_only: bool = False
+) -> list[str]:
+    """What is wrong with how one test file's expectations were changed, as
+    one sentence per problem; ``[]`` when the change has the accepted shape.
+
+    ``expectation_only`` turns on the "nothing outside an Expectations block"
+    rule, which only makes sense for a patch that changes literals alone: a real
+    fix may well edit test code."""
+    old_blocks, new_blocks = _blocks(old_src), _blocks(new_src)
+    if old_blocks is None or new_blocks is None:
+        return []
+    problems: list[str] = []
+    old_by_scope = {b.scope: b for b in old_blocks}
+
+    for new in new_blocks:
+        old = old_by_scope.get(new.scope)
+        where = f"`{path}` (line {new.first})"
+        if old is None:
+            extra = [
+                k for k in new.entries if k != GENERIC_KEY and k not in RUNNER_KEYS
+            ]
+            if extra:
+                problems.append(
+                    f"{where}: a new Expectations block may only hold `(None, None)` "
+                    f"and the runner key, not {', '.join(_fmt(k) for k in extra)}."
+                )
+            continue
+        old_values = set(old.entries.values())
+        for key, value in old.entries.items():
+            if key in RUNNER_KEYS or new.entries.get(key) == value:
+                continue
+            demoted = (
+                key not in new.entries
+                and GENERIC_KEY not in old.entries
+                and new.entries.get(GENERIC_KEY) == value
+            )
+            if demoted:
+                continue
+            verb = "changed" if key in new.entries else "removed"
+            problems.append(
+                f"{where}: {verb} the {_fmt(key)} value, which is not the CI runner's key."
+            )
+        for key in RUNNER_KEYS & new.entries.keys():
+            before = old.entries.get(key) or new.entries.get(GENERIC_KEY)
+            drift = _drift(before, new.entries[key]) if before else None
+            if drift is not None and drift > MAX_RUNNER_DRIFT:
+                problems.append(
+                    f"{where}: the {_fmt(key)} values move by {drift:.0%} of the "
+                    f"block's scale; hardware drift is about 1%, so this looks like "
+                    "a regression, not a new baseline."
+                )
+        for key, value in new.entries.items():
+            if key in old.entries or key in RUNNER_KEYS:
+                continue
+            if key == GENERIC_KEY and value in old_values:
+                continue  # a demotion; checked from the old side above
+            what = (
+                "a `(None, None)` value that is not an existing one"
+                if key == GENERIC_KEY
+                else f"a {_fmt(key)} entry"
+            )
+            problems.append(f"{where}: added {what}; only the runner key may be added.")
+
+    old_lines, new_lines = old_src.splitlines(), new_src.splitlines()
+    matcher = difflib.SequenceMatcher(a=old_lines, b=new_lines, autojunk=False)
+    for tag, i1, i2, j1, j2 in matcher.get_opcodes():
+        if tag != "equal":
+            loosened = _loosened(old_lines[i1:i2], new_lines[j1:j2])
+            if loosened:
+                problems.append(
+                    f"`{path}` (line {j1 + 1}): loosened a tolerance "
+                    f"({', '.join(loosened)})."
+                )
+    if expectation_only:
+        for tag, i1, i2, j1, j2 in matcher.get_opcodes():
+            if tag == "equal":
+                continue
+            in_block = any(_within(i + 1, old_blocks) for i in range(i1, i2)) or any(
+                _within(j + 1, new_blocks) for j in range(j1, j2)
+            )
+            changed = [
+                ln
+                for ln in new_lines[j1:j2] + old_lines[i1:i2]
+                if ln.strip() and not ln.lstrip().startswith("#")
+            ]
+            glue = changed and all(
+                _GLUE.search(ln) for ln in new_lines[j1:j2] if ln.strip()
+            )
+            if in_block or glue or not changed:
+                continue
+            if _loosened(old_lines[i1:i2], new_lines[j1:j2]):
+                continue  # already reported above
+            if not any(_EXPECTED_LINE.search(ln) for ln in changed):
+                continue
+            sample = (new_lines[j1:j2] or old_lines[i1:i2])[0].strip()
+            problems.append(
+                f"`{path}` (line {j1 + 1}): changed `{sample[:80]}` outside any "
+                "Expectations block."
+            )
+    return problems
diff --git a/reviewbot/tasks.py b/reviewbot/tasks.py
index 345b600..1d50079 100644
--- a/reviewbot/tasks.py
+++ b/reviewbot/tasks.py
@@ -34,6 +34,8 @@
 from .compression import MessageCompressor
 from .config import Config
 from .expectation_guard import PatchClassification, classify_patch
+from .expectation_keys import RULE_TEXT as _EXPECTATION_RULE_TEXT
+from .expectation_keys import expectation_key_problems
 from .github_client import SERGE_GIT_EMAIL, GitHubClient
 from .guidance import GuidanceResult, check_patch
 from .llm_client import ChatCompletionClient
@@ -251,6 +253,10 @@ class TaskPlan:
     # rather than recomputed at publish time because the call that produced it
     # is billed to this task's metrics.
     guidance_note: str = ""
+    # Expectation-shape problems (reviewbot/expectation_keys.py) still present in
+    # the final patch after the correction budget ran out. Empty when the patch
+    # has the accepted shape or touches no expectations.
+    expectation_problems: list[str] = field(default_factory=list)
 
 
 @dataclass
@@ -901,6 +907,7 @@ def _validate_patch(
         # Cleared per attempt: a patch synthesized from an accepted answer must
         # never outlive the answer it came from.
         report["patch"] = ""
+        report["expectation_problems"] = []
 
     try:
         result = _extract_json(content, _TASK_JSON_KEYS)
@@ -996,6 +1003,29 @@ def _validate_patch(
             )
             return msg, False
 
+    problems = _expectation_shape_problems(cfg, checkout, patch)
+    if report is not None:
+        report["expectation_problems"] = problems
+    if problems:
+        clone_cache.reset_worktree(checkout)
+        emit("rejected_patch", _bounded_rejected_patch(patch))
+        emit(
+            "log",
+            "Expectation change rejected: " + " ".join(problems),
+        )
+        if report is not None:
+            report["rejection"] = "expectations"
+        return (
+            "Your patch changes test expectations in a way transformers "
+            "maintainers do not accept:\n\n"
+            + "\n".join(f"- {p}" for p in problems)
+            + f"\n\n{_EXPECTATION_RULE_TEXT}\n\nReturn the corrected change. "
+            "If the new values cannot be expressed that way (they moved far more "
+            "than hardware drift, or the fix belongs in the code), return an "
+            "empty `edits` list and say why in `body`.",
+            False,
+        )
+
     # The patch is applied and nothing has reformatted it yet, so the file's
     # line numbers are still the diff's new-side ones: the one point where the
     # comments the patch added can be located exactly. Shorten them here so the
@@ -1064,6 +1094,40 @@ def _validate_patch(
     return None, True
 
 
+def _expectation_shape_problems(
+    cfg: Config, checkout: Checkout, patch: str
+) -> list[str]:
+    """Shape problems in how the applied ``patch`` changes test expectations.
+
+    Compares each changed test file in the worktree with its committed version
+    (:mod:`reviewbot.expectation_keys`). Best-effort: a file git cannot show (new,
+    or outside the repo) is skipped, never an error."""
+    if not cfg.task_expectation_key_check:
+        return []
+    classification = classify_patch(patch)
+    problems: list[str] = []
+    for path in classification.test_files:
+        full = os.path.join(checkout.path, path)
+        try:
+            with open(full, encoding="utf-8") as fh:
+                new_src = fh.read()
+            old_src = subprocess.run(
+                ["git", "-C", checkout.path, "show", f"HEAD:{path}"],
+                capture_output=True,
+                text=True,
+                check=True,
+            ).stdout
+        except (OSError, subprocess.CalledProcessError):
+            continue
+        problems += expectation_key_problems(
+            path,
+            old_src,
+            new_src,
+            expectation_only=classification.expectation_only,
+        )
+    return problems
+
+
 _HUNK_HEADER_RE = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+\d+(?:,\d+)? @@")
 _PATCH_TARGET_RE = re.compile(r"^\+\+\+ b/(.+)$")
 # Lines of real file content to show around a rejected hunk. Wide enough that
@@ -1928,6 +1992,7 @@ def _validate(chat) -> Optional[str]:
         session=session_record(metrics),
         model=llm.model,
         worktree_prepared=outcome["prepared"],
+        expectation_problems=list(outcome.get("expectation_problems") or []),
         guidance_note=guidance.pr_section() if guidance is not None else "",
     )
 
@@ -2082,6 +2147,16 @@ def _decorate_body(
     # the serge footer (not tacked on after them).
     if verification_footer:
         body += f"\n{verification_footer}"
+    # A shape problem the correction budget could not fix
+    # (reviewbot/expectation_keys.py). The PR still opens, because a human may
+    # judge the value fine, but the reviewer is told before reading the diff.
+    if plan.expectation_problems:
+        body += (
+            "\n\n### ⚠️ Expectation change not in the maintainer shape\n"
+            'Only the CI runner\'s `("cuda", 8)` entry should change; serge could '
+            "not get the patch into that shape:\n"
+            + "\n".join(f"- {p}" for p in plan.expectation_problems)
+        )
     # §3.4, directly under the verification it qualifies: both sections are
     # evidence about this change, and this one is the weaker of the two — an
     # LLM judgement about review prose, not a test result — so it reads after
diff --git a/tests/test_expectation_keys.py b/tests/test_expectation_keys.py
new file mode 100644
index 0000000..2498949
--- /dev/null
+++ b/tests/test_expectation_keys.py
@@ -0,0 +1,178 @@
+"""Tests for the expectation-shape rule (reviewbot/expectation_keys.py).
+
+Each case is cut down from a real serge PR on huggingface/transformers; the
+number in the test name is the PR."""
+
+import textwrap
+import unittest
+
+from reviewbot.expectation_keys import expectation_key_problems
+
+
+def _src(body: str) -> str:
+    return textwrap.dedent(body).lstrip()
+
+
+BASE = _src(
+    """
+    def test_logits(self):
+        expectations = Expectations(
+            {
+                (None, None): [0.9989, 0.1953, -0.4382],
+                ("xpu", 5): [0.9986, 0.1953, -0.4381],
+            }
+        )
+        torch.testing.assert_close(out, expected, rtol=1e-4, atol=1e-4)
+    """
+)
+
+
+def problems(old: str, new: str, eo: bool = True) -> list[str]:
+    return expectation_key_problems("test_x.py", old, new, expectation_only=eo)
+
+
+class AcceptedShapes(unittest.TestCase):
+    def test_48838_adds_the_runner_key(self):
+        new = BASE.replace(
+            '("xpu", 5)',
+            '("cuda", 8): [0.9987, 0.1953, -0.4381],\n                ("xpu", 5)',
+        )
+        self.assertEqual(problems(BASE, new), [])
+
+    def test_49040_adds_a_major_minor_runner_key(self):
+        new = BASE.replace(
+            '("xpu", 5)',
+            '("cuda", (8, 6)): [0.9987, 0.1953, -0.4381],\n                ("xpu", 5)',
+        )
+        self.assertEqual(problems(BASE, new), [])
+
+    def test_48882_demotes_then_updates_the_runner_key(self):
+        old = _src(
+            """
+            def test_bf16(self):
+                e = Expectations({("cuda", 8): [-0.3145, -5.6562], ("xpu", 3): [-0.31, -5.65]})
+            """
+        )
+        new = _src(
+            """
+            def test_bf16(self):
+                e = Expectations({
+                    (None, None): [-0.3145, -5.6562],
+                    ("cuda", 8): [-0.3184, -5.6875],
+                    ("xpu", 3): [-0.31, -5.65],
+                })
+            """
+        )
+        self.assertEqual(problems(old, new), [])
+
+    def test_48198_demotes_cuda_none_to_generic(self):
+        old = _src(
+            """
+            def test_t(self):
+                e = Expectations({("cuda", None): [-6.7329, -16.9590]})
+            """
+        )
+        new = _src(
+            """
+            def test_t(self):
+                e = Expectations({(None, None): [-6.7329, -16.9590], ("cuda", 8): [-6.7667, -16.9915]})
+            """
+        )
+        self.assertEqual(problems(old, new), [])
+
+    def test_converting_a_plain_literal(self):
+        old = _src(
+            """
+            from transformers.testing_utils import require_torch
+
+            def test_t(self):
+                expected_slice = torch.tensor([-0.2962, 0.4487])
+            """
+        )
+        new = _src(
+            """
+            from transformers.testing_utils import Expectations, require_torch
+
+            def test_t(self):
+                expectations = Expectations({(None, None): [-0.2962, 0.4487], ("cuda", 8): [-0.2970, 0.4490]})
+                expected_slice = torch.tensor(expectations.get_expectation())
+            """
+        )
+        self.assertEqual(problems(old, new), [])
+
+    def test_47697_test_code_fix_is_not_this_rules_business(self):
+        old = 'def test_t(self):\n    config.rope_parameters["type"] = "default"\n'
+        new = 'def test_t(self):\n    config.rope_parameters["rope_type"] = "default"\n'
+        self.assertEqual(problems(old, new), [])
+
+
+class RejectedShapes(unittest.TestCase):
+    def test_47616_overwrites_the_generic_key(self):
+        new = BASE.replace("(None, None): [0.9989", "(None, None): [0.9950")
+        self.assertIn("(None, None)", problems(BASE, new)[0])
+
+    def test_48580_overwrites_cuda_none(self):
+        old = 'def test_t(self):\n    e = Expectations({("cuda", None): ["a Pystylus cat"]})\n'
+        new = 'def test_t(self):\n    e = Expectations({("cuda", None): ["a snow leopard cat"]})\n'
+        self.assertIn('("cuda", None)', problems(old, new)[0])
+
+    def test_48955_rewrites_another_vendor(self):
+        new = BASE.replace('("xpu", 5): [0.9986', '("xpu", 5): [0.9000')
+        self.assertIn('"xpu"', problems(BASE, new)[0])
+
+    def test_49407_adds_a_broad_key(self):
+        new = BASE.replace(
+            '("xpu", 5)',
+            '("cuda", None): [1.0, 0.2, -0.4],\n                ("xpu", 5)',
+        )
+        out = problems(BASE, new)
+        self.assertEqual(len(out), 1)
+        self.assertIn('added a ("cuda", None) entry', out[0])
+
+    def test_new_generic_value_that_is_not_a_demotion(self):
+        old = 'def test_t(self):\n    e = Expectations({("xpu", 3): [1.0]})\n'
+        new = 'def test_t(self):\n    e = Expectations({(None, None): [2.0], ("xpu", 3): [1.0]})\n'
+        self.assertIn("(None, None)", problems(old, new)[0])
+
+    def test_47750_runner_key_moving_far_past_drift(self):
+        old = 'def test_t(self):\n    e = Expectations({(None, None): [-1.4192, -1.9158], ("cuda", 8): [-1.4194, -1.9161]})\n'
+        new = 'def test_t(self):\n    e = Expectations({(None, None): [-1.4192, -1.9158], ("cuda", 8): [-1.0951, -1.9902]})\n'
+        out = problems(old, new)
+        self.assertEqual(len(out), 1)
+        self.assertIn("regression", out[0])
+
+    def test_49068_small_bf16_drift_is_fine(self):
+        old = 'def test_t(self):\n    e = Expectations({("cuda", (8, 6)): [0.0608, -0.0933, 0.1348]})\n'
+        new = 'def test_t(self):\n    e = Expectations({("cuda", (8, 6)): [0.0588, -0.0928, 0.1357]})\n'
+        self.assertEqual(problems(old, new), [])
+
+    def test_49143_loosened_tolerance_even_with_code_changes(self):
+        new = BASE.replace("rtol=1e-4, atol=1e-4", "rtol=1e-2, atol=1e-2")
+        out = problems(BASE, new, eo=False)
+        self.assertEqual(len(out), 1)
+        self.assertIn("atol 0.0001 -> 0.01", out[0])
+
+    def test_47840_plain_literal_overwritten_in_place(self):
+        old = (
+            "def test_t(self):\n    expected_slice = torch.tensor([-0.2962, 0.4487])\n"
+        )
+        new = (
+            "def test_t(self):\n    expected_slice = torch.tensor([-0.2782, 0.4466])\n"
+        )
+        self.assertIn("outside any Expectations block", problems(old, new)[0])
+
+    def test_plain_literal_rule_only_for_expectation_only_patches(self):
+        old = (
+            "def test_t(self):\n    expected_slice = torch.tensor([-0.2962, 0.4487])\n"
+        )
+        new = (
+            "def test_t(self):\n    expected_slice = torch.tensor([-0.2782, 0.4466])\n"
+        )
+        self.assertEqual(problems(old, new, eo=False), [])
+
+    def test_unparseable_source_is_not_judged(self):
+        self.assertEqual(problems("def broken(:\n", "def broken(:\n  x\n"), [])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_tasks.py b/tests/test_tasks.py
index 469a02b..fed161c 100644
--- a/tests/test_tasks.py
+++ b/tests/test_tasks.py
@@ -2350,3 +2350,83 @@ def test_the_checker_env_asks_for_patch_scoping(self):
         # Without this the checker's `diff_only` is False and it scans the
         # whole tree however well-shaped the worktree is.
         self.assertEqual(tasks_module._CHECKER_ENV["CI_PULL_REQUEST"], "1")
+
+
+class ExpectationShapeValidationTests(unittest.TestCase):
+    """A patch that overwrites a non-runner Expectations key is sent back for
+    correction, like a normalizer failure (reviewbot/expectation_keys.py)."""
+
+    _TEST = "tests/models/x/test_modeling_x.py"
+    _OLD = (
+        "def test_logits(self):\n"
+        "    e = Expectations({\n"
+        "        (None, None): [0.9989, 0.1953],\n"
+        "    })\n"
+    )
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        root = self._tmp.name
+        src = os.path.join(root, "src")
+        os.makedirs(os.path.join(src, "tests/models/x"))
+        with open(os.path.join(src, self._TEST), "w") as f:
+            f.write(self._OLD)
+        _git(src, "init", "--quiet", "-b", "main")
+        _git(src, "add", "-A")
+        _git(src, "commit", "--quiet", "-m", "main commit")
+        self.cache = CloneCache(os.path.join(root, "cache"))
+        self.co = self.cache.acquire_ref(
+            token="",
+            owner="acme",
+            repo="widget",
+            ref="main",
+            job_id="keys1234",
+            remote_url=src,
+        )
+
+    def _validate(self, old_line, new_lines, **cfg):
+        edits = [{"path": self._TEST, "old": old_line, "new": new_lines}]
+        report: dict = {}
+        feedback, _ = _validate_patch(
+            _make_cfg(
+                helper_sandbox="off",
+                task_sandbox_backend="bwrap",
+                task_normalize_command=["true"],
+                task_normalize_timeout=30,
+                **cfg,
+            ),
+            checkout=self.co,
+            clone_cache=self.cache,
+            content=json.dumps({"title": "t", "body": "b", "edits": edits}),
+            emit=lambda *a: None,
+            report=report,
+        )
+        return feedback, report
+
+    def test_overwriting_the_generic_key_is_sent_back(self):
+        feedback, report = self._validate(
+            "(None, None): [0.9989, 0.1953],", "(None, None): [0.9950, 0.1953],"
+        )
+        self.assertIn("maintainers do not accept", feedback)
+        self.assertIn('("cuda", 8)', feedback)
+        self.assertEqual(report["rejection"], "expectations")
+        self.assertTrue(report["expectation_problems"])
+        with open(os.path.join(self.co.path, self._TEST)) as fh:
+            self.assertEqual(fh.read(), self._OLD)  # worktree reset
+
+    def test_adding_the_runner_key_is_accepted(self):
+        feedback, report = self._validate(
+            "(None, None): [0.9989, 0.1953],",
+            '(None, None): [0.9989, 0.1953],\n        ("cuda", 8): [0.9987, 0.1953],',
+        )
+        self.assertIsNone(feedback)
+        self.assertEqual(report["expectation_problems"], [])
+
+    def test_kill_switch(self):
+        feedback, _ = self._validate(
+            "(None, None): [0.9989, 0.1953],",
+            "(None, None): [0.9950, 0.1953],",
+            task_expectation_key_check=False,
+        )
+        self.assertIsNone(feedback)