bench: refuse to report a comparison whose two sides have different calibers

Two runs on different calibers produce a report that looks exactly like a
valid one — no step errors, the numbers just mean something else. This repo
has published three figures that were an artifact of such a difference:
router_gemm -8.2% (record vs record), mm "945 variants -> 33" (shared vs
inherited libtuner cache), replication_pad3d +43% vs +124.5% (timer).

batch_summary --ab / --ab-rounds now reads every participating run's
run_identity.json and compares backend/device/timer/bench_level/shape_mode/
tuner_cache/direct_repeat before parsing a single latency. Mismatch, or a
missing identity, prints !!! [caliber] and exits 4. autotune_mode is judged
separately: record pinning a replay is the normal shape, both sides record
is the exact trap above and is refused by name. --allow-caliber-mismatch
releases it, still printing what it found.

Recording on a *shared* tuner cache warns rather than refuses: the pinned
config depends on what else ran on the machine, which leaves relative
conclusions intact and absolute latencies unquotable. Refusing it would
reject the default every A/B here currently uses.

ab_repeat.sh closes the one axis that had no check: after round 1 it
fingerprints both sides' ttgir (relative path + content, so a changed
variant *set* counts too). Byte-identical means the switch changed no
generated code for this op, and the remaining N-1 rounds would measure
noise and report it as "no difference" — so the op is skipped with
SWITCH_NO_IR_DIFF unless SWITCH_ALLOW_NO_IR_DIFF=1.

README leads with a caliber table: each axis, its measured magnitude, what
pins it, and which code checks it. The fourth column is code because
documentation did not hold — the README already said "the B side must use
REPLAY_FROM" when the record-vs-record number was published.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

test_backends.py

@@ -431,5 +431,115 @@ class AutotuneReplayBackendTests(unittest.TestCase):
             self.assertFalse(_replay_backend_ok(tmp))
 
 
+class CaliberGuardTests(unittest.TestCase):
+    """两次运行口径不同产出的报告，和有效报告长得一模一样。
+
+    这些用例钉住的是本仓已经踩过的三次：record 对 record、共享 tuner cache、
+    timer 不同。它们都不会让任何一步报错，只会让数字变成另一个含义。
+    """
+
+    BASE = {
+        "backend": "musa", "device": "S5000", "timer": "cuda_graph",
+        "bench_level": "core", "shape_mode": "default",
+        "tuner_cache": "isolated", "direct_repeat": 5,
+        "autotune_mode": "replay",
+    }
+
+    def _run(self, root, name, **over):
+        d = Path(root, name)
+        d.mkdir(parents=True)
+        if over.get("_no_identity"):
+            return d
+        ident = dict(self.BASE)
+        ident.update({k: v for k, v in over.items() if not k.startswith("_")})
+        Path(d, "run_identity.json").write_text(
+            json.dumps(ident), encoding="utf-8")
+        return d
+
+    def test_same_caliber_has_no_complaints(self):
+        from batch_summary import caliber_complaints
+        with tempfile.TemporaryDirectory() as tmp:
+            runs = [("A", self._run(tmp, "a")), ("B", self._run(tmp, "b"))]
+            self.assertEqual(caliber_complaints(runs), [])
+
+    def test_record_side_pinning_the_other_is_normal(self):
+        """A/B 的正常形态：一侧 record 定 config，另一侧 replay 它。"""
+        from batch_summary import caliber_complaints
+        with tempfile.TemporaryDirectory() as tmp:
+            runs = [("r1_A", self._run(tmp, "a", autotune_mode="record")),
+                    ("r1_B", self._run(tmp, "b"))]
+            self.assertEqual(caliber_complaints(runs), [])
+
+    def test_record_against_record_is_flagged(self):
+        from batch_summary import caliber_complaints
+        with tempfile.TemporaryDirectory() as tmp:
+            runs = [("A", self._run(tmp, "a", autotune_mode="record")),
+                    ("B", self._run(tmp, "b", autotune_mode="record"))]
+            self.assertTrue(any("record" in c for c in caliber_complaints(runs)))
+
+    def test_each_caliber_field_is_checked(self):
+        from batch_summary import CALIBER_FIELDS, caliber_complaints
+        others = {"backend": "cuda", "device": "H20", "timer": "direct",
+                  "bench_level": "comprehensive", "shape_mode": "yaml",
+                  "tuner_cache": "shared", "direct_repeat": 1}
+        self.assertEqual(set(others), set(CALIBER_FIELDS))
+        for field, other in others.items():
+            with tempfile.TemporaryDirectory() as tmp:
+                runs = [("A", self._run(tmp, "a")),
+                        ("B", self._run(tmp, "b", **{field: other}))]
+                got = caliber_complaints(runs)
+                self.assertTrue(any(c.startswith(field) for c in got),
+                                f"{field} 不一致时没被拦下: {got}")
+
+    def test_missing_identity_is_flagged_not_assumed_equal(self):
+        from batch_summary import caliber_complaints
+        with tempfile.TemporaryDirectory() as tmp:
+            runs = [("A", self._run(tmp, "a")),
+                    ("B", self._run(tmp, "b", _no_identity=True))]
+            self.assertTrue(any("run_identity" in c
+                                for c in caliber_complaints(runs)))
+
+    def test_shared_cache_record_warns_but_does_not_block(self):
+        from batch_summary import caliber_complaints, caliber_hazards
+        with tempfile.TemporaryDirectory() as tmp:
+            runs = [("r1_A", self._run(tmp, "a", autotune_mode="record",
+                                       tuner_cache="shared(~/.flaggems)")),
+                    ("r1_B", self._run(tmp, "b",
+                                       tuner_cache="shared(~/.flaggems)"))]
+            self.assertEqual(caliber_complaints(runs), [])
+            self.assertTrue(caliber_hazards(runs))
+
+    def test_isolated_cache_has_no_hazard(self):
+        from batch_summary import caliber_hazards
+        with tempfile.TemporaryDirectory() as tmp:
+            runs = [("A", self._run(tmp, "a", autotune_mode="record")),
+                    ("B", self._run(tmp, "b"))]
+            self.assertEqual(caliber_hazards(runs), [])
+
+    def test_strict_refuses_and_permissive_continues(self):
+        from batch_summary import _report_caliber
+        with tempfile.TemporaryDirectory() as tmp:
+            runs = [("A", self._run(tmp, "a")),
+                    ("B", self._run(tmp, "b", timer="direct"))]
+            self.assertFalse(_report_caliber(runs, "op", strict=True))
+            self.assertTrue(_report_caliber(runs, "op", strict=False))
+
+    LOG = ("Operator: op  (dtype=torch.float16, mode=kernel, level=core)\n"
+           "SUCCESS 1.000 2.000 0.500 shape\n")
+
+    def test_report_ab_refuses_mismatched_pair(self):
+        from batch_summary import CaliberMismatch, report_ab
+        with tempfile.TemporaryDirectory() as tmp:
+            a = self._run(tmp, "a")
+            b = self._run(tmp, "b", timer="direct")
+            for d in (a, b):
+                Path(d, "run.log").write_text(self.LOG, encoding="utf-8")
+            with self.assertRaises(CaliberMismatch):
+                report_ab(a / "run.log", b / "run.log", "op")
+            # 放行开关在，但必须是显式的
+            self.assertEqual(
+                len(report_ab(a / "run.log", b / "run.log", "op", strict=False)), 1)
+
+
 if __name__ == "__main__":
     unittest.main()