bench: orchestrate a multi-card campaign without touching the measurement path

campaign.sh fans a screened operator list across cards, one ab_repeat.sh per
card, serial within a card. It does no measuring: ab_repeat.sh is the path that
produced the four operator verdicts already on record, and it stays byte for
byte as it was.

Each card gets its own AB_ROOT. Eight ab_repeat.sh started in the same second
would otherwise compute the same timestamped directory name and two cards would
write into one.

ZL_ISOLATE_TUNER_CACHE=1 is forced, since eight processes on one sqlite is the
thing to avoid. The cost is that pinned configs differ from a shared-cache
batch, so absolute latencies cannot be quoted across the two; the caliber guard
refuses such comparisons on its own, which is why this is a note and not a
warning someone has to remember.

partition.py's shell output gained the per-card weight, because EXTRA_OPS has
to land on the lightest card and counting operators gets that wrong — one heavy
operator outweighs three light ones. It writes through stdout.buffer: Python
turns "\n" into CRLF on Windows, and the trailing \r then rides into the last
field and breaks the numeric comparison that picks the card.

README.md

@@ -94,7 +94,7 @@ softmax:
 | `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` |
+| `ZL_ENABLE_CUDAGRAPH` | `auto` | `auto`/`1` 在**声明了 `supports_cuda_graph` 的后端**上走 Graph 计时（目前 cuda 与 musa），其余后端即使设 `1` 也拒绝并保持 `timer=direct`。注意 `timer=cuda_graph` 只表示**尝试过**捕获，不表示有任何一行捕获成功——S5000 上 `dropout`/`rand`/`masked_select` 是逐 shape 全部回退（每行 `BENCHMARK_DIRECT_NO_CUDAGRAPH`，零毒化、不丢行），整张表实际是 direct 计时。这种情况下设 `0` 走的是同一条路径，只是记录更诚实 |
 | `ZL_ENABLE_AUTOTUNE` | `1` | `0` 关闭 autotune record/replay 插件 |
 | `ZL_ENABLE_IR` | `1` | `0` 关闭 TTGIR / 编译元数据插件 |
 | `ZL_ENABLE_MM_CLUSTER` | `0` | `1` 且后端支持时加载 Hopper mm cluster 规避 |
@@ -137,6 +137,137 @@ ttgir 关键设计：
 - **命名 = 组内有区分度的参数**：同一 (shape, kernel) 组内取值相同的 constexpr 不进文件名；多词参数缩写为首字母（`BLOCK_SIZE_M` → `BSM`，图例见 naming.md）；同名冲突依次用 dtype（`__fp16`）、参数对齐特化（`__EMdiv16`）、hash 前缀消歧。
 - **崩溃也能拿到 IR**：dump 挂在进程 atexit 上，CUDA crash 后仍会落盘已编译部分。
 
+## 多卡跑一批：campaign.sh
+
+`ab_repeat.sh` 是单卡串行的。一批算子要并行时用 `campaign.sh` 编排——它只分活，
+测量仍然全部由 `ab_repeat.sh` 做，那个脚本一行不动。
+
+```bash
+SWITCH=FLAGTREE_MUSA_RLC_ENHANCE DEVICES=0,1,2,3,4,5,6,7   SCREEN=runs/ab_rlc_XXX/screen.tsv ZL_BACKEND=musa   FLAGGEMS_DIR=/path/to/FlagGems DRY_RUN=1 bash campaign.sh
+```
+
+`DRY_RUN=1` 先看方案，确认了再去掉它起跑。
+
+- **只取筛选表里 `IR_CHANGED` 的算子**——第一步的判定在这里生效
+- **按实测耗时贪心分片**（`partition.py`），不是 hash 取模。S5000 上 16 个算子实测：
+  hash 分片关键路径 120 分钟、有一张卡空转；贪心 70 分钟。**下界由最重的单个算子
+  决定**（`topk` 一个就 70.6 分钟），加卡不解决，分片只保证别的卡不空等
+- **每卡独立 `AB_ROOT`**——八个 `ab_repeat.sh` 同一秒启动，默认的时间戳目录名会撞
+- **强制 `ZL_ISOLATE_TUNER_CACHE=1`**。8 个进程写同一个 sqlite 是要避开的事。代价是
+  pin 的 config 与共享缓存批次不同，**两批的绝对延迟不可互引**——口径守卫会自己拒绝
+- `EXTRA_OPS` 追加筛选表里没有的算子（如 benchmark 刚修好的），放到**负载最小**的卡上
+- 跑完合并各卡 `status.tsv`（首列是卡号），并统计非零退出的格与三个标记的总数
+
+设备可见性变量名从 backends 层取，不写死 `MUSA_VISIBLE_DEVICES`。
+
+## 两步：先筛选，再实测
+
+跑一个算子的六轮 A/B 要 13~44 分钟（S5000 实测：`rand` 15 shape 13.0 分钟，
+`sort` 25 shape 43.4 分钟）。用 `ab_rlc_test.sh` 筛一对只要约 0.8 分钟。
+**差 20~50 倍**，所以先筛后测：
+
+```bash
+# 第一步：一对定去留（A 侧 record，B 侧 REPLAY_FROM 它，config 锁死）
+OPS="col2im:col2im topk:topk perf_fft:fft" VENDOR=MUSA bash ab_rlc_test.sh
+
+# 第二步：只把 IR_CHANGED 的送进来
+SWITCH=FLAGTREE_MUSA_RLC_ENHANCE ROUNDS=6 OPS="topk:topk" bash ab_repeat.sh
+```
+
+`ab_rlc_test.sh` 除了原有的 speedup 区间，还给每个算子写一行判定到
+`<root>/screen.tsv`：
+
+| 判定 | 含义 | 该怎么办 |
+|---|---|---|
+| `IR_CHANGED` | 两侧 ttgir 不同 | 排六轮 A/B |
+| `NO_IR_DIFF` | 逐字节相同（config 已锁死） | 跳过。跑六轮只会测到噪声，并报成一个"无差异"结论 |
+| `NO_TRITON` | 跑出了行，但没编出 `.ttgir` | 本次运行没有任何东西经过 Triton。算子本身可能完全正常——多半跑的是 aten 或厂商库的 kernel——只是被测的编译器 pass 没有作用对象。**既不是"开关无作用"，也不是"没有 kernel"**。`run_pytest.sh` 每个 run 用独立 `TRITON_CACHE_DIR`，所以"缓存命中没重新编译"这个解释已被排除 |
+| `NO_ROWS` | 没有测量行 | 先修可运行性：多半是 dtype 不被该算子接受，或它在本后端拿不到 config |
+
+两件容易做错的：
+
+- **判据是 ttgir 指纹，不是 `convert_layout` 计数。** cvt 数相同不等于生成代码
+  相同，而要测的是"生成代码变了没有"。两个方向都会骗人：`addmm` 的 cvt 是 15→15
+  没变，15 个 ttgir 文件却实质全不同；`tanh`/`softmax` 的 cvt 是 0→0，生成代码也
+  真的没动。cvt 数照样记，它解释的是变化的性质。
+- **指纹先抹掉开关自己盖的注解。** FlagTree 的 `_apply_musa_rlc_policy(mod)` 在开关
+  打开时把整套 RLC 策略旋钮盖成 module 属性（`ttg.rlc-*`），跟这个 pass 后面有没有
+  真的改动什么无关。不抹的话判据有个常量偏移——**凡是能编出 kernel 的算子都必然判成
+  "变了"，筛选器谁都不排除**。24 个算子实测：抹之前 23 个 `IR_CHANGED`，抹之后 7 个
+  落回 `NO_IR_DIFF`。零对照抓不到这个，因为两侧同值时属性也相同。
+- **不要全局写死一个 `DTYPES`。** `OPS` 元素支持 `OP:OP_FILE:DTYPES`，留空表示用
+  该算子自己的全套。`fft` 只吃 complex64，被一刀切的 `DTYPES=float16` 判死过一次
+  ——那是 harness 的问题，被记成了算子的问题。
+
+同一份 `_ir_fingerprint.sh` 也被 `ab_repeat.sh` 用于第 1 轮后的对比轴有效性检查，
+两处判据不会漂开。
+
+判定还管住报告：`NO_ROWS` 与 `NO_TRITON` 不出 speedup 区间（前者没有可比的测量，
+后者与被测开关无关）；`NO_IR_DIFF` 照出，但明说那是**本机对该算子的噪声底**，不是
+开关的收益——它离零多远，就是该算子单次测量的可信下限。判定刚说完这一对不可比，
+下一行就给个 `+0.3%`，正是这套工具一路在防的"看起来像结论的表"。
+
+`screen.tsv` 末七列是口径（`backend device timer bench_level shape_mode
+tuner_cache direct_repeat`），由 `python batch_summary.py --caliber <run_dir>` 取——
+`CALIBER_FIELDS` 只有那一处定义，筛选表不重列一遍。两份不同 `--level` 的
+`screen.tsv` 放一起比时，这几列是唯一能拦住的东西。
+
+每对有 `CELL_TIMEOUT`（默认 2400 秒）兜底，超时打 `!!! CELL_TIMEOUT` 并按耗时问题
+报告，不混进 dtype 那条解释里。`elapsed_before/after` 也进表：两步流程的全部依据
+就是"一对 0.8 分钟对六轮 13~44 分钟"这个比例，那这一步该记下自己那一半。
+
+## 算子跑不起来时的排查清单
+
+`NO_ROWS` / `NO_TRITON` 说的是"这一格没测出东西"，不是"开关无作用"。按下面的顺序查，
+每一条都踩过：
+
+| # | 症状 | 查什么 | 实例 |
+|---|---|---|---|
+| 1 | `PREFLIGHT_FAILED check=interpreter imports triton + flag_gems` | 用的是哪个 python。triton 通常来自 PYTHONPATH 指向的编译器构建 | 一次噪声标定因此废掉 12 格 + 160 次负载 run |
+| 2 | `PREFLIGHT_FAILED check=benchmark file exists` | `OPS` 元素是 `OP:OP_FILE`，写反就找不到文件 | `test_perf_sort` 在 `test_sort.py` 里，要写 `perf_sort:sort` |
+| 3 | `ValueError: Given dtype(s) '...' can't be supported by this op` | 该算子只吃特定 dtype，别全局写死 `DTYPES` | `fft` 只吃 complex64，被一刀切的 fp16 判成"跑不出行"，而它是 cvt 消除量最大的算子 |
+| 4 | `dynamic_func() missing ... BLOCK_M, BLOCK_N ...` 之类 JIT 深处的报错 | **benchmark 是不是导入了别家厂商的实现**（见下） | `router_gemm` 拿 NVIDIA Hopper 的 kernel 跑在 MUSA 上 |
+| 5 | 跑出行但 `kernels=0` | 该算子在本后端没走 Triton，走的是 aten 或厂商库 | — |
+
+### 第 4 条：import 成功不等于这个后端支持
+
+FlagGems 的厂商实现放在 `flag_gems/runtime/backend/_<vendor>/ops/`。benchmark 里常见
+这种可用性判据：
+
+```python
+try:
+    from flag_gems.runtime.backend._nvidia.hopper.ops.mm import router_gemm
+    ROUTER_GEMM_AVAILABLE = True          # ← import 成功就置 True
+except Exception:
+    ROUTER_GEMM_AVAILABLE = False
+```
+
+**`import` 只证明那个 `.py` 在磁盘上，它不碰设备，所以在任何后端上都成功。** gate 于是
+形同虚设：MUSA 上照样置 True，benchmark 拿 NVIDIA Hopper 的 kernel 跑在摩尔线程卡上，
+最后死在 JIT 深处的 `dynamic_func() missing BLOCK_M, BLOCK_N, BLOCK_K, SPLIT_K`——
+**那个报错完全指不向真正的原因**。而 `_mthreads/ops/mm.py` 下明明有一份能用的实现，
+benchmark 从头到尾没导入过。
+
+正确写法是先判 vendor（同一个文件里 Cambricon 那支就是对的）：
+
+```python
+if base.vendor_name == "nvidia":
+    try:
+        from flag_gems.runtime.backend._nvidia.hopper.ops.mm import router_gemm
+        ROUTER_GEMM_AVAILABLE = True
+    except Exception:
+        pass
+```
+
+`_device_setup_plugin` 会在 session 结束时扫 `sys.modules`，载入过别家厂商实现时打：
+
+```
+FOREIGN_VENDOR_MODULE active=mthreads loaded=nvidia
+```
+
+**它不拦，只报**——载入别家模块本身不一定有问题（有些是共享工具），但一个算子结果可疑
+时，这行是第一个该看的地方。
+
 ## A/B 对比测试标准流程
 
 ```bash
@@ -188,6 +319,13 @@ benchmark 目录，注册 `--level`/`--mode` 的 `conftest.py` 就没被加载
 只校验节点的文件那一半；`::test_<fn>` 交给 pytest，函数不存在时它自己会清楚地报
 `not found: ...`。
 
+还校验**解释器**——`python -c "import triton, flag_gems"`。triton 通常来自 PYTHONPATH
+指向的编译器构建产物、flag_gems 来自源码检出，所以一个能摸到全部正确路径的 shell
+仍然可能是错的那个 Python。不查的话，失败以每格一次 `pytest exit 4`（conftest 的
+ImportError）出现，多轮 campaign 会在有人读日志之前把每一格都烧掉——这不是假想，
+2026-09-10 的噪声标定实验就这么废了 12 格加 160 次负载 run。报错里带上
+`$(command -v python)`，因为要看的就是"用的到底是哪个解释器"。
+
 ### 口径核对：出报告前的拒绝
 
 `batch_summary.py --ab` / `--ab-rounds` 在解析任何 latency 之前，先读参与对比的每个
@@ -224,6 +362,25 @@ SWITCH=FLAGTREE_MUSA_RLC_ENHANCE OPS="topk masked_select" ROUNDS=6   ZL_BACKEND=
 `REPLAY_FROM` 那一份，config 因此跨轮跨侧全部锁死；逐轮交替先后顺序抵消单向漂移。
 有轮次失败的算子不出报告——缺轮的报告会让人误以为是完整的 N 轮。
 
+每跑完一格往 `<root>/status.tsv` 追一行：`round op side switch rc elapsed_s success
+failed replay_fallback no_cudagraph capture_poisoned`。README 要求必查的三个标记就在
+这张表里，跑的时候就能看见它们从 0 变成非 0，不用等结束再 grep——事后 grep 只在有人
+想起来的时候才发生。每格另有 `CELL_TIMEOUT`（默认 2400 秒）兜底，超时记 `rc=124` 并打
+`!!! CELL_TIMEOUT`，不会让一格卡死拖住整批。
+
+全程还采一份 `<root>/telemetry.tsv`：
+
+```
+timestamp            index  sm_clock_mhz  mem_clock_mhz  temp_c  power_w  util_pct
+2026-09-10T07:41:12  7      1888          2500           36      477.68   87
+```
+
+列在 cuda 与 musa 上完全一致（`nvidia-smi --query-gpu` 与 `mthreads-gmi -q --json`
+各自解析后归一），所以两台机器的 campaign 可以并排读。理由很直接：六轮 A/B 声称的是
+0.01% 量级的单侧噪声，**降频和热漂移能移动比这大得多的延迟，而且在 run.log 里不留
+任何痕迹**——报告照样干净。`TELEMETRY_INTERVAL`（默认 5 秒）设 0 关闭；采样器缺失或
+没权限时 `telemetry.py` 安静退出并把原因写进 `telemetry.err`，不影响测试本身。
+
 第 1 轮 A/B 跑完后，脚本比两侧 `ttgir/` 的指纹（相对路径 + 内容）。**逐字节相同就
 跳过这个算子**，并打 `!!! SWITCH_NO_IR_DIFF op=... switch=...`：这个开关没改变该算子的
 生成代码，剩下 N-1 轮测的全是噪声，最后还会报成一个"无差异"的结论。先查开关名拼写
@@ -354,8 +511,8 @@ python run_batch.py --ops-file my_ops.txt --replay-root runs/batch_xxx
 | `_shape_iter_inject_plugin` | 覆盖在 `get_input_iter` 里硬编码 shape 的类（conv/pool 等） | 不关 |
 | `_bespoke_shape_plugin` | 覆盖特殊输入构造的算子（upsample/flash_mla/cutlass 等，按类名注册） | 测这些算子之外可关 |
 | `_autotune_record_plugin` | record/replay Triton autotune 选择；`autotune_records/meta.json` 记录 backend，跨后端 replay 拒绝 | `ZL_ENABLE_AUTOTUNE=0` |
-| `_timing_plugin` | 按后端选择计时路径：CUDA 可走 Graph（内部调用 `_cudagraph_plugin`），其余后端一开始就是上游 `do_bench`（`timer=direct`） | `ZL_ENABLE_CUDAGRAPH=0` 强制 direct |
-| `_cudagraph_plugin` | 仅由 `_timing_plugin` 在 CUDA 上调用。`do_bench` → `do_bench_cudagraph`，capture 前按 warmup 预算显式预热；无法 capture 时回退并打 `BENCHMARK_DIRECT_NO_CUDAGRAPH` | 不要直接 `-p` 加到非 CUDA 运行 |
+| `_timing_plugin` | 按后端的 `supports_cuda_graph` 选计时路径：声明了的走 Graph（内部调用 `_cudagraph_plugin`，逐 shape 回退），没声明的一开始就是上游 `do_bench`（`timer=direct`） | `ZL_ENABLE_CUDAGRAPH=0` 强制 direct |
+| `_cudagraph_plugin` | 仅由 `_timing_plugin` 在声明了 `supports_cuda_graph` 的后端上调用。`do_bench` → `do_bench_cudagraph`，capture 前按 warmup 预算显式预热；无法 capture 时回退并打 `BENCHMARK_DIRECT_NO_CUDAGRAPH` | 不要直接 `-p` 加到非 CUDA 运行 |
 | `_musa_benchmark_compat_plugin` | 剥掉 FlagGems 上游若干 benchmark 上过时的 `torch.cuda.is_available()` 跳过门——被测代码本就经 `flag_gems.device` 分发，在 MThreads 上那些门把可用算子变成「未测量」。每条覆盖都钉死到一个文件与一条确切的 skip 理由，且仅在 `flag_gems.vendor_name == "mthreads"` 时生效 | 由 `backends/musa.py` 的 `compat_plugins` 声明，其他后端不加载 |
 | `_pretty_report_plugin` | 结果表整理 + 着色：全表相同的输入折叠成表头下一行图例，每行 Size Detail 只留随行变化的部分（shape 保持 `torch.Size([...])` 原样，MoE 类算子单行从 ~400 字符缩到一屏内）；SUCCESS 绿 / FAILED 红。列名与 `SUCCESS` 字样保持上游原文，`run.log` 的 grep/解析不受影响 | 需要与上游原始表格逐字对齐时去掉 |
 | `_ir_meta_plugin` | 编译期记录每个变体的 constexpr/签名/特化；launch 钩子统计每个 (kernel, shape) 的实际使用；退出时按上述结构 dump ttgir。Triton knobs 缺失时 fail-soft | `ZL_ENABLE_IR=0` |

_device_setup_plugin.py

@@ -67,3 +67,46 @@ def pytest_configure(config):
 
 
 _setup_at_import()
+
+# FlagGems 的厂商实现按 flag_gems.runtime.backend._<vendor>.ops.* 组织。benchmark
+# 常写成 `try: import <某厂商实现>; AVAILABLE = True`——但 import 只证明那个 .py 在
+# 磁盘上，它不碰设备，所以在任何后端上都会成功。于是 gate 形同虚设，测出来的是别家
+# 的 kernel 跑在本机上。router_gemm 就是这样：test_router_gemm.py 无条件导入
+# _nvidia.hopper 版，在 MUSA 上 gate 照过，最后死在 JIT 深处的
+# "dynamic_func() missing BLOCK_M, BLOCK_N, BLOCK_K, SPLIT_K"——那个报错完全指不
+# 向真正的原因，而 _mthreads 下明明有一份能用的实现。
+#
+# 这里不拦，只报：跑完看一眼载入过谁家的实现。
+_VENDOR_MOD_PREFIX = "flag_gems.runtime.backend._"
+
+
+def foreign_vendor_modules(active_vendor: str, modules=None) -> list:
+    """本次进程里载入过的、不属于当前厂商的 FlagGems 后端模块。"""
+    if modules is None:
+        modules = sys.modules
+    out = set()
+    for name in list(modules):
+        if not name.startswith(_VENDOR_MOD_PREFIX):
+            continue
+        # 只保留厂商根，_nvidia.hopper.ops.mm 与 _nvidia.ops.mm 算同一条。
+        # 按段比而不是按前缀：前缀会让 _<vendor>_something 被误认成自家。
+        root = name[len(_VENDOR_MOD_PREFIX):].split(".")[0]
+        if root != active_vendor:
+            out.add(root)
+    return sorted(out)
+
+
+def pytest_sessionfinish(session, exitstatus):
+    from _term_style import tag
+    try:
+        vendor = get_backend().flaggems_vendor
+    except Exception:
+        return
+    foreign = foreign_vendor_modules(vendor)
+    if not foreign:
+        return
+    print(f"{tag('[device-setup-plugin]')} FOREIGN_VENDOR_MODULE active={vendor} "
+          f"loaded={','.join(foreign)}", file=sys.stderr, flush=True)
+    print(f"{tag('[device-setup-plugin]')} 本次载入了别家厂商的 FlagGems 实现。"
+          "如果这个算子的结果可疑，先查 benchmark 的可用性判据是不是只做了 import "
+          "而没判 vendor_name", file=sys.stderr, flush=True)

_ir_fingerprint.sh

@@ -0,0 +1,31 @@
+#!/bin/bash
+# 由 ab_repeat.sh 与 screen.sh 共用；两处必须用同一个判据，否则筛选放行的算子
+# 到了 A/B 那步又被拦下，或者反过来。
+
+# 一个 run 生成代码的指纹：只取 .ttgir 的相对路径 + 内容，并先抹掉开关自己盖上去的
+# 注解。
+#
+# 只认 .ttgir：dump 目录里还有 index.tsv 和 naming.md，一个 kernel 都没编出来的 run
+# 只剩这两个文件、两侧完全相同，会把"什么都没生成"报成"开关没改变生成代码"。
+#
+# 抹掉 ttg.rlc-*：FlagTree 的 _apply_musa_rlc_policy(mod) 在开关打开时把整套 RLC 策略
+# 旋钮盖成 module 属性，跟这个 pass 后面有没有真的改动什么无关。不抹的话判据有个常量
+# 偏移——凡是能编出 kernel 的算子都必然判成"变了"，筛选器谁都不排除。实测 24 个算子
+# 里有 7 个（tanh gelu mul softmax mm bmm replication_pad3d）除这行属性外逐字节相同。
+# 零对照抓不到它：两侧同值时属性也相同。
+#
+# 零 kernel 返回非零，交给调用方当作"没有可比的 IR"，而不是"一致"。
+ir_fingerprint() {
+  local dir="$1/ttgir"
+  [[ -d "$dir" ]] || return 1
+  local n
+  n=$(find "$dir" -name '*.ttgir' -type f 2>/dev/null | wc -l)
+  (( n > 0 )) || return 1
+  local out
+  out=$( { find "$dir" -name '*.ttgir' -type f | LC_ALL=C sort | while IFS= read -r f; do
+             printf '%s
+' "${f#"$dir"/}"
+             sed -E 's/"ttg[.]rlc-[^"]*" = [0-9-]+ : i32, ?//g' "$f"
+           done; } | sha256sum | cut -d' ' -f1 )
+  [[ -n "$out" ]] && printf '%s' "$out"
+}

_timing_plugin.py

@@ -1,10 +1,15 @@
 """Select the timing path from backend capabilities, before any shape runs.
 
-CUDA keeps the existing `_cudagraph_plugin` behaviour (graph capture, with
-per-shape fallback to plain do_bench). Other backends do not attempt CUDA
-Graph at all: they leave `triton.testing.do_bench` in place and record
-`timer=direct`. Trying capture-then-fallback on a non-CUDA device would
-produce FATAL_DEVICE noise and mixed-caliber rows.
+Whether capture is attempted comes from the backend's `supports_cuda_graph`,
+not from "is this CUDA". CUDA and MUSA both declare it, so both get graph
+capture with per-shape fallback to plain do_bench; a backend that declares
+nothing leaves `triton.testing.do_bench` in place and records `timer=direct`.
+
+`timer=cuda_graph` therefore means capture was *attempted*, not that any row
+was captured. On S5000 every shape of dropout/rand/masked_select falls back
+(BENCHMARK_DIRECT_NO_CUDAGRAPH on every row, nothing poisoned, no rows lost),
+so the whole table is direct-timed under a cuda_graph label. Setting
+ZL_ENABLE_CUDAGRAPH=0 there selects the same timing path and says so plainly.
 """
 
 from __future__ import annotations

ab_repeat.sh

@@ -30,31 +30,68 @@ 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"
-}
+source "$(dirname "${BASH_SOURCE[0]}")/_ir_fingerprint.sh"
 
 : "${SWITCH:?SWITCH is required, e.g. FLAGTREE_MUSA_RLC_ENHANCE}"
 OPS="${OPS:-}"
 : "${OPS:?OPS is required, e.g. \"topk masked_select\"}"
 ROUNDS="${ROUNDS:-6}"
+CELL_TIMEOUT="${CELL_TIMEOUT:-2400}"
 
 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)"
+# AB_ROOT 供 campaign.sh 指定每卡独立的产物根：八个 ab_repeat.sh 同一秒启动时，
+# 默认的时间戳目录名会撞在一起，两张卡往同一个目录里写。
+ROOT="${AB_ROOT:-runs/ab_repeat_${SWITCH}_$(date +%Y%m%d-%H%M%S)}"
 mkdir -p "$ROOT"
 LOG="$ROOT/summary.log"
+STATUS="$ROOT/status.tsv"
+
+# 全程采时钟/温度/功耗。六轮 A/B 声称的是 0.01% 量级的单侧噪声，降频和热漂移
+# 能移动比这大得多的延迟，而且在 run.log 里不留痕迹——报告照样干净。
+# TELEMETRY_INTERVAL=0 关闭。采不到时 telemetry.py 自己安静退出，不影响测试。
+TELEMETRY_INTERVAL="${TELEMETRY_INTERVAL:-5}"
+TELE_PID=""
+if [[ "$TELEMETRY_INTERVAL" != "0" ]]; then
+  python telemetry.py --backend "${ZL_BACKEND:-cuda}" --device "${ZL_DEVICE:-}" \
+    --interval "$TELEMETRY_INTERVAL" --run \
+    > "$ROOT/telemetry.tsv" 2> "$ROOT/telemetry.err" &
+  TELE_PID=$!
+  trap '[[ -n "$TELE_PID" ]] && kill "$TELE_PID" 2>/dev/null' EXIT
+fi
+printf 'round\top\tside\tswitch\trc\telapsed_s\tsuccess\tfailed\treplay_fallback\tno_cudagraph\tcapture_poisoned\n' > "$STATUS"
+
+# 跑一格并逐格登记。事后 grep 也能拿到这些数，但只在有人想起来时才发生；
+# 缺格、超时、标记从 0 变成非 0，都要在跑的时候就看得见。
+run_cell() {   # round side op op_file out_dir [replay_from]
+  local round="$1" side="$2" op="$3" op_file="$4" dir="$5" replay="${6:-}"
+  local val=0; [[ "$side" == B ]] && val=1
+  local start rc elapsed log ok fail fb ng po
+  start=$(date +%s)
+  if [[ -n "$replay" ]]; then
+    OP="$op" OP_FILE="$op_file" OUT_DIR="$dir" REPLAY_FROM="$replay" \
+      timeout -k 30 "$CELL_TIMEOUT" env "$SWITCH=$val" bash run_pytest.sh
+  else
+    OP="$op" OP_FILE="$op_file" OUT_DIR="$dir" \
+      timeout -k 30 "$CELL_TIMEOUT" env "$SWITCH=$val" bash run_pytest.sh
+  fi
+  rc=$?
+  elapsed=$(( $(date +%s) - start ))
+  (( rc == 124 || rc == 137 )) && echo "!!! CELL_TIMEOUT ${CELL_TIMEOUT}s op=$op round=$round side=$side"
+  log="$dir/run.log"
+  # grep -c 没命中时打印 0 但以 1 退出；不吞掉的话会被当成本格失败
+  ok=$(grep -c '^SUCCESS' "$log" 2>/dev/null || true)
+  fail=$(grep -c '^FAILED' "$log" 2>/dev/null || true)
+  fb=$(grep -c 'AUTOTUNE_REPLAY_FALLBACK' "$log" 2>/dev/null || true)
+  ng=$(grep -c 'BENCHMARK_DIRECT_NO_CUDAGRAPH' "$log" 2>/dev/null || true)
+  po=$(grep -c 'BENCHMARK_CAPTURE_POISONED' "$log" 2>/dev/null || true)
+  printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
+    "$round" "$op" "$side" "$val" "$rc" "$elapsed" \
+    "${ok:-0}" "${fail:-0}" "${fb:-0}" "${ng:-0}" "${po:-0}" >> "$STATUS"
+  return $rc
+}
 
 {
   echo ">>> switch=$SWITCH  rounds=$ROUNDS  ops=$OPS"
@@ -67,8 +104,7 @@ LOG="$ROOT/summary.log"
 
     # 第 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
+    if ! run_cell 1 A "$op" "$op_file" "$out/r1_A"; then
       echo "!!! FAILED $op round 1 A; skipping this operator"
       continue
     fi
@@ -82,8 +118,7 @@ LOG="$ROOT/summary.log"
         [[ $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
+        if ! run_cell "$k" "$side" "$op" "$op_file" "$out/r${k}_${side}" "$REC"; then
           echo "!!! FAILED $op round $k $side"
           ok=0
         fi
@@ -93,7 +128,8 @@ LOG="$ROOT/summary.log"
         fa=$(ir_fingerprint "$out/r1_A") || fa=""
         fb=$(ir_fingerprint "$out/r1_B") || fb=""
         if [[ -z "$fa" || -z "$fb" ]]; then
-          echo "??? $op: 没有 ttgir 产物，跳过对比轴有效性检查"
+          echo "??? $op: 没有编出任何 .ttgir（可能该算子没走 Triton，或本轮失败），"
+          echo "??? 跳过对比轴有效性检查——这不等于开关无作用"
         elif [[ "$fa" == "$fb" ]]; then
           echo "!!! SWITCH_NO_IR_DIFF op=$op switch=$SWITCH"
           echo "!!! 两侧 ttgir 逐字节相同：这个开关没有改变 $op 的生成代码。"
@@ -125,5 +161,8 @@ LOG="$ROOT/summary.log"
 sed -i -E $'s/\x1b\\[[0-9;]*[A-Za-z]//g' "$LOG"
 
 echo ">>> 总 log: $(pwd)/$LOG"
+echo ">>> 逐格状态表: $(pwd)/$STATUS"
+[[ -s "$ROOT/telemetry.tsv" ]] && echo ">>> 时钟/温度: $(pwd)/$ROOT/telemetry.tsv"
+[[ -s "$ROOT/telemetry.err" ]] && echo ">>> 遥测提示: $(cat "$ROOT/telemetry.err")"
 echo ">>> 逐算子重出报告: python batch_summary.py --ab-rounds $ROOT/<op> --label <op>"
 echo ">>> 解读要点: 逐轮区间宽说明判定不稳；单侧噪声大说明机器不稳；两者含义不同"

ab_rlc_test.sh

@@ -10,15 +10,79 @@
 #
 # 跑完用 batch_summary.py --ab <before> <after> 得到该算子的 speedup 区间。
 #
+# 这一轮同时充当 ab_repeat.sh 的筛选前置：一对的代价约 0.8 分钟，六轮 A/B 是
+# 13~44 分钟（S5000 实测），差 20~50 倍。所以先用它过一遍，只把 IR_CHANGED 的
+# 算子送进 ab_repeat.sh。判定写在 <root>/screen.tsv，四种之一：
+#
+#   IR_CHANGED  两侧 ttgir 不同        -> 排六轮
+#   NO_IR_DIFF  逐字节相同（config 已锁死）-> 跳过，跑六轮只会测到噪声
+#   NO_TRITON   跑出行但没编出 .ttgir  -> 该算子在本后端没走 Triton，
+#                                        编译器开关无从谈起。不是"开关无作用"
+#   NO_ROWS     没有测量行            -> 先修可运行性，多半是 dtype
+#
+# 判据是 ttgir 指纹而不是 convert_layout 计数：cvt 数相同不等于生成代码相同。
+# cvt 数照样记，它解释的是变化的性质。
+#
 # 用法：
 #   OPS="softmax gelu_backward" bash ab_rlc_test.sh
-#   OPS="sort:perf_sort topk" bash ab_rlc_test.sh     # 冒号分隔 OP:OP_FILE
+#   OPS="sort:perf_sort topk" bash ab_rlc_test.sh          # OP:OP_FILE
+#   OPS="perf_fft:fft:complex64 topk:topk" bash ab_rlc_test.sh   # 再加 DTYPES
 #
-# 注意：RLC 开关名带厂商前缀，用 VENDOR 指定（默认 MUSA）。套用前先确认开关有效：
-# grep 该开关名于 FlagTree 源码应有命中，否则两侧跑的是同一份代码。
+# DTYPES 留空表示用该算子自己的全套。**不要全局写死一个 dtype**：fft 只吃
+# complex64，被一刀切的 DTYPES=float16 判死过一次——那是 harness 的问题，
+# 却被记成了算子的问题。
+#
+# 注意：RLC 开关名带厂商前缀，用 VENDOR 指定（默认 MUSA）。开关名拼错时两侧跑
+# 同一份代码，此时判定会是 NO_IR_DIFF——先查拼写，再下"开关无作用"的结论。
 set -uo pipefail
 
 cd "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+source "./_ir_fingerprint.sh"
+
+# RESCORE=<已有的 root>：判据改了之后，用当前的 ir_fingerprint 对盘上已有的 ttgir
+# 重新打分并改写该 root 的 screen.tsv，不重跑任何测量。
+#
+# 需要它是因为判据确实改过：_ir_fingerprint.sh 起初把 index.tsv 也算进哈希，后来又
+# 发现要抹掉开关自己盖的 ttg.rlc-* 属性。同一批 24 个算子，结论从 23 个 IR_CHANGED
+# 变成 16 个。重打分是秒级的，重跑那批筛选要 79 分钟。
+#
+# 原表存为 screen.tsv.before-rescore，判定有变化的算子逐个打出来。
+RESCORE="${RESCORE:-}"
+if [[ -n "$RESCORE" ]]; then
+  [[ -d "$RESCORE" ]] || { echo "!!! 目录不存在: $RESCORE" >&2; exit 2; }
+  TSV="$RESCORE/screen.tsv"
+  [[ -f "$TSV" ]] || { echo "!!! 没有 screen.tsv: $TSV" >&2; exit 2; }
+  cp "$TSV" "$TSV.before-rescore"
+
+  hdr=$(head -1 "$TSV.before-rescore")
+  # 列名 -> 序号，不写死列序
+  idx() { echo "$hdr" | tr '\t' '\n' | grep -nxF "$1" | cut -d: -f1; }
+  c_op=$(idx op); c_rb=$(idx rows_before); c_ra=$(idx rows_after); c_v=$(idx verdict)
+  [[ -n "$c_op" && -n "$c_v" ]] || { echo "!!! screen.tsv 缺 op/verdict 列" >&2; exit 2; }
+
+  printf '%s\n' "$hdr" > "$TSV"
+  changed=0
+  while IFS= read -r line; do
+    op=$(printf '%s' "$line" | cut -f"$c_op")
+    old=$(printf '%s' "$line" | cut -f"$c_v")
+    rb=$(printf '%s' "$line" | cut -f"${c_rb:-0}"); ra=$(printf '%s' "$line" | cut -f"${c_ra:-0}")
+    fa=$(ir_fingerprint "$RESCORE/${op}_before") || fa=""
+    fb=$(ir_fingerprint "$RESCORE/${op}_after")  || fb=""
+    if [[ "${rb:-0}" == 0 || "${ra:-0}" == 0 ]]; then v=NO_ROWS
+    elif [[ -z "$fa" || -z "$fb" ]]; then v=NO_TRITON
+    elif [[ "$fa" == "$fb" ]]; then v=NO_IR_DIFF
+    else v=IR_CHANGED; fi
+    printf '%s\n' "$line" | awk -v FS='\t' -v OFS='\t' -v i="$c_v" -v v="$v" '{$i=v; print}' >> "$TSV"
+    if [[ "$old" != "$v" ]]; then
+      printf '  %-22s %-12s -> %s\n' "$op" "$old" "$v" >&2
+      changed=$((changed+1))
+    fi
+  done < <(tail -n +2 "$TSV.before-rescore")
+
+  echo ">>> 重打分完成，$changed 个算子的判定变了；原表存为 $TSV.before-rescore"
+  awk -v i="$c_v" -F'\t' 'NR>1{c[$i]++} END{for (k in c) printf("  %-12s %d\n", k, c[k])}' "$TSV"
+  exit 0
+fi
 
 VENDOR="${VENDOR:-MUSA}"
 ENHANCE_VAR="FLAGTREE_${VENDOR}_RLC_ENHANCE"
@@ -26,38 +90,103 @@ ENHANCE_VAR="FLAGTREE_${VENDOR}_RLC_ENHANCE"
 # 空格分隔的算子表。元素为 OP 或 OP:OP_FILE（省略时 OP_FILE=OP）。
 OPS="${OPS:-softmax}"
 
+# 筛选是第一次接触一个算子的地方，最容易撞上挂死的那个。没有上限时一个算子
+# 能把整批停在那里，而筛选的全部意义就是快。
+CELL_TIMEOUT="${CELL_TIMEOUT:-2400}"
+
 if [[ -t 1 && -z "${NO_COLOR:-}" ]]; then
   export FLAGGEMS_PERF_COLOR=always
 fi
 
-LOG="runs/ab_rlc_$(date +%Y%m%d-%H%M%S).log"
-mkdir -p runs
+# 显式 OUT_DIR，不再用 `ls -dt runs/<op>_*` 猜"最新那个就是我刚跑的"。串行下
+# 那个假设成立，多卡并行共用 runs/ 就不成立了。
+ROOT="runs/ab_rlc_$(date +%Y%m%d-%H%M%S)"
+mkdir -p "$ROOT"
+LOG="$ROOT/summary.log"
+TSV="$ROOT/screen.tsv"
+printf 'op	op_file	dtypes	rows_before	rows_after	kernels_before	kernels_after	cvt_before	cvt_after	elapsed_before	elapsed_after	verdict	backend	device	timer	bench_level	shape_mode	tuner_cache	direct_repeat\n' > "$TSV"
+
+count_cvt()     { find "$1/ttgir" -name '*.ttgir' -type f -exec cat {} + 2>/dev/null | grep -c convert_layout || true; }
+count_kernels() { find "$1/ttgir" -name '*.ttgir' -type f 2>/dev/null | wc -l; }
+count_rows()    { grep -c '^SUCCESS' "$1/run.log" 2>/dev/null || true; }
 
 {
   echo ">>> vendor=$VENDOR  switch=$ENHANCE_VAR  ops=$OPS"
   for entry in $OPS; do
-    op="${entry%%:*}"
-    op_file="${entry#*:}"
-    [[ "$op_file" == "$entry" ]] && op_file="$op"
+    op="${entry%%:*}"; rest="${entry#*:}"
+    if [[ "$rest" == "$entry" ]]; then op_file="$op"; dtypes=""
+    else
+      op_file="${rest%%:*}"; d="${rest#*:}"
+      [[ "$d" == "$rest" ]] && dtypes="" || dtypes="$d"
+    fi
+    BEFORE="$ROOT/${op}_before"; AFTER="$ROOT/${op}_after"
 
     echo "########## $op | $ENHANCE_VAR=0 before (record) ##########"
-    if ! OP="$op" OP_FILE="$op_file" env "$ENHANCE_VAR=0" bash run_pytest.sh; then
-      echo "!!! FAILED before: $op"
-      continue
-    fi
-    BEFORE=$(ls -dt "runs/${op}_"* | head -1)
+    t0=$(date +%s)
+    OP="$op" OP_FILE="$op_file" OUT_DIR="$BEFORE" DTYPES="$dtypes" \
+      timeout -k 30 "$CELL_TIMEOUT" env "$ENHANCE_VAR=0" bash run_pytest.sh
+    rc_before=$?; el_before=$(( $(date +%s) - t0 ))
 
     echo "########## $op | $ENHANCE_VAR=1 after (replay $BEFORE) ##########"
-    if ! OP="$op" OP_FILE="$op_file" REPLAY_FROM="$BEFORE" env "$ENHANCE_VAR=1" bash run_pytest.sh; then
-      echo "!!! FAILED after: $op"
-      continue
-    fi
-    AFTER=$(ls -dt "runs/${op}_"* | head -1)
+    t0=$(date +%s)
+    OP="$op" OP_FILE="$op_file" OUT_DIR="$AFTER" DTYPES="$dtypes" \
+      REPLAY_FROM="$BEFORE" timeout -k 30 "$CELL_TIMEOUT" env "$ENHANCE_VAR=1" bash run_pytest.sh
+    rc_after=$?; el_after=$(( $(date +%s) - t0 ))
+    for rc in "$rc_before" "$rc_after"; do
+      (( rc == 124 || rc == 137 )) && echo "!!! CELL_TIMEOUT ${CELL_TIMEOUT}s op=$op"
+    done
 
+    rb=$(count_rows "$BEFORE");     ra=$(count_rows "$AFTER")
+    kb=$(count_kernels "$BEFORE");  ka=$(count_kernels "$AFTER")
+    cb=$(count_cvt "$BEFORE");      ca=$(count_cvt "$AFTER")
+    fb=$(ir_fingerprint "$BEFORE") || fb=""
+    fa=$(ir_fingerprint "$AFTER")  || fa=""
+
+    if (( rc_before != 0 || rc_after != 0 )) || [[ "${rb:-0}" == 0 || "${ra:-0}" == 0 ]]; then
+      verdict=NO_ROWS
+      if (( rc_before == 124 || rc_after == 124 || rc_before == 137 || rc_after == 137 )); then
+        note="超过 CELL_TIMEOUT=${CELL_TIMEOUT}s 被截断（rc=$rc_before/$rc_after）。这是耗时问题，不是对开关的判定——先看它为什么这么慢"
+      else
+        note="跑不出测量行（rc=$rc_before/$rc_after）。看 run.log——多半是 dtype 不被该算子接受，或它在本后端拿不到 config"
+      fi
+    elif [[ -z "$fb" || -z "$fa" ]]; then
+      verdict=NO_TRITON
+      note="跑出了行但没编出 .ttgir：本次运行没有任何东西经过 Triton。算子本身可能完全正常（多半跑的是 aten 或厂商库），只是被测的编译器 pass 没有作用对象。不是'开关无作用'，也不是'没有 kernel'"
+    elif [[ "$fb" == "$fa" ]]; then
+      verdict=NO_IR_DIFF
+      note="两侧 ttgir 逐字节相同（config 已锁死）：开关没改变生成代码。先查开关名拼写，再下结论"
+    else
+      verdict=IR_CHANGED
+      note="生成代码有变（cvt ${cb}→${ca}，指纹 ${fb:0:8}/${fa:0:8}）：值得排六轮 A/B"
+    fi
+    # 口径从 batch_summary.py 取，那里是 CALIBER_FIELDS 的唯一定义。两份不同
+    # --level 的 screen.tsv 放一起比时，这几列是唯一能拦住的东西。
+    cal=$(python batch_summary.py --caliber "$BEFORE" 2>/dev/null)
+    printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
+      "$op" "$op_file" "${dtypes:-<all>}" "${rb:-0}" "${ra:-0}" \
+      "$kb" "$ka" "${cb:-0}" "${ca:-0}" "$el_before" "$el_after" "$verdict" \
+      "$cal" >> "$TSV"
+    echo ">>> $op: $verdict — $note"
+
+    # 判定说这一对不可比，就别在下一行给出一个看起来像结论的区间
+    case "$verdict" in
+      NO_ROWS)
+        echo "!!! $op 没有可对比的测量，不出 speedup 区间"; continue ;;
+      NO_TRITON)
+        echo "!!! $op 没有经过 Triton，speedup 区间与被测开关无关，不出"; continue ;;
+      NO_IR_DIFF)
+        echo ">>> $op 两侧生成代码相同：下面这个区间是本机对该算子的噪声底，"
+        echo ">>> 不是开关的收益。它离零多远，就是该算子单次测量的可信下限。" ;;
+    esac
     echo "########## $op | summary ##########"
     python batch_summary.py --ab "$BEFORE/run.log" "$AFTER/run.log" --label "$op" \
       || echo "!!! summary failed: $op"
   done
+
+  echo "########## 筛选汇总 ##########"
+  awk -F'\t' 'NR>1{printf "  %-12s %-22s cvt %s->%s  kernels %s/%s  rows %s/%s  %ss\n", \
+       $12,$1,$8,$9,$6,$7,$4,$5,$10+$11}' "$TSV" | sort
+  echo ">>> 只把 IR_CHANGED 的算子送进 ab_repeat.sh。"
   echo "########## ALL DONE ##########"
 } 2>&1 | tee "$LOG"
 
@@ -65,4 +194,5 @@ mkdir -p runs
 sed -i -E $'s/\x1b\\[[0-9;]*[A-Za-z]//g' "$LOG"
 
 echo ">>> 总 log: $(pwd)/$LOG"
-echo ">>> 每算子两份 run.log 在 runs/<op>_<时间戳>/ 下，按时间先后即 before / after"
+echo ">>> 筛选表: $(pwd)/$TSV"
+echo ">>> 每算子两份 run.log 在 $ROOT/<op>_{before,after}/ 下"

backends/base.py

@@ -72,6 +72,34 @@ class DeviceBackend:
     # backend. Skipped up front so one of them cannot cost a whole run.
     capture_deny_ops: Tuple[str, ...] = ()
 
+    # One-shot device telemetry sampler. Empty means this backend has none, and
+    # the campaign runs without a clock/thermal record rather than pretending.
+    # "{dev}" is dropped from the argv when no device is pinned, so the sampler
+    # covers every visible device instead of guessing an index.
+    telemetry_argv: Tuple[str, ...] = ()
+
+    def telemetry_command(self, device: str = "") -> List[str]:
+        if not self.telemetry_argv:
+            return []
+        out: List[str] = []
+        for a in self.telemetry_argv:
+            if "{dev}" in a:
+                if not device:
+                    # Drop the flag the placeholder belonged to as well; a bare
+                    # trailing "-i" makes the sampler fail instead of widening.
+                    if out and out[-1].startswith("-"):
+                        out.pop()
+                    continue
+                a = a.replace("{dev}", device)
+            out.append(a)
+        return out
+
+    def parse_telemetry(self, raw: str) -> List[Dict[str, str]]:
+        """One dict per device: index, sm_clock_mhz, mem_clock_mhz, temp_c,
+        power_w, util_pct. Normalised so a CUDA and a MUSA campaign produce the
+        same columns and can be read side by side."""
+        return []
+
     def classify_capture_error(self, exc: BaseException) -> str:
         """-> "fatal" | "poisoned" | "recoverable".
 

backends/cuda.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import Dict, List
+
 from .base import DeviceBackend
 
 
@@ -26,3 +28,22 @@ class CudaBackend(DeviceBackend):
     # No CUDA wording is known to poison the context without also matching one
     # of the fatal markers above, so nothing is listed here rather than guessed.
     poisoning_error_markers = ()
+
+    telemetry_argv = (
+        "nvidia-smi",
+        "--query-gpu=index,clocks.sm,clocks.mem,temperature.gpu,power.draw,utilization.gpu",
+        "--format=csv,noheader,nounits",
+        "-i", "{dev}",
+    )
+
+    def parse_telemetry(self, raw: str) -> List[Dict[str, str]]:
+        out: List[Dict[str, str]] = []
+        for line in raw.splitlines():
+            parts = [p.strip() for p in line.split(",")]
+            if len(parts) != 6 or not parts[0].isdigit():
+                continue
+            out.append(dict(zip(
+                ("index", "sm_clock_mhz", "mem_clock_mhz",
+                 "temp_c", "power_w", "util_pct"), parts)))
+        return out
+

backends/musa.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import Dict, List
+
 from .base import DeviceBackend
 
 
@@ -50,3 +52,50 @@ class MusaBackend(DeviceBackend):
     # Device count comes from torch.musa or MUSA_VISIBLE_DEVICES. Do not guess
     # a vendor CLI here; confirm the real listing tool before adding it.
     list_command = None
+
+    # mthreads-gmi has no CSV query mode; --json is the only machine-readable
+    # form it offers. Values arrive with their units glued on ("36C", "1888MHz",
+    # "477.68W", "87%"), so the parser strips trailing non-numerics rather than
+    # trusting a fixed suffix.
+    telemetry_argv = (
+        "mthreads-gmi", "-q",
+        "-d", "TEMPERATURE,CLOCK,POWER,UTILIZATION",
+        "--json",
+        "-i", "{dev}",
+    )
+
+    def parse_telemetry(self, raw: str) -> List[Dict[str, str]]:
+        import json
+        import re
+
+        def num(v):
+            m = re.match(r"^\s*(-?[\d.]+)", str(v or ""))
+            return m.group(1) if m else ""
+
+        start = raw.find("{")
+        if start < 0:
+            return []
+        try:
+            data = json.loads(raw[start:])
+        except Exception:
+            return []
+        gpus = data.get("GPU") or []
+        if isinstance(gpus, dict):
+            gpus = [gpus]
+        out: List[Dict[str, str]] = []
+        for g in gpus:
+            clocks = g.get("Clocks") or {}
+            power = g.get("Power Readings") or {}
+            # 键名带尾随空格（"Power Draw "），别按字面写死
+            draw = next((v for k, v in power.items()
+                         if k.strip() == "Power Draw"), "")
+            out.append({
+                "index": str(g.get("Index", "")).strip(),
+                "sm_clock_mhz": num(clocks.get("Graphics")),
+                "mem_clock_mhz": num(clocks.get("Memory")),
+                "temp_c": num((g.get("Temperature") or {}).get("GPU Current Temp")),
+                "power_w": num(draw),
+                "util_pct": num((g.get("Utilization") or {}).get("Gpu")),
+            })
+        return out
+

batch_summary.py

@@ -485,10 +485,17 @@ def main() -> None:
                    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("--caliber", type=Path, default=None, metavar="RUN_DIR",
+                   help="打印该 run 的口径字段（TSV 一行），供筛选表复用")
     p.add_argument("--allow-caliber-mismatch", action="store_true",
                    help="口径不一致时仍出报告（默认拒绝）。只在确认两次运行"
                         "确实可比时使用")
     args = p.parse_args()
+    if args.caliber is not None:
+        # 口径的定义只有 CALIBER_FIELDS 一处；筛选表等复用它，不要在别处重列
+        cal = caliber_of(args.caliber) or {}
+        print("	".join(str(cal.get(k, "")) for k in CALIBER_FIELDS))
+        return
     if args.ab_rounds:
         if not args.ab_rounds.is_dir():
             raise SystemExit(f"目录不存在: {args.ab_rounds}")

campaign.sh

@@ -0,0 +1,128 @@
+#!/bin/bash
+# 多卡跑一批算子的六轮 A/B：每卡一个 ab_repeat.sh，卡内串行。
+#
+# 它只做编排，不做测量。ab_repeat.sh 一行不动——那是已经跑出结论的、验证过的串行
+# 路径，不该为了并行去改它。分片交给 partition.py（按实测耗时贪心，比 hash 分片在
+# 这批活上快 50 分钟）。
+#
+# 并行必须隔离 tuner cache：8 个进程写同一个 sqlite 是要避开的事。代价是 pin 下来
+# 的 config 与共享缓存下跑的批次不同，两批的绝对延迟不可互引——batch_summary.py
+# 的口径守卫会自己拒绝这种对比，不用记着。
+#
+# 用法：
+#   SWITCH=FLAGTREE_MUSA_RLC_ENHANCE DEVICES=0,1,2,3,4,5,6,7 \
+#     SCREEN=runs/ab_rlc_XXX/screen.tsv \
+#     ZL_BACKEND=musa FLAGGEMS_DIR=/path/to/FlagGems bash campaign.sh
+#
+#   DRY_RUN=1      只打印分片方案，不起跑
+#   EXTRA_OPS=".." 追加筛选表里没有的算子（如 benchmark 刚修好的），塞给最空的卡
+#   ROUNDS=6  CELL_TIMEOUT=2400  VERDICT=IR_CHANGED
+set -uo pipefail
+
+cd "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+: "${SWITCH:?SWITCH is required, e.g. FLAGTREE_MUSA_RLC_ENHANCE}"
+: "${SCREEN:?SCREEN is required, 指向 ab_rlc_test.sh 的 screen.tsv}"
+DEVICES="${DEVICES:-0}"
+ROUNDS="${ROUNDS:-6}"
+CELL_TIMEOUT="${CELL_TIMEOUT:-2400}"
+VERDICT="${VERDICT:-IR_CHANGED}"
+EXTRA_OPS="${EXTRA_OPS:-}"
+DRY_RUN="${DRY_RUN:-0}"
+TAB=$'\t'
+
+[[ -f "$SCREEN" ]] || { echo "!!! 筛选表不存在: $SCREEN" >&2; exit 2; }
+
+# 设备可见性变量名从 backends 层拿，不写死 MUSA_VISIBLE_DEVICES
+VIS_KEY=$(python -c "import sys; sys.path.insert(0, '.')
+from backends import get_backend
+print(get_backend().visibility_env_keys[0])" 2>/dev/null) || VIS_KEY=""
+[[ -n "$VIS_KEY" ]] || { echo "!!! 取不到后端的可见性变量名，检查 ZL_BACKEND" >&2; exit 2; }
+
+echo ">>> switch=$SWITCH  rounds=$ROUNDS  devices=$DEVICES  可见性变量=$VIS_KEY"
+python partition.py --screen "$SCREEN" --devices "$DEVICES" \
+  --verdict "$VERDICT" --rounds "$ROUNDS" || exit 2
+
+# 每卡一行 "<device><TAB><OPS><TAB><权重秒>"
+mapfile -t PLAN < <(python partition.py --screen "$SCREEN" --devices "$DEVICES" \
+  --verdict "$VERDICT" --format shell) || exit 2
+
+if [[ -n "$EXTRA_OPS" ]]; then
+  # 追加到当前负载最小的卡。按算子个数选是错的——一个重算子能顶三个轻的，
+  # partition.py 的第三列给的就是权重秒数
+  lightest=0; least=""
+  for i in "${!PLAN[@]}"; do
+    IFS="$TAB" read -r _d _o _w <<< "${PLAN[$i]}"
+    if [[ -z "$least" ]] || (( ${_w:-0} < least )); then least=${_w:-0}; lightest=$i; fi
+  done
+  IFS="$TAB" read -r d o w <<< "${PLAN[$lightest]}"
+  PLAN[$lightest]="${d}${TAB}${o:+$o }${EXTRA_OPS}${TAB}${w}"
+  echo ">>> EXTRA_OPS 追加到卡${d}（当前最空，$(( ${w:-0} * ROUNDS / 60 )) 分钟）: $EXTRA_OPS"
+fi
+
+if [[ "$DRY_RUN" == "1" ]]; then
+  echo ">>> DRY_RUN=1，不起跑。逐卡将执行："
+  for row in "${PLAN[@]}"; do
+    IFS="$TAB" read -r d o w <<< "$row"
+    printf '    卡%s  %4s 分钟  OPS="%s"\n' "$d" "$(( ${w:-0} * ROUNDS / 60 ))" "$o"
+  done
+  exit 0
+fi
+
+ROOT="runs/campaign_${SWITCH}_$(date +%Y%m%d-%H%M%S)"
+mkdir -p "$ROOT"
+printf '%s\n' "${PLAN[@]}" > "$ROOT/plan.tsv"
+echo ">>> 产物: $(pwd)/$ROOT"
+
+declare -a PIDS=() CARDS=()
+for row in "${PLAN[@]}"; do
+  IFS="$TAB" read -r dev ops _w <<< "$row"
+  [[ -n "$ops" ]] || { echo ">>> 卡$dev 没有分到活，跳过"; continue; }
+  out="$ROOT/card${dev}"
+  mkdir -p "$out"
+  # AB_ROOT 让每卡有独立产物根：八个 ab_repeat.sh 同一秒启动，默认的时间戳目录名会撞
+  env "$VIS_KEY=$dev" ZL_DEVICE="$dev" ZL_ISOLATE_TUNER_CACHE=1 \
+      AB_ROOT="$out" SWITCH="$SWITCH" ROUNDS="$ROUNDS" \
+      CELL_TIMEOUT="$CELL_TIMEOUT" OPS="$ops" NO_COLOR=1 \
+      bash ab_repeat.sh > "$ROOT/card${dev}.log" 2>&1 &
+  PIDS+=($!); CARDS+=("$dev")
+  echo ">>> 卡$dev pid=$! ops=$ops"
+done
+
+(( ${#PIDS[@]} )) || { echo "!!! 没有一张卡分到活" >&2; exit 2; }
+echo ">>> ${#PIDS[@]} 张卡已启动，等待全部结束……"
+
+fail=0
+for i in "${!PIDS[@]}"; do
+  if wait "${PIDS[$i]}"; then
+    echo ">>> 卡${CARDS[$i]} 结束 rc=0"
+  else
+    rc=$?; fail=$((fail+1))
+    echo "!!! 卡${CARDS[$i]} 结束 rc=$rc —— 看 $ROOT/card${CARDS[$i]}.log"
+  fi
+done
+
+# 合并逐格记录：卡号加在最前面，便于事后按卡追
+{
+  first=1
+  for d in "${CARDS[@]}"; do
+    f="$ROOT/card${d}/status.tsv"
+    [[ -f "$f" ]] || continue
+    if (( first )); then printf 'device%s%s\n' "$TAB" "$(head -1 "$f")"; first=0; fi
+    tail -n +2 "$f" | sed "s/^/${d}${TAB}/"
+  done
+} > "$ROOT/status.tsv"
+
+echo
+echo "########## 汇总 ##########"
+echo ">>> 失败的卡: $fail / ${#PIDS[@]}"
+if [[ -s "$ROOT/status.tsv" ]]; then
+  echo ">>> 逐格记录: $(pwd)/$ROOT/status.tsv  ($(( $(wc -l < "$ROOT/status.tsv") - 1 )) 格)"
+  awk -F"$TAB" 'NR>1 && ($6+0)!=0 {n++} END{printf ">>> 非零退出的格: %d\n", n+0}' "$ROOT/status.tsv"
+  awk -F"$TAB" 'NR>1 {fb+=$10; ng+=$11; po+=$12}
+       END{printf ">>> 三个标记合计: replay_fallback=%d no_cudagraph=%d capture_poisoned=%d\n",
+           fb+0, ng+0, po+0}' "$ROOT/status.tsv"
+fi
+echo ">>> 逐算子报告在各卡的 summary.log 里；单独重出："
+echo "    python batch_summary.py --ab-rounds $ROOT/card<N>/<op> --label <op>"
+echo ">>> 注意：本批 ZL_ISOLATE_TUNER_CACHE=1，绝对延迟不可与共享缓存的批次互引"

partition.py

@@ -0,0 +1,145 @@
+#!/usr/bin/env python3
+"""把筛选通过的算子按实测耗时分到各张卡上。
+
+为什么不用 `run_batch.py` 的 hash 分片：它按算子名的 hash 取模，不看重量。S5000 上
+17 个算子的六轮 campaign，hash 分片关键路径 120 分钟、有一张卡空转；按耗时贪心是
+70 分钟。同一批活，差 50 分钟。
+
+贪心（最长优先塞进当前最空的卡）在这里够用，也不可能更好：`topk` 一个算子就要 70.6
+分钟，而理想均分是 45.4——**关键路径的下界由最重的那个算子决定，加卡不解决**。分片
+只能保证除它之外的卡不空等。
+
+权重来自 `screen.tsv` 的 `elapsed_before + elapsed_after`（**一对 = 2 格**的秒数）。
+N 轮是 2N 格，所以预估耗时是权重的 N 倍——不是 N/6 倍，这里错过一次，把 topk 的 70.6
+分钟报成了 11.8。缺列时退化成等权重，并在 stderr 说明——等权重的分片仍然
+比不分好，但别让人以为它按耗时排过。
+
+用法：
+    python partition.py --screen runs/ab_rlc_XXX/screen.tsv --devices 0,1,2,3
+    python partition.py --screen ... --devices 0,1 --verdict IR_CHANGED --format shell
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import sys
+from pathlib import Path
+from typing import Dict, List, Optional, Sequence, Tuple
+
+Item = Tuple[str, float]
+
+
+def plan(items: Sequence[Item], slots: Sequence[str]) -> Dict[str, List[Item]]:
+    """最长优先贪心：每次把最重的活放进当前总量最小的卡。
+
+    卡少于 1 张是调用方的错，直接抛；活比卡少时后面的卡就是空的，这不是错误。
+    """
+    if not slots:
+        raise ValueError("需要至少一张卡")
+    out: Dict[str, List[Item]] = {s: [] for s in slots}
+    load: Dict[str, float] = {s: 0.0 for s in slots}
+    # 同重量时按名字排，保证同一份输入每次得到同一个方案
+    for name, weight in sorted(items, key=lambda kv: (-kv[1], kv[0])):
+        target = min(slots, key=lambda s: (load[s], s))
+        out[target].append((name, weight))
+        load[target] += weight
+    return out
+
+
+def imbalance(assignment: Dict[str, List[Item]]) -> float:
+    """最慢的卡 / 理想均分。1.0 是完美；下界由最重的单个算子决定。"""
+    loads = [sum(w for _, w in v) for v in assignment.values()]
+    total = sum(loads)
+    if total <= 0:
+        return 1.0
+    return max(loads) / (total / len(loads))
+
+
+def read_screen(path: Path, verdict: str = "IR_CHANGED"
+                ) -> Tuple[List[Item], Dict[str, str], List[str]]:
+    """(算子, 一对耗时秒), op -> op_file, 被判定挡掉的算子。"""
+    with path.open(newline="", encoding="utf-8") as f:
+        rows = list(csv.DictReader(f, delimiter="\t"))
+    if not rows:
+        raise SystemExit(f"筛选表没有数据行: {path}")
+    items: List[Item] = []
+    files: Dict[str, str] = {}
+    skipped: List[str] = []
+    warned = False
+    for r in rows:
+        op = r.get("op", "")
+        if not op:
+            continue
+        if verdict and r.get("verdict") != verdict:
+            skipped.append(f"{op}({r.get('verdict', '?')})")
+            continue
+        try:
+            w = float(r["elapsed_before"]) + float(r["elapsed_after"])
+        except (KeyError, TypeError, ValueError):
+            w = 1.0
+            if not warned:
+                print("!!! 筛选表没有 elapsed_* 列，退化成等权重分片——"
+                      "分片仍然比不分好，但它没有按耗时排过", file=sys.stderr)
+                warned = True
+        items.append((op, w))
+        files[op] = r.get("op_file") or op
+    return items, files, skipped
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    p = argparse.ArgumentParser(description=__doc__.splitlines()[0],
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--screen", type=Path, required=True, help="ab_rlc_test.sh 的 screen.tsv")
+    p.add_argument("--devices", required=True, help="逗号分隔的设备列表，如 0,1,2,3")
+    p.add_argument("--verdict", default="IR_CHANGED",
+                   help="只取这个判定的算子；空字符串表示全要")
+    # 权重是一对（2 格）的秒数；N 轮是 2N 格，所以预估耗时是权重的 N 倍
+    p.add_argument("--rounds", type=int, default=6, help="仅用于打印预估耗时")
+    p.add_argument("--format", choices=("human", "shell"), default="human",
+                   help="shell 输出每卡一行 `<device>\\t<OPS>`，供驱动脚本读")
+    args = p.parse_args(argv)
+
+    if not args.screen.is_file():
+        raise SystemExit(f"筛选表不存在: {args.screen}")
+    devices = [d.strip() for d in args.devices.split(",") if d.strip()]
+    if not devices:
+        raise SystemExit("--devices 为空")
+
+    items, files, skipped = read_screen(args.screen, args.verdict)
+    if not items:
+        raise SystemExit(f"筛选表里没有 verdict={args.verdict} 的算子——"
+                         "先确认筛选跑过，且判据没有把所有算子都挡掉")
+
+    assignment = plan(items, devices)
+    if args.format == "shell":
+        # 第三列是该卡的权重秒数，供驱动脚本决定往哪张卡追加算子——按负载选，不是
+        # 按算子个数：一个重算子能顶三个轻的。
+        # 显式写 "\n" 而不用 print：Windows 上 print 会发 CRLF，读的一侧会把 \r
+        # 带进最后一个字段，数值比较就废了。
+        for d in devices:
+            ops = " ".join(f"{op}:{files[op]}" for op, _ in assignment[d])
+            load = sum(w for _, w in assignment[d])
+            sys.stdout.buffer.write(f"{d}\t{ops}\t{load:.0f}\n".encode())
+        sys.stdout.buffer.flush()
+        return 0
+
+    total = sum(w for _, w in items) * args.rounds
+    print(f">>> {len(items)} 个算子 / {len(devices)} 张卡 / {args.rounds} 轮")
+    if skipped:
+        print(f">>> 判定挡掉 {len(skipped)} 个: {', '.join(skipped)}")
+    for d in devices:
+        load = sum(w for _, w in assignment[d]) * args.rounds
+        names = " ".join(op for op, _ in assignment[d]) or "（空）"
+        print(f"    卡{d}  {load/60:6.1f} 分钟  {names}")
+    crit = max(sum(w for _, w in v) for v in assignment.values()) * args.rounds
+    heaviest = max(items, key=lambda kv: kv[1])
+    print(f">>> 单卡串行 {total/3600:.2f} 小时；{len(devices)} 卡关键路径 "
+          f"{crit/60:.1f} 分钟，失衡 {imbalance(assignment):.2f}x")
+    print(f">>> 下界由最重的算子决定：{heaviest[0]} 单独就要 "
+          f"{heaviest[1]*args.rounds/60:.1f} 分钟，加卡不解决")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

run_pytest.sh

@@ -114,6 +114,12 @@ preflight "FlagGems directory exists: $FLAGGEMS_DIR" test -d "$FLAGGEMS_DIR"
 # The node's file half; the ::test_<fn> half is left to pytest, which reports a
 # missing function clearly ("not found: ...") while a missing FILE does not.
 preflight "benchmark file exists: ${TEST_FILE%%::*}" test -f "${TEST_FILE%%::*}"
+# The interpreter, not just the paths. triton and flag_gems normally come from a
+# PYTHONPATH pointing at a compiler build and a source checkout, so a shell that
+# merely reaches the right files can still be the wrong Python. Without this the
+# failure surfaces per cell as `pytest exit 4` from a conftest ImportError, and a
+# multi-round campaign burns every cell before anyone reads a log.
+preflight "interpreter imports triton + flag_gems ($(command -v python))"   sh -c 'python -c "import triton, flag_gems" 2>/dev/null'
 if [[ -n "$SHAPE_FILE" ]]; then
   preflight "shape file exists: $SHAPE_FILE" test -f "$SHAPE_FILE"
 fi

telemetry.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""采样设备时钟/温度/功耗，跨后端归一化成同一组列。
+
+为什么要有它：六轮 A/B 会声称单侧噪声在 0.01% 量级。降频和热漂移足以移动比这大
+得多的延迟，而且在 run.log 里不留任何痕迹——报告照样干净。NV 上复现 PR763 的那次
+全程 `nvidia-smi -l 5` 写 clocks.csv，S5000 这边一直没有对应的东西。
+
+输出每设备每次采样一行 TSV：
+
+    timestamp  index  sm_clock_mhz  mem_clock_mhz  temp_c  power_w  util_pct
+
+列在 cuda 与 musa 上完全一致，所以两边的 campaign 可以并排读。采样器不存在时安静
+退出：拿不到遥测是缺一份证据，不是让整批测试失败的理由。
+
+用法：
+    python telemetry.py --backend musa --device 7 --interval 5 --run > telemetry.tsv
+    python telemetry.py --backend cuda            # 只打印将要执行的采样命令
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import subprocess
+import sys
+import time
+from typing import List, Optional
+
+FIELDS = ("index", "sm_clock_mhz", "mem_clock_mhz", "temp_c", "power_w", "util_pct")
+
+
+def resolve_device(backend, explicit: Optional[str]) -> str:
+    """显式设备优先，否则从后端自己的可见性变量取。
+
+    单值才用得上：`MUSA_VISIBLE_DEVICES=7` 指一张卡，`0,1,2` 指一组，后者不传
+    `-i`、采全部，比挑第一张更诚实。
+    """
+    if explicit:
+        return explicit.strip()
+    for key in getattr(backend, "visibility_env_keys", ()):
+        raw = os.environ.get(key, "").strip()
+        if raw and "," not in raw:
+            return raw
+    return ""
+
+
+def sample_once(backend, argv: List[str]) -> List[dict]:
+    try:
+        out = subprocess.run(argv, capture_output=True, text=True, timeout=30)
+    except (OSError, subprocess.SubprocessError):
+        return []
+    return backend.parse_telemetry(out.stdout or "")
+
+
+def run(backend, device: str, interval: float, out=sys.stdout) -> int:
+    argv = backend.telemetry_command(device)
+    if not argv:
+        print(f"[telemetry] backend={backend.name} 没有采样器；本批不记时钟/温度",
+              file=sys.stderr, flush=True)
+        return 0
+    if not sample_once(backend, argv):
+        # 第一次就取不到，说明工具缺失或没权限。不要留一个空文件让人以为采过了
+        print(f"[telemetry] {argv[0]} 取不到读数；本批不记时钟/温度",
+              file=sys.stderr, flush=True)
+        return 0
+    print("timestamp\t" + "\t".join(FIELDS), file=out, flush=True)
+    while True:
+        ts = time.strftime("%Y-%m-%dT%H:%M:%S")
+        for row in sample_once(backend, argv):
+            print(ts + "\t" + "\t".join(row.get(f, "") for f in FIELDS),
+                  file=out, flush=True)
+        time.sleep(interval)
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    p = argparse.ArgumentParser(description=__doc__,
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--backend", default=os.environ.get("ZL_BACKEND", "cuda"))
+    p.add_argument("--device", default=os.environ.get("ZL_DEVICE", ""))
+    p.add_argument("--interval", type=float, default=5.0)
+    p.add_argument("--run", action="store_true",
+                   help="持续采样直到被杀；不给则只打印采样命令")
+    args = p.parse_args(argv)
+
+    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+    from backends import get_backend
+    backend = get_backend(args.backend)
+    device = resolve_device(backend, args.device)
+    if not args.run:
+        print(" ".join(backend.telemetry_command(device)))
+        return 0
+    try:
+        return run(backend, device, args.interval)
+    except KeyboardInterrupt:
+        return 0
+    except BrokenPipeError:
+        # 读端先走（管给 head，或 campaign 结束时收掉）。这是正常终止，不该在
+        # telemetry.err 里留一段 traceback 让人以为遥测本身坏了。按 CPython 的
+        # 惯用做法把 stdout 指到 devnull——只吞异常不够，解释器退出时还会 flush
+        # 一次，再抛一条 "Exception ignored" 到 stderr。
+        try:
+            devnull = os.open(os.devnull, os.O_WRONLY)
+            os.dup2(devnull, sys.stdout.fileno())
+        except Exception:
+            pass
+        return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

test_backends.py

@@ -541,5 +541,198 @@ class CaliberGuardTests(unittest.TestCase):
                 len(report_ab(a / "run.log", b / "run.log", "op", strict=False)), 1)
 
 
+class TelemetryTests(unittest.TestCase):
+    """采样器命令与解析。两边归一到同一组列，campaign 才能并排读。"""
+
+    # 取自 S5000 实机：值带单位，"Power Draw " 的键名有尾随空格
+    MUSA_JSON = """
+{
+    "Attached GPUs": "8",
+    "GPU": [
+        {
+            "Index": "7",
+            "Temperature": {"GPU Current Temp": "36C"},
+            "Clocks": {"Graphics": "1888MHz", "Memory": "2500MHz", "Video": "0MHz"},
+            "Power Readings": {"Power Draw ": "477.68W", "Current Power Limit": "950.00W"},
+            "Utilization": {"Gpu": "87%", "Memory": "14%"}
+        }
+    ]
+}
+"""
+    CUDA_CSV = "0, 1980, 2619, 41, 122.35, 63\n"
+
+    def test_musa_parses_units_off_values(self):
+        rows = get_backend("musa").parse_telemetry(self.MUSA_JSON)
+        self.assertEqual(rows, [{
+            "index": "7", "sm_clock_mhz": "1888", "mem_clock_mhz": "2500",
+            "temp_c": "36", "power_w": "477.68", "util_pct": "87"}])
+
+    def test_cuda_parses_csv(self):
+        rows = get_backend("cuda").parse_telemetry(self.CUDA_CSV)
+        self.assertEqual(rows, [{
+            "index": "0", "sm_clock_mhz": "1980", "mem_clock_mhz": "2619",
+            "temp_c": "41", "power_w": "122.35", "util_pct": "63"}])
+
+    def test_both_backends_yield_the_same_columns(self):
+        import telemetry
+        a = get_backend("musa").parse_telemetry(self.MUSA_JSON)[0]
+        b = get_backend("cuda").parse_telemetry(self.CUDA_CSV)[0]
+        self.assertEqual(set(a), set(b))
+        self.assertEqual(set(a), set(telemetry.FIELDS))
+
+    def test_garbage_parses_to_nothing_rather_than_raising(self):
+        for backend in ("musa", "cuda"):
+            for raw in ("", "no such tool", "{broken json"):
+                self.assertEqual(get_backend(backend).parse_telemetry(raw), [])
+
+    def test_no_device_drops_the_flag_not_just_the_value(self):
+        # 留下裸 "-i" 会让采样器直接失败，比采全部更糟
+        argv = get_backend("musa").telemetry_command("")
+        self.assertNotIn("-i", argv)
+        self.assertIn("--json", argv)
+        self.assertEqual(get_backend("musa").telemetry_command("7")[-2:], ["-i", "7"])
+
+    def test_backend_without_sampler_returns_empty(self):
+        from backends.base import DeviceBackend
+        self.assertEqual(DeviceBackend().telemetry_command("0"), [])
+
+    def test_device_comes_from_the_visibility_var_when_unpinned(self):
+        import telemetry
+        backend = get_backend("musa")
+        old = os.environ.get("MUSA_VISIBLE_DEVICES")
+        try:
+            os.environ["MUSA_VISIBLE_DEVICES"] = "7"
+            self.assertEqual(telemetry.resolve_device(backend, ""), "7")
+            # 一组卡不挑第一张，交给采样器采全部
+            os.environ["MUSA_VISIBLE_DEVICES"] = "0,1,2"
+            self.assertEqual(telemetry.resolve_device(backend, ""), "")
+            self.assertEqual(telemetry.resolve_device(backend, "3"), "3")
+        finally:
+            if old is None:
+                os.environ.pop("MUSA_VISIBLE_DEVICES", None)
+            else:
+                os.environ["MUSA_VISIBLE_DEVICES"] = old
+
+
+class ForeignVendorModuleTests(unittest.TestCase):
+    """benchmark 的可用性判据常写成 `try: import <厂商实现>; AVAILABLE = True`。
+
+    import 只证明那个 .py 在磁盘上——它不碰设备，所以在任何后端上都成功，gate 于是
+    形同虚设。router_gemm 就这样让 NVIDIA Hopper 的 kernel 跑在了 MUSA 卡上，最后
+    死在 JIT 深处的 "dynamic_func() missing BLOCK_M ..."，报错完全指不向真正的原因。
+    """
+
+    P = "flag_gems.runtime.backend._"
+
+    def _mods(self, *names):
+        return {self.P + n: object() for n in names} | {"torch": object()}
+
+    def test_only_own_vendor_is_clean(self):
+        from _device_setup_plugin import foreign_vendor_modules
+        mods = self._mods("mthreads", "mthreads.ops.mm")
+        self.assertEqual(foreign_vendor_modules("mthreads", mods), [])
+
+    def test_foreign_vendor_is_reported_once_per_root(self):
+        from _device_setup_plugin import foreign_vendor_modules
+        # 同一厂商的多个子模块只报一条
+        mods = self._mods("mthreads.ops.mm",
+                          "nvidia.hopper.ops.mm", "nvidia.ops.mm", "nvidia")
+        self.assertEqual(foreign_vendor_modules("mthreads", mods), ["nvidia"])
+
+    def test_several_foreign_vendors(self):
+        from _device_setup_plugin import foreign_vendor_modules
+        mods = self._mods("mthreads", "nvidia.ops.mm", "cambricon.ops.mm")
+        self.assertEqual(foreign_vendor_modules("mthreads", mods),
+                         ["cambricon", "nvidia"])
+
+    def test_a_vendor_whose_name_extends_ours_is_still_foreign(self):
+        # 按前缀比会把 _mthreads_extra 认成自家
+        from _device_setup_plugin import foreign_vendor_modules
+        mods = self._mods("mthreads", "mthreads_extra.ops")
+        self.assertEqual(foreign_vendor_modules("mthreads", mods),
+                         ["mthreads_extra"])
+
+    def test_unrelated_modules_are_ignored(self):
+        from _device_setup_plugin import foreign_vendor_modules
+        mods = {"torch": object(), "triton": object(),
+                "flag_gems.ops.mm": object()}
+        self.assertEqual(foreign_vendor_modules("mthreads", mods), [])
+
+
+class PartitionTests(unittest.TestCase):
+    """按耗时把算子分到各卡。hash 分片在这批活上比它慢 50 分钟。"""
+
+    HEADER = "op\top_file\telapsed_before\telapsed_after\tverdict"
+
+    def _load(self, assignment):
+        return sorted(sum(w for _, w in v) for v in assignment.values())
+
+    def _screen(self, tmp, rows, header=None):
+        p = Path(tmp, "screen.tsv")
+        p.write_text((header or self.HEADER) + "\n" + "\n".join(rows) + "\n",
+                     encoding="utf-8")
+        return p
+
+    def test_heavy_first_keeps_cards_even(self):
+        from partition import plan
+        items = [("a", 10), ("b", 9), ("c", 8), ("d", 7), ("e", 6), ("f", 5)]
+        self.assertEqual(self._load(plan(items, ["0", "1", "2"])), [15, 15, 15])
+
+    def test_one_item_heavier_than_the_ideal_share_sets_the_floor(self):
+        # topk 单个 70 分钟、理想均分 45——加卡不解决，分片只保证别人不空等
+        from partition import imbalance, plan
+        items = [("topk", 70)] + [(f"op{i}", 5) for i in range(10)]
+        a = plan(items, [str(i) for i in range(8)])
+        self.assertEqual(max(sum(w for _, w in v) for v in a.values()), 70)
+        self.assertGreater(imbalance(a), 1.0)
+
+    def test_more_cards_than_operators_leaves_some_empty(self):
+        from partition import plan
+        a = plan([("a", 1), ("b", 1)], ["0", "1", "2", "3"])
+        self.assertEqual(sum(1 for v in a.values() if not v), 2)
+
+    def test_same_input_gives_the_same_plan(self):
+        from partition import plan
+        items = [("a", 5), ("b", 5), ("c", 5), ("d", 5)]
+        first = plan(items, ["0", "1"])
+        for _ in range(5):
+            self.assertEqual(plan(list(reversed(items)), ["0", "1"]), first)
+
+    def test_no_cards_is_the_callers_error(self):
+        from partition import plan
+        with self.assertRaises(ValueError):
+            plan([("a", 1)], [])
+
+    def test_perfect_balance_is_one(self):
+        from partition import imbalance, plan
+        self.assertEqual(imbalance(plan([("a", 5), ("b", 5)], ["0", "1"])), 1.0)
+
+    def test_verdict_filters_and_reports_what_it_dropped(self):
+        from partition import read_screen
+        with tempfile.TemporaryDirectory() as tmp:
+            f = self._screen(tmp, ["a\ta\t10\t10\tIR_CHANGED",
+                                   "b\tbb\t5\t5\tNO_IR_DIFF",
+                                   "c\tc\t1\t1\tNO_TRITON"])
+            items, files, skipped = read_screen(f)
+            self.assertEqual(items, [("a", 20.0)])
+            self.assertEqual(files["a"], "a")
+            self.assertEqual(len(skipped), 2)
+
+    def test_op_file_defaults_to_op_name(self):
+        from partition import read_screen
+        with tempfile.TemporaryDirectory() as tmp:
+            f = self._screen(tmp, ["a\t\t1\t1\tIR_CHANGED"])
+            _, files, _ = read_screen(f)
+            self.assertEqual(files["a"], "a")
+
+    def test_missing_elapsed_falls_back_to_equal_weights(self):
+        from partition import read_screen
+        with tempfile.TemporaryDirectory() as tmp:
+            f = self._screen(tmp, ["a\ta\tIR_CHANGED", "b\tb\tIR_CHANGED"],
+                             header="op\top_file\tverdict")
+            items, _, _ = read_screen(f)
+            self.assertEqual([w for _, w in items], [1.0, 1.0])
+
+
 if __name__ == "__main__":
     unittest.main()