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>

README.md

@@ -6,6 +6,30 @@
 2. **可解释**：自动按 shape 收集每次运行实际使用的 ttgir（带可读的变体命名），供 IR 级 diff；
 3. **口径统一**：cudagraph 计时消除 launch 开销，小 kernel 的对比不被 CPU 侧噪声淹没；capture 前按 warmup 预算显式预热吸收 autotune/JIT，首轮即稳态、多轮一致。
 
+## 先读这个：口径决定数字的含义
+
+同一个算子、同一台机器、同一份代码，换一条测试口径，测出来的数字可以完全不同。
+下表每一行都是本仓踩过的，量级是实测的，不是估计：
+
+| 口径轴 | 改一下的量级 | 由谁固定 | 由谁检查 |
+|---|---|---|---|
+| autotune config（record vs replay） | router_gemm 差 8pp | `REPLAY_FROM` | `caliber_complaints`：两侧都 record 直接拒绝出报告 |
+| timer（cuda_graph vs direct） | replication_pad3d 从 +43% 变成 +124.5% | `ZL_TIMER` / backend 能力 | 同上，`timer` 字段不一致即拒绝 |
+| libtuner 持久缓存（共享 vs 隔离） | mm 看起来从 945 个变体降到 33 个——实际是 B 侧继承了 A 侧的缓存 | `ZL_ISOLATE_TUNER_CACHE=1` | 共享缓存上 record 会打 `警告 [caliber]` |
+| `--level core` vs `comprehensive` | mm 的 shape 数 5 → 29，能不能覆盖到大 GEMM 全看这个 | `BENCH_LEVEL` | `bench_level` 字段不一致即拒绝 |
+| 重复轮数 | 收益落在零附近时，单次测量给不出方向：topk 42 个 shape 有 17 个在两次等效运行间翻了符号 | `ROUNDS` / `ZL_DIRECT_REPEAT` | `direct_repeat` 字段不一致即拒绝 |
+| 被测开关本身是否真的接上 | 开关名拼错，两侧跑同一份代码，脚本照样报出"无差异" | 开关名正确 | `ab_repeat.sh`：第 1 轮后比两侧 ttgir 指纹，相同即跳过该算子 |
+
+**结论：一次运行的数字没有意义，只有一对同口径运行的差值有意义。** 每次运行把自己的
+口径写进 `runs/<run>/run_identity.json`；`batch_summary.py --ab` / `--ab-rounds` 在出
+报告前核对这些字段，不一致就拒绝，因为**口径不同的两次运行会产出一份和有效报告长得
+一模一样的表**——没有任何一步会报错，只是数字换了含义。确认两次运行确实可比时，用
+`--allow-caliber-mismatch` 显式放行。
+
+文档拦不住人：README 里早就写着"B 侧必须走 REPLAY_FROM"和"只有 record 数据支撑的
+结论按未验证处理"，本仓仍然发布过一个 record 对 record 得到的 router_gemm 数字。
+所以上表第四列是代码，不是提醒。
+
 单算子精测流程之上另有批量筛查层（`run_batch.py`，见「批量多算子测试」），可对双仓库上千个算子做例行体检，产物结构与单算子完全同构。
 
 ## 快速开始
@@ -67,6 +91,7 @@ softmax:
 | `USE_FLAGTUNE` | `0` | `0` 走普通 autotune（跳过搜索、快速验证）；`1` 用 FlagTune 扩展调优空间（首跑全量搜索、慢）。A/B 两侧须取同值 |
 | `BENCH_LEVEL` | `core` | 透传上游 `--level`。`comprehensive` 相对 `core` 做两件事：① 在 `set_shapes()` 里并入子类 `set_more_shapes()`（`mm` 因此从 5 个 shape 变 29 个，含 N/K 到 152064 的大 K 与 `models_shapes/*.yaml` 的模型 shape）；② 在 `get_input_iter()` 里对每个 shape 再产出一次变体（BLAS 是转置侧），用例数翻倍。注意 ① 只在 `SHAPE_MODE=upstream` 下可见——传了 `SHAPE_FILE` 时 `_shape_inject_plugin` 会在上游合并之后再覆盖一次、把 shape 集拉回 yaml 内容（BLAS 那类 arity 不一致的情形除外，插件会退让）；② 不经插件，任何 shape 来源下都生效。sweep 成本同步放大，shape 多时配合 `PARALLEL_WARMUP_GPUS`。A/B 两侧须取同值 |
 | `ZL_ISOLATE_TUNER_CACHE` | `0` | FlagGems 的 libtuner winner 存在 `$FLAGGEMS_CACHE_DIR/config_cache` 的 sqlite 里（默认 `~/.flaggems`），跨 run 长存。`1` 给本次 run 一个私有缓存目录，两侧各自冷调优（README「各自最优」口径），跑完随 Triton 缓存一起删——选中的 config 已在 `autotune_records/` 里，DB 本身无留存价值。共享缓存的风险是单向的：先跑的一侧写入 winner，后跑的一侧直接命中、只编译那几个胜出变体。实测 `router_gemm` 共享模式下第二次运行编译变体从 234 掉到 10、耗时 52s→17s，隔离模式下两次都是 234 个变体。走 `REPLAY_FROM` 的标准 A/B 不受影响（config 本就锁死），record 对 record 才会中招。取值无论如何都写进 `run_identity.json` 的 `tuner_cache` |
+| `ZL_DIRECT_REPEAT` | `1` | direct 计时的重复次数，取中位。只作用于 direct 路径：整体 `timer=direct` 的运行，以及图捕获运行里逐行回退的那些点（标记会带上 `repeat=N`）；捕获成功的行不受影响。`do_bench` 自己已在调用内取中位，压住的是迭代间抖动；跨调用重复补的是它看不见的那层——时钟爬坡、邻居进程进来这类漂移。调用方要 `quantiles` 或 `return_mode="all"` 时返回序列、跨调用中位无定义，自动不重复。在 MThreads 上这不是可选项：cvt 消除量最大的 `sort`/`dropout`/`rand` 恰好都无法捕获，那批结论正是由这条路径产出的。取值写进 `run_identity.json` 的 `direct_repeat` |
 | `DTYPES` | 空（上游默认 dtype 扫描） | 空格分隔的 dtype 白名单（如 `"bfloat16 float16"`），逐个转为上游 `--dtypes`。算子不支持时上游报 `can't be supported by this op` |
 | `PARALLEL_WARMUP_GPUS` | 空（串行） | 设为 `N`（≥2）时把 autotune sweep 分片到 N 卡并行，测量仍单卡串行；不设或 `<2` 则完全串行（见「加速：多卡并行 sweep」） |
 | `ZL_ENABLE_CUDAGRAPH` | `auto` | `auto`/`1` 仅在 CUDA 上走 Graph 计时；非 CUDA 即使设 `1` 也拒绝并保持 `timer=direct` |
@@ -88,7 +113,9 @@ softmax:
 ```
 runs/softmax_20260716-031752/
 ├── run.log                  # 完整日志（含 SUCCESS 行的 latency/speedup 表；已去 ANSI 色的纯文本）
-├── run_identity.json        # provider/backend/device/timer/vendor（插件合并写入）
+├── run_identity.json        # 本次运行的口径：provider/backend/device/timer/vendor/
+│                          # bench_level/shape_mode/tuner_cache/direct_repeat/autotune_mode
+│                          # （shell 写 stub，插件合并写入；对比工具据此判可比性）
 ├── shapes.yaml              # 本次实际使用的 shape（存档）
 ├── autotune_records/        # Triton autotune 选中的 config（供 replay）
 │   ├── softmax.json
@@ -161,6 +188,65 @@ benchmark 目录，注册 `--level`/`--mode` 的 `conftest.py` 就没被加载
 只校验节点的文件那一半；`::test_<fn>` 交给 pytest，函数不存在时它自己会清楚地报
 `not found: ...`。
 
+### 口径核对：出报告前的拒绝
+
+`batch_summary.py --ab` / `--ab-rounds` 在解析任何 latency 之前，先读参与对比的每个
+run 目录的 `run_identity.json`，逐字段核对 `backend / device / timer / bench_level /
+shape_mode / tuner_cache / direct_repeat`。任何一项不一致，或任何一侧缺 identity，
+就打 `!!! [caliber] ...` 并以 **4** 退出，不出报告。
+
+```
+!!! [caliber] sort: timer 不一致：before=cuda_graph, after=direct
+!!! [caliber] sort: 拒绝出报告。确认口径确实可比后加 --allow-caliber-mismatch
+```
+
+单独判 `autotune_mode`：一侧 record 定 config、另一侧 replay 它是正常形态；**两侧都
+record 直接拒绝**——那正是本仓发布过错误 router_gemm 数字的那次形态。
+
+不拒绝但会警告的一类：在**共享** libtuner 缓存上 record。被 pin 的 config 取决于此前
+在这台机器上跑过什么，相对结论仍成立，绝对延迟与变体数不可跨机器引用。
+
+`--allow-caliber-mismatch` 放行，仍会把问题打出来。用它之前先能说清为什么这两次运行
+可比；跨口径引用别人的数字，比自己重跑一遍贵得多。
+
+### 多轮重复：ab_repeat.sh 与 --ab-rounds
+
+单对观测无法说明自己稳不稳。S5000 实测 `topk` 消除了 32 个 `convert_layout`，
+42 个 shape 里却有 17 个在两次同等有效的运行之间**改变了 delta 的符号**——那不是
+计时口径问题（图捕获与 direct 的延迟比是 1.00x），是收益本身落在零附近、单次
+测量给不出方向。
+
+```bash
+SWITCH=FLAGTREE_MUSA_RLC_ENHANCE OPS="topk masked_select" ROUNDS=6   ZL_BACKEND=musa FLAGGEMS_DIR=/path/to/FlagGems bash ab_repeat.sh
+```
+
+口径：第 1 轮 A 侧 record config，此后每一次运行（含 A 侧的后续轮次）都
+`REPLAY_FROM` 那一份，config 因此跨轮跨侧全部锁死；逐轮交替先后顺序抵消单向漂移。
+有轮次失败的算子不出报告——缺轮的报告会让人误以为是完整的 N 轮。
+
+第 1 轮 A/B 跑完后，脚本比两侧 `ttgir/` 的指纹（相对路径 + 内容）。**逐字节相同就
+跳过这个算子**，并打 `!!! SWITCH_NO_IR_DIFF op=... switch=...`：这个开关没改变该算子的
+生成代码，剩下 N-1 轮测的全是噪声，最后还会报成一个"无差异"的结论。先查开关名拼写
+与它在被测源码里的读取点；确属只影响运行时、不影响 IR 的开关，用
+`SWITCH_ALLOW_NO_IR_DIFF=1` 放行。
+
+报告分开给两种离散度，它们回答不同的问题：
+
+| 指标 | 含义 |
+|---|---|
+| 逐轮区间 | N 轮各自 delta 的 min~max，说明**判定**稳不稳 |
+| 单侧噪声 | 同一侧 N 次延迟的极差，说明**机器**吵不吵 |
+| 符号不稳行数 | delta 在轮次间改变符号的 shape 数，这些行的方向不可结论 |
+
+已有产物可以单独重出报告，目录形态是 `<root>/r<k>_<A|B>/run.log`：
+
+```bash
+python batch_summary.py --ab-rounds runs/ab_repeat_XXX/topk --label topk
+```
+
+NV 上验证过这条路径：`router_gemm` 六轮把逐轮区间收窄到 -16.6% ~ -16.2%、单侧
+噪声 0.4%，把单对观测升级成了结论。
+
 ### ab_fold_test.sh / ab_rlc_test.sh：A/B 测试参考示例
 
 `ab_fold_test.sh` 是上述 A/B 流程的现成参考：以 FlagGems `FLAGGEMS_MXFP4_FOLDSCALE`（fold_scale 优化）开关为对比对象，对 4 个 DeepSeek-V4-Flash 真实 trace shape 集（`shape/DeepSeek-V4-Flash-p{1024,4096,32768,65536}d1024.yaml`，随仓库提供）各跑一对 FOLD=0（基线，record）→ FOLD=1（replay 同 config），共 8 轮：

_cudagraph_plugin.py

@@ -33,6 +33,47 @@ import triton.testing as _tt
 _ORIGINAL_DO_BENCH = _tt.do_bench
 
 
+def direct_repeat() -> int:
+    """ZL_DIRECT_REPEAT, clamped to >= 1. 1 keeps today's single call."""
+    try:
+        return max(1, int(os.environ.get("ZL_DIRECT_REPEAT", "1")))
+    except ValueError:
+        return 1
+
+
+def run_direct(fn, warmup, rep, grad_to_none, quantiles, return_mode):
+    """Upstream do_bench, optionally repeated with the median taken.
+
+    Direct timing walks the full Python launch path once per iteration, so on
+    small kernels it carries launch overhead and CPU jitter that graph capture
+    does not. do_bench already returns a median over the iterations inside one
+    call, which suppresses iteration-level jitter but not drift across a call:
+    clocks ramping, another process arriving. Repeating the call and taking the
+    median over calls is the part the inner median cannot see.
+
+    It matters most where it is least optional. On MThreads the operators with
+    the largest RLC convert reductions -- sort, dropout, rand -- are exactly the
+    ones that cannot be graph-captured, so the headline numbers there are
+    produced by this path.
+
+    Repetition is skipped when the caller asks for quantiles or return_mode
+    "all", because the result is then a sequence and a median across calls is
+    not defined for it; the single call is returned unchanged.
+    """
+    n = direct_repeat()
+    if n == 1 or quantiles is not None or return_mode == "all":
+        return _ORIGINAL_DO_BENCH(
+            fn, warmup=warmup, rep=rep, grad_to_none=grad_to_none,
+            quantiles=quantiles, return_mode=return_mode)
+    import statistics
+    samples = [
+        _ORIGINAL_DO_BENCH(fn, warmup=warmup, rep=rep, grad_to_none=grad_to_none,
+                           quantiles=None, return_mode=return_mode)
+        for _ in range(n)
+    ]
+    return statistics.median(samples)
+
+
 def _device_interface():
     """Triton's active device interface: torch.cuda on NVIDIA, torch_musa on MThreads."""
     from triton.runtime import driver
@@ -194,16 +235,14 @@ def _patched_do_bench(fn, warmup=25, rep=100, grad_to_none=None,
 
     def _direct(note):
         from _term_style import YELLOW, paint
-        print(paint(f"BENCHMARK_DIRECT_NO_CUDAGRAPH op={op} {note}", YELLOW),
+        n = direct_repeat()
+        # The repeat count belongs in the marker: these rows are already a
+        # different caliber from the captured ones, and now they can also be a
+        # different number of samples.
+        rpt = "" if n == 1 else f" repeat={n}"
+        print(paint(f"BENCHMARK_DIRECT_NO_CUDAGRAPH op={op} {note}{rpt}", YELLOW),
               flush=True)
-        return _ORIGINAL_DO_BENCH(
+        return run_direct(fn, warmup, rep, grad_to_none, quantiles, return_mode)
-            fn,
-            warmup=warmup,
-            rep=rep,
-            grad_to_none=grad_to_none,
-            quantiles=quantiles,
-            return_mode=return_mode,
-        )
 
     backend = None
     try:

_timing_plugin.py

@@ -37,11 +37,29 @@ def pytest_configure(config):
     if _graph_requested(backend):
         import _cudagraph_plugin as _cg
         _cg.pytest_configure(config)
-        update_identity(timer="cuda_graph", graph_capture="attempted")
+        # Recorded on this branch too: the fallback rows a captured run produces
+        # are timed by the direct path and take the same repeat count.
+        update_identity(timer="cuda_graph", graph_capture="attempted",
+                        direct_repeat=_cg.direct_repeat())
         print(f"{tag('[timing-plugin]')} timer=cuda_graph backend={backend.name}",
               file=sys.stderr, flush=True)
         return
-    update_identity(timer="direct", graph_capture="disabled")
+    # A wholly-direct run gets the same repetition the fallback rows get, so the
+    # "put everything on one timer" advice does not silently cost the samples.
+    # Patching only when repeat > 1 keeps the default path untouched: upstream's
+    # do_bench stays in place exactly as before.
+    import _cudagraph_plugin as _cg
+    repeat = _cg.direct_repeat()
+    if repeat > 1:
+        def _repeated(fn, warmup=25, rep=100, grad_to_none=None,
+                      quantiles=None, return_mode="mean"):
+            return _cg.run_direct(fn, warmup, rep, grad_to_none,
+                                  quantiles, return_mode)
+        import triton.testing as _tt
+        _tt.do_bench = _repeated
+    update_identity(timer="direct", graph_capture="disabled",
+                    direct_repeat=repeat)
     print(f"{tag('[timing-plugin]')} timer=direct backend={backend.name} "
-          "(upstream do_bench; graph capture not selected)",
+          f"(upstream do_bench; graph capture not selected)"
+          + (f"; median of {repeat} calls" if repeat > 1 else ""),
           file=sys.stderr, flush=True)

ab_repeat.sh

@@ -0,0 +1,129 @@
+#!/bin/bash
+# 多轮 A/B：同一对比在 N 轮独立进程里各跑一次，逐 shape 取轮次中位。
+#
+# 为什么需要它：单对观测无法说明自己稳不稳。S5000 实测 topk 消除了 32 个
+# convert_layout，42 个 shape 里却有 17 个在两次同等有效的运行之间改变了 delta
+# 的符号——那不是计时口径的问题（图捕获与 direct 的延迟比是 1.00x），是收益本身
+# 落在零附近、单次测量给不出方向。多轮之后才谈得上"这个判定成立"。
+#
+# 口径：第 1 轮 A 侧 record autotune config，此后每一次运行（含 A 侧自己的后续
+# 轮次）都 REPLAY_FROM 那一份。config 因此跨轮跨侧全部锁死，剩下的变量只有被测
+# 开关与测量噪声。逐轮交替先后顺序，抵消单向漂移。
+#
+# 用法：
+#   SWITCH=FLAGTREE_MUSA_RLC_ENHANCE OPS="topk masked_select" ROUNDS=6 \
+#     ZL_BACKEND=musa FLAGGEMS_DIR=/path/to/FlagGems bash ab_repeat.sh
+#
+#   OPS 元素为 OP 或 OP:OP_FILE（省略时 OP_FILE=OP）。
+#   SWITCH 是被测环境变量，A 侧设 0、B 侧设 1。
+#
+# 对比轴有效性由脚本自己查：第 1 轮 A/B 跑完后比两侧的 ttgir 指纹，完全一致说明
+# 这个开关在这个算子上没改变任何生成代码——继续跑满 N 轮只会得到一个"无差异"的假
+# 结论（同 ab_fold_test.sh 的告诫）。这类算子默认跳过；开关确实只影响运行时而不影响
+# IR 时，用 SWITCH_ALLOW_NO_IR_DIFF=1 放行。
+#
+# 产物：runs/ab_repeat_<开关>_<时间戳>/<op>/r<k>_<A|B>/，跑完对每个算子调用
+# batch_summary.py --ab-rounds 出报告。
+set -uo pipefail
+
+cd "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+SWITCH_ALLOW_NO_IR_DIFF="${SWITCH_ALLOW_NO_IR_DIFF:-0}"
+
+# 一个 run 生成代码的指纹：ttgir 相对路径 + 内容。路径进指纹是因为"变体集合变了"
+# 与"某个变体内容变了"同样算改变，而只 cat 内容会把前者洗掉。
+ir_fingerprint() {
+  local dir="$1/ttgir"
+  [[ -d "$dir" ]] || return 1
+  local out
+  out=$( { find "$dir" -type f | LC_ALL=C sort | while IFS= read -r f; do
+             printf '%s
+' "${f#"$dir"/}"; cat "$f"
+           done; } | sha256sum | cut -d' ' -f1 )
+  [[ -n "$out" ]] && printf '%s' "$out"
+}
+
+: "${SWITCH:?SWITCH is required, e.g. FLAGTREE_MUSA_RLC_ENHANCE}"
+OPS="${OPS:-}"
+: "${OPS:?OPS is required, e.g. \"topk masked_select\"}"
+ROUNDS="${ROUNDS:-6}"
+
+if [[ -t 1 && -z "${NO_COLOR:-}" ]]; then
+  export FLAGGEMS_PERF_COLOR=always
+fi
+
+ROOT="runs/ab_repeat_${SWITCH}_$(date +%Y%m%d-%H%M%S)"
+mkdir -p "$ROOT"
+LOG="$ROOT/summary.log"
+
+{
+  echo ">>> switch=$SWITCH  rounds=$ROUNDS  ops=$OPS"
+  for entry in $OPS; do
+    op="${entry%%:*}"
+    op_file="${entry#*:}"
+    [[ "$op_file" == "$entry" ]] && op_file="$op"
+    out="$ROOT/$op"
+    mkdir -p "$out"
+
+    # 第 1 轮 A 侧 record，其余全部回放它，config 跨轮跨侧一致
+    echo "########## $op | round 1 | A ($SWITCH=0, record) ##########"
+    if ! OP="$op" OP_FILE="$op_file" OUT_DIR="$out/r1_A" \
+         env "$SWITCH=0" bash run_pytest.sh; then
+      echo "!!! FAILED $op round 1 A; skipping this operator"
+      continue
+    fi
+    REC="$out/r1_A"
+
+    ok=1; reason="有失败轮次"
+    for ((k = 1; k <= ROUNDS; k++)); do
+      # 奇数轮先 A 后 B，偶数轮反过来
+      if (( k % 2 == 1 )); then order="A B"; else order="B A"; fi
+      for side in $order; do
+        [[ $k -eq 1 && $side == A ]] && continue   # 已经跑过
+        val=$([[ $side == A ]] && echo 0 || echo 1)
+        echo "########## $op | round $k | $side ($SWITCH=$val, replay) ##########"
+        if ! OP="$op" OP_FILE="$op_file" OUT_DIR="$out/r${k}_${side}" \
+             REPLAY_FROM="$REC" env "$SWITCH=$val" bash run_pytest.sh; then
+          echo "!!! FAILED $op round $k $side"
+          ok=0
+        fi
+      done
+
+      if (( k == 1 && ok )); then
+        fa=$(ir_fingerprint "$out/r1_A") || fa=""
+        fb=$(ir_fingerprint "$out/r1_B") || fb=""
+        if [[ -z "$fa" || -z "$fb" ]]; then
+          echo "??? $op: 没有 ttgir 产物，跳过对比轴有效性检查"
+        elif [[ "$fa" == "$fb" ]]; then
+          echo "!!! SWITCH_NO_IR_DIFF op=$op switch=$SWITCH"
+          echo "!!! 两侧 ttgir 逐字节相同：这个开关没有改变 $op 的生成代码。"
+          echo "!!! 继续跑 $ROUNDS 轮只会测出噪声，并报成一个'无差异'结论。"
+          echo "!!! 先确认开关名拼写与它在被测源码里的读取点；确属只影响运行时的"
+          echo "!!! 开关，用 SWITCH_ALLOW_NO_IR_DIFF=1 重跑。"
+          if [[ "$SWITCH_ALLOW_NO_IR_DIFF" != "1" ]]; then
+            ok=0; reason="开关未改变该算子的生成代码"
+            break
+          fi
+        else
+          echo ">>> 对比轴有效：$op 两侧 ttgir 不同（A=${fa:0:12} B=${fb:0:12}）"
+        fi
+      fi
+    done
+
+    echo "########## $op | report ##########"
+    if (( ok )); then
+      python batch_summary.py --ab-rounds "$out" --label "$op" \
+        || echo "!!! report failed: $op"
+    else
+      # 缺轮的算子照样出报告会让人误以为是完整的 N 轮
+      echo "!!! $op $reason，不出报告；产物在 $out"
+    fi
+  done
+  echo "########## ALL DONE ##########"
+} 2>&1 | tee "$LOG"
+
+sed -i -E $'s/\x1b\\[[0-9;]*[A-Za-z]//g' "$LOG"
+
+echo ">>> 总 log: $(pwd)/$LOG"
+echo ">>> 逐算子重出报告: python batch_summary.py --ab-rounds $ROOT/<op> --label <op>"
+echo ">>> 解读要点: 逐轮区间宽说明判定不稳；单侧噪声大说明机器不稳；两者含义不同"

batch_summary.py

@@ -16,8 +16,10 @@ from __future__ import annotations
 
 import argparse
 import csv
+import itertools
 import os
 import re
+import statistics
 import sys
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple
@@ -220,8 +222,16 @@ def ab_rows(before_text: str, after_text: str) -> List[Dict[str, str]]:
     return paired
 
 
-def report_ab(before: Path, after: Path, label: str = "") -> List[Dict[str, str]]:
+def report_ab(before: Path, after: Path, label: str = "",
-    """打印一个算子的前后对比：逐行 speedup 与整体区间。"""
+              strict: bool = True) -> List[Dict[str, str]]:
+    """打印一个算子的前后对比：逐行 speedup 与整体区间。
+
+    先核对两侧口径；不一致时默认拒绝出报告，因为不可比的两次运行会产出
+    一份和有效报告长得一模一样的表。
+    """
+    if not _report_caliber([("before", before.parent), ("after", after.parent)],
+                           label or "ab", strict):
+        raise CaliberMismatch()
     before_text = before.read_text(encoding="utf-8", errors="replace")
     after_text = after.read_text(encoding="utf-8", errors="replace")
     rows = ab_rows(before_text, after_text)
@@ -244,6 +254,225 @@ def report_ab(before: Path, after: Path, label: str = "") -> List[Dict[str, str]
     return rows
 
 
+# Fields whose disagreement makes two runs incomparable. Measured magnitudes,
+# so the list is not precautionary: an unpinned autotune config moved
+# router_gemm by 8pp; the timer moved replication_pad3d from +43% to +124.5%;
+# a shared tuner cache made mm look like 945 compiled variants had become 33;
+# --level core sees 5 mm shapes where comprehensive sees 29.
+CALIBER_FIELDS = (
+    "backend", "device", "timer", "bench_level", "shape_mode",
+    "tuner_cache", "direct_repeat",
+)
+
+
+def caliber_of(run_dir: Path) -> Optional[Dict[str, str]]:
+    """Caliber fields from a run's run_identity.json, or None if unreadable."""
+    if not (run_dir / "run_identity.json").is_file():
+        return None
+    try:
+        from _run_identity import load_identity
+        ident = load_identity(str(run_dir))
+    except Exception:
+        return None
+    return {k: str(ident.get(k, "")) for k in CALIBER_FIELDS} | {
+        "autotune_mode": str(ident.get("autotune_mode", ""))
+    }
+
+
+def caliber_complaints(runs: List[Tuple[str, Path]]) -> List[str]:
+    """Everything about this set of runs that makes their comparison unsound.
+
+    Reported rather than silently tolerated because the failure is invisible in
+    the numbers: two runs on different calibers produce a report that looks
+    exactly like a valid one. This repository has already published three
+    figures that were an artifact of one such difference.
+    """
+    out: List[str] = []
+    cals = [(name, caliber_of(d)) for name, d in runs]
+    missing = [n for n, c in cals if c is None]
+    if missing:
+        out.append(f"口径不可核实：{', '.join(missing)} 缺 run_identity.json"
+                   "（早于该字段的产物）")
+    known = [(n, c) for n, c in cals if c is not None]
+    if len(known) < 2:
+        return out
+
+    for field in CALIBER_FIELDS:
+        seen = {c[field] for _, c in known}
+        if len(seen) > 1:
+            vals = ", ".join(f"{n}={c[field] or '<空>'}" for n, c in known)
+            out.append(f"{field} 不一致：{vals}")
+
+    # A/B needs one side to pin the configs the other replays. Two record runs
+    # each sweep independently, and the winner can differ; the README puts that
+    # at the 10% level and one-directional per shape, which is enough to swamp
+    # what is being measured.
+    modes = [(n, c["autotune_mode"]) for n, c in known]
+    if all(m.startswith("record") for _, m in modes):
+        out.append("两侧都是 record 模式：config 未锁定，各自 sweep 可能选出不同 "
+                   "winner。按 README，只有 record 数据支撑的结论按未验证处理")
+    return out
+
+
+def caliber_hazards(runs: List[Tuple[str, Path]]) -> List[str]:
+    """Not mismatches — shared state that makes a matched caliber less solid.
+
+    Warned about rather than refused, because refusing would reject the
+    default configuration every A/B in this repo currently uses.
+    """
+    cals = [(n, c) for n, c in ((n, caliber_of(d)) for n, d in runs) if c]
+    recording = [n for n, c in cals if c["autotune_mode"].startswith("record")]
+    shared = [n for n, c in cals if c["tuner_cache"].startswith("shared")]
+    if recording and shared:
+        # 一次 record 跑在共享 libtuner cache 上，sweep 的起点是别的运行留下的
+        # 结果。曾把 mm 的 945 个变体读成 33 个，误报为编译期收益。A/B 两侧都
+        # replay 同一份 config 时结论仍成立，绝对值不成立。
+        return [f"{', '.join(recording)} 在共享 tuner cache 上 record："
+                "被 pin 的 config 取决于此前跑过什么。相对结论不受影响，"
+                "绝对延迟/变体数不可跨机器引用（ZL_ISOLATE_TUNER_CACHE=1 可消除）"]
+    return []
+
+
+class CaliberMismatch(SystemExit):
+    """Strict-mode refusal. Exit 4 so a driver script can tell it from a crash."""
+
+    def __init__(self) -> None:
+        super().__init__(4)
+
+
+def _report_caliber(runs: List[Tuple[str, Path]], label: str, strict: bool) -> bool:
+    """Print caliber problems. Returns False when strict mode should stop."""
+    for h in caliber_hazards(runs):
+        print(f"警告 [caliber] {label}: {h}")
+    problems = caliber_complaints(runs)
+    if not problems:
+        return True
+    head = "!!!" if strict else "警告"
+    for p in problems:
+        print(f"{head} [caliber] {label}: {p}")
+    if strict:
+        print(f"!!! [caliber] {label}: 拒绝出报告。确认口径确实可比后加 "
+              "--allow-caliber-mismatch")
+        return False
+    return True
+
+
+def _lat_index(text: str) -> Dict[Tuple[str, str, int], float]:
+    """(dtype, size_detail, occurrence) -> gems latency, SUCCESS rows only.
+
+    Occurrence disambiguates a shape a benchmark measures twice in one table,
+    which BLAS does at --level comprehensive.
+    """
+    out: Dict[Tuple[str, str, int], float] = {}
+    seen: Dict[Tuple[str, str], int] = {}
+    for row in parse_run_log(text):
+        if row.get("row_status") != "SUCCESS":
+            continue
+        try:
+            lat = float(row.get("gems_lat", ""))
+        except ValueError:
+            continue
+        if lat <= 0:
+            continue
+        base = (row.get("dtype", ""), row.get("size_detail", ""))
+        i = seen.get(base, 0)
+        seen[base] = i + 1
+        out[base + (i,)] = lat
+    return out
+
+
+def ab_round_rows(root: Path) -> Tuple[List[Dict[str, str]], List[float], int]:
+    """Aggregate an N-round A/B laid out as <root>/r<k>_<A|B>/run.log.
+
+    Returns the per-shape rows, the per-round deltas, and the round count.
+
+    A single pair cannot say whether its own verdict is stable, which is the
+    question that matters once a delta lands near zero: measured on an S5000,
+    topk eliminated 32 converts yet 17 of its 42 shapes changed the SIGN of
+    their delta between two equally valid runs. Two dispersions answer different
+    halves of that and both are reported:
+
+      side spread  max/min-1 within one side's samples -> how noisy the machine
+                   was while measuring
+      round spread min~max of the per-round deltas -> whether the VERDICT holds,
+                   which is what a single pair cannot show
+    """
+    rounds: List[Tuple[Dict, Dict]] = []
+    for k in itertools.count(1):
+        a, b = root / f"r{k}_A" / "run.log", root / f"r{k}_B" / "run.log"
+        if not (a.is_file() and b.is_file()):
+            break
+        rounds.append((
+            _lat_index(a.read_text(encoding="utf-8", errors="replace")),
+            _lat_index(b.read_text(encoding="utf-8", errors="replace")),
+        ))
+    if not rounds:
+        return [], [], 0
+
+    keys = set(rounds[0][0]) & set(rounds[0][1])
+    for a, b in rounds[1:]:
+        keys &= set(a) & set(b)
+
+    per_round = [
+        statistics.median((a[k] / b[k] - 1.0) * 100.0 for k in keys)
+        for a, b in rounds
+    ] if keys else []
+
+    rows: List[Dict[str, str]] = []
+    for key in sorted(keys):
+        av = [a[key] for a, _ in rounds]
+        bv = [b[key] for _, b in rounds]
+        am, bm = statistics.median(av), statistics.median(bv)
+        deltas = [(a / b - 1.0) * 100.0 for a, b in zip(av, bv)]
+        rows.append({
+            "dtype": key[0], "size_detail": key[1],
+            "before_lat": f"{am:.6f}", "after_lat": f"{bm:.6f}",
+            "speedup_percent": f"{(am / bm - 1.0) * 100.0:.2f}",
+            "delta_min": f"{min(deltas):.2f}", "delta_max": f"{max(deltas):.2f}",
+            "sign_stable": "yes" if all(d > 0 for d in deltas)
+                           or all(d <= 0 for d in deltas) else "NO",
+            "before_spread": f"{(max(av) / min(av) - 1.0) * 100.0:.2f}",
+            "after_spread": f"{(max(bv) / min(bv) - 1.0) * 100.0:.2f}",
+        })
+    return rows, per_round, len(rounds)
+
+
+def report_ab_rounds(root: Path, label: str = "",
+                     strict: bool = True) -> List[Dict[str, str]]:
+    """Print the multi-round A/B: per-shape medians plus both dispersions."""
+    name = label or root.name
+    cells = sorted(d for d in root.glob("r*_[AB]") if (d / "run.log").is_file())
+    if cells and not _report_caliber([(d.name, d) for d in cells], name, strict):
+        raise CaliberMismatch()
+    rows, per_round, n = ab_round_rows(root)
+    if n < 2:
+        print(f"[ab-rounds] {name}: 需要至少 2 轮 <root>/r<k>_<A|B>/run.log"
+              f"（找到 {n} 轮）")
+        return rows
+    if not rows:
+        print(f"[ab-rounds] {name}: {n} 轮，但没有全轮次都成功的配对行")
+        return rows
+    for r in rows:
+        flag = "" if r["sign_stable"] == "yes" else "  符号不稳"
+        print(f"  {r['dtype']:<16} {float(r['speedup_percent']):+7.2f}%  "
+              f"[{float(r['delta_min']):+.2f}%~{float(r['delta_max']):+.2f}%]  "
+              f"{r['before_lat']} -> {r['after_lat']}  "
+              f"{r['size_detail']}{flag}")
+    med = [float(r["speedup_percent"]) for r in rows]
+    spread = [float(r["before_spread"]) for r in rows] + \
+             [float(r["after_spread"]) for r in rows]
+    unstable = sum(1 for r in rows if r["sign_stable"] != "yes")
+    print(f"[ab-rounds] {name}: {n} 轮 × {len(rows)} shape，"
+          f"speedup 中位 {statistics.median(med):+.2f}%，"
+          f"逐轮 {min(per_round):+.2f}% ~ {max(per_round):+.2f}%，"
+          f"单侧噪声中位 {statistics.median(spread):.2f}%")
+    if unstable:
+        # The verdict, not the machine, is what these rows fail to pin down.
+        print(f"[ab-rounds] {name}: {unstable}/{len(rows)} 个 shape 的 delta "
+              "在轮次间改变符号——这些行的方向不可结论，与计时口径无关")
+    return rows
+
+
 def main() -> None:
     p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
     p.add_argument("batch_dir", type=Path, nargs="?",
@@ -252,13 +481,26 @@ def main() -> None:
                    help="输出 CSV（默认 <batch_dir>/summary.csv）")
     p.add_argument("--ab", nargs=2, type=Path, metavar=("BEFORE", "AFTER"),
                    help="两份 run.log，按 shape 配对输出该算子的 speedup 区间")
-    p.add_argument("--label", default="", help="--ab 报告里的算子名")
+    p.add_argument("--ab-rounds", type=Path, default=None, metavar="ROOT",
+                   help="多轮 A/B 目录，含 r<k>_A/run.log 与 r<k>_B/run.log；"
+                        "逐 shape 取轮次中位，并报逐轮区间与符号稳定性")
+    p.add_argument("--label", default="", help="--ab / --ab-rounds 报告里的算子名")
+    p.add_argument("--allow-caliber-mismatch", action="store_true",
+                   help="口径不一致时仍出报告（默认拒绝）。只在确认两次运行"
+                        "确实可比时使用")
     args = p.parse_args()
+    if args.ab_rounds:
+        if not args.ab_rounds.is_dir():
+            raise SystemExit(f"目录不存在: {args.ab_rounds}")
+        report_ab_rounds(args.ab_rounds, args.label,
+                         strict=not args.allow_caliber_mismatch)
+        return
     if args.ab:
         for path in args.ab:
             if not path.is_file():
                 raise SystemExit(f"run.log 不存在: {path}")
-        report_ab(args.ab[0], args.ab[1], args.label)
+        report_ab(args.ab[0], args.ab[1], args.label,
+                  strict=not args.allow_caliber_mismatch)
         return
     if args.batch_dir is None:
         raise SystemExit("需要 batch_dir，或用 --ab BEFORE AFTER")

test_backends.py

@@ -10,6 +10,7 @@ import json
 import os
 import sys
 import tempfile
+import types
 import unittest
 from pathlib import Path
 
@@ -279,6 +280,82 @@ class CaptureClassificationTests(unittest.TestCase):
         self.assertFalse(musa.denies_capture("unique"))
 
 
+class DirectRepeatTests(unittest.TestCase):
+    """Repetition on the direct timer.
+
+    _cudagraph_plugin imports triton at module scope, so it is loaded here
+    against a stub: the repeat logic is plain arithmetic and should stay
+    testable on a machine with no GPU and no Triton.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        cls._saved = {k: sys.modules.get(k) for k in ("triton", "triton.testing")}
+        triton = types.ModuleType("triton")
+        testing = types.ModuleType("triton.testing")
+        testing.do_bench = lambda *a, **k: 0.0
+        testing.do_bench_cudagraph = lambda *a, **k: 0.0
+        testing._summarize_statistics = lambda *a, **k: 0.0
+        triton.testing = testing
+        sys.modules["triton"] = triton
+        sys.modules["triton.testing"] = testing
+        import _cudagraph_plugin  # noqa: E402
+        cls.cg = _cudagraph_plugin
+
+    @classmethod
+    def tearDownClass(cls):
+        for name, mod in cls._saved.items():
+            if mod is None:
+                sys.modules.pop(name, None)
+            else:
+                sys.modules[name] = mod
+        sys.modules.pop("_cudagraph_plugin", None)
+
+    def _run(self, value, **kwargs):
+        """-> (result, how many times do_bench was called)"""
+        seq, calls = [3.0, 1.0, 2.0, 9.0, 5.0], []
+        self.cg._ORIGINAL_DO_BENCH = (
+            lambda fn, **kw: (calls.append(1), seq[(len(calls) - 1) % len(seq)])[1])
+        old = os.environ.get("ZL_DIRECT_REPEAT")
+        os.environ["ZL_DIRECT_REPEAT"] = value
+        try:
+            out = self.cg.run_direct(lambda: None, 25, 100, None,
+                                     kwargs.get("quantiles"),
+                                     kwargs.get("return_mode", "mean"))
+        finally:
+            if old is None:
+                os.environ.pop("ZL_DIRECT_REPEAT", None)
+            else:
+                os.environ["ZL_DIRECT_REPEAT"] = old
+        return out, len(calls)
+
+    def test_default_is_one_call(self):
+        self.assertEqual(self._run("1"), (3.0, 1))
+
+    def test_unset_is_one_call(self):
+        old = os.environ.pop("ZL_DIRECT_REPEAT", None)
+        try:
+            self.assertEqual(self.cg.direct_repeat(), 1)
+        finally:
+            if old is not None:
+                os.environ["ZL_DIRECT_REPEAT"] = old
+
+    def test_median_over_calls(self):
+        # 3, 1, 2, 9, 5 -> 3
+        self.assertEqual(self._run("5"), (3.0, 5))
+
+    def test_garbage_and_zero_fall_back_to_one(self):
+        self.assertEqual(self._run("abc"), (3.0, 1))
+        self.assertEqual(self._run("0"), (3.0, 1))
+        self.assertEqual(self._run("-4"), (3.0, 1))
+
+    def test_sequence_results_are_not_repeated(self):
+        # quantiles / return_mode="all" make do_bench return a sequence, and a
+        # median across calls is undefined for it; take the single call.
+        self.assertEqual(self._run("5", quantiles=[0.5])[1], 1)
+        self.assertEqual(self._run("5", return_mode="all")[1], 1)
+
+
 class DeviceArgTests(unittest.TestCase):
     def test_devices_preferred(self):
         self.assertEqual(resolve_device_arg("0,1", None), "0,1")
@@ -354,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()