feat: report a per-operator speedup range across two runs

The speedup column in a result table compares gems against torch, not one build
against another, so an A/B needs the two runs paired by shape. batch_summary
gains an --ab mode that indexes both run.logs by operator, dtype and size
detail, computes before/after per pair, and prints the per-shape rows plus the
range. Only rows that succeeded on both sides with a positive latency count, so
an incomplete pairing cannot masquerade as a complete range. When either side
carries a BENCHMARK_DIRECT_NO_CUDAGRAPH marker the range is annotated, because
a fallback row was timed under a different regime than a captured one.

ab_rlc_test.sh drives the pairing for the FlagTree RemoveLayoutConversions
switch the way ab_fold_test.sh does for the FlagGems one: per operator, run with
the enhancement off while recording autotune configs, then replay those configs
with it on, so the two sides differ by that switch alone. Each operator
therefore leaves exactly two run.log files, and the summary of each pair is
printed as it completes.

Checked on an S5000 against gelu_backward over five shapes: both sides captured
with no fallback, five rows paired, range -0.43% to +0.09%.

_cudagraph_plugin.py

@@ -1,4 +1,4 @@
-"""pytest plugin: swap `triton.testing.do_bench` for cudagraph-based timing.
+"""pytest plugin: swap `triton.testing.do_bench` for graph-capture timing.
 
 Loaded via `-p _cudagraph_plugin` in run_pytest.sh's PLUGINS list (comment that
 line out to time with plain do_bench). Monkey-patches do_bench →
@@ -8,9 +8,18 @@ fall back to plain do_bench and print a BENCHMARK_DIRECT_NO_CUDAGRAPH marker
 into the log. Before capturing, fn is explicitly warmed up for the caller's
 warmup time budget (do_bench_cudagraph's own 5-iter warmup is too short to
 settle autotune/JIT for the MoE kernels, which makes the first captured
-measurement unstable run-to-run). The capture itself runs under a scratch CUDA
+measurement unstable run-to-run). The capture itself runs under a scratch RNG
-RNG state so that an aborted capture cannot poison the process-wide generator
+state so that an aborted capture cannot poison the process-wide generator
 (rationale: _capture_with_scratch_rng).
+
+Everything here goes through Triton's active device interface rather than
+torch.cuda. FlagTree's do_bench is already written that way, but its
+do_bench_cudagraph is not: it hardcodes torch.cuda.stream/Event/CUDAGraph, so
+on a non-CUDA backend the graph path raised and every measurement silently fell
+back to plain do_bench. On MThreads the interface is torch_musa, which supplies
+Stream, Event, synchronize, graph and MUSAGraph, so the same capture works once
+the calls are taken from the interface. CUDA still calls upstream's
+do_bench_cudagraph, so its numbers are unchanged.
 """
 
 from __future__ import annotations
@@ -24,8 +33,33 @@ import triton.testing as _tt
 _ORIGINAL_DO_BENCH = _tt.do_bench
 
 
-def _is_fatal_cuda_error(exc: Exception) -> bool:
+def _device_interface():
-    """True for CUDA context errors (illegal address, OOM, device-side assert):
+    """Triton's active device interface: torch.cuda on NVIDIA, torch_musa on MThreads."""
+    from triton.runtime import driver
+
+    return driver.active.get_device_interface()
+
+
+def _is_cuda(di) -> bool:
+    import torch
+
+    return di is torch.cuda
+
+
+def _graph_class(di):
+    """The interface's graph type: CUDAGraph, MUSAGraph, or whatever it names."""
+    for name in ("CUDAGraph", "MUSAGraph"):
+        graph = getattr(di, name, None)
+        if graph is not None:
+            return graph
+    for name in dir(di):
+        if name.endswith("Graph"):
+            return getattr(di, name)
+    raise RuntimeError("device interface exposes no graph type")
+
+
+def _is_fatal_device_error(exc: Exception) -> bool:
+    """True for device context errors (illegal address, OOM, device-side assert):
     the context is poisoned, so re-raise rather than fall back and record a bogus
     latency as success.
 
@@ -47,8 +81,12 @@ def _is_fatal_cuda_error(exc: Exception) -> bool:
     return any(m in msg for m in markers)
 
 
-def _warmup_before_capture(fn, warmup_ms, grad_to_none):
+# Retained under its historical name; the marker list is backend independent.
-    """Warm fn up before cudagraph capture.
+_is_fatal_cuda_error = _is_fatal_device_error
+
+
+def _warmup_before_capture(di, fn, warmup_ms, grad_to_none):
+    """Warm fn up before graph capture.
 
     do_bench_cudagraph only warms up 5 iterations internally — too few for the
     MoE kernels, where the first calls trigger Triton autotune config
@@ -66,8 +104,6 @@ def _warmup_before_capture(fn, warmup_ms, grad_to_none):
     """
     import time
 
-    import torch
-
     def _run():
         if grad_to_none is not None:
             for x in grad_to_none:
@@ -76,23 +112,23 @@ def _warmup_before_capture(fn, warmup_ms, grad_to_none):
 
     # First call absorbs autotune/JIT compilation (can take seconds); discard it.
     _run()
-    torch.cuda.synchronize()
+    di.synchronize()
     # Size the warmup loop from one timed steady-state run. dt_ms includes
     # launch overhead, so n errs low for tiny kernels; warmup needn't be exact.
     t0 = time.perf_counter()
     _run()
-    torch.cuda.synchronize()
+    di.synchronize()
     dt_ms = (time.perf_counter() - t0) * 1000.0
     budget = warmup_ms if (warmup_ms and warmup_ms > 0) else 25.0
     n = int(budget / dt_ms) if dt_ms > 0 else 25
     n = max(5, min(n, 200))
     for _ in range(n):
         _run()
-    torch.cuda.synchronize()
+    di.synchronize()
 
 
-def _capture_with_scratch_rng(call):
+def _capture_with_scratch_rng(di, call):
-    """Run a graph capture under a throwaway CUDA RNG state.
+    """Run a graph capture under a throwaway device RNG state.
 
     torch.cuda.graph's capture_begin unconditionally registers the default CUDA
     RNG generator and sets its capturing flag; when a capture aborts midway the
@@ -104,12 +140,21 @@ def _capture_with_scratch_rng(call):
     before capturing and restore the original state object afterwards, success
     or failure. The real generator never takes part in a capture, so poisoning
     can only land on the discarded scratch state, and _seed_plugin determinism
-    is left untouched."""
+    is left untouched.
+
+    An interface that does not expose default_generators has no such registration
+    to protect against (torch_musa is one), so the capture simply runs.
+    """
     import torch
-    dev = torch.cuda.current_device()
+
-    gen = torch.cuda.default_generators[dev]
+    generators = getattr(di, "default_generators", None)
+    if not generators:
+        return call()
+    gen = generators[di.current_device()]
+    if not hasattr(gen, "graphsafe_get_state"):
+        return call()
     saved = gen.graphsafe_get_state()
-    scratch = torch.Generator(device="cuda")
+    scratch = torch.Generator(device=torch.device(gen.device))
     scratch.manual_seed(gen.initial_seed())
     gen.graphsafe_set_state(scratch)
     try:
@@ -118,24 +163,85 @@ def _capture_with_scratch_rng(call):
         gen.graphsafe_set_state(saved)
 
 
+def _do_bench_graph(di, fn, rep, grad_to_none, quantiles, return_mode, n_retries=10):
+    """do_bench_cudagraph, taken from the device interface instead of torch.cuda.
+
+    A step-for-step port of upstream's implementation: estimate one iteration,
+    unroll that many calls into a captured graph so host launch overhead drops
+    out, then time independent replays. Upstream is used verbatim on CUDA; this
+    exists because upstream names torch.cuda directly and so cannot run on any
+    other backend.
+
+    The replay timings are handed to _summarize_statistics as a plain list of
+    floats, which is what both do_bench and upstream's do_bench_cudagraph pass:
+    this fork's helper runs Python's statistics module over them, so wrapping
+    them in a tensor makes "mean" raise and makes a quantile request return
+    tensors that later fail to serialize.
+    """
+    with di.stream(di.Stream()):
+        fn()
+        if grad_to_none is not None:
+            for x in grad_to_none:
+                x.detach_()
+                x.requires_grad_(True)
+                x.grad = None
+        start_event = di.Event(enable_timing=True)
+        end_event = di.Event(enable_timing=True)
+        start_event.record()
+        for _ in range(5):
+            fn()
+        end_event.record()
+        di.synchronize()
+        estimate_ms = start_event.elapsed_time(end_event) / 5
+        n_repeat = 1000 if estimate_ms == 0 else max(1, int(rep / estimate_ms))
+        graph = _graph_class(di)()
+        with di.graph(graph):
+            for _ in range(n_repeat):
+                if grad_to_none is not None:
+                    for x in grad_to_none:
+                        x.grad = None
+                fn()
+        di.synchronize()
+        ret = []
+        for _ in range(n_retries):
+            start_event = di.Event(enable_timing=True)
+            end_event = di.Event(enable_timing=True)
+            start_event.record()
+            graph.replay()
+            end_event.record()
+            di.synchronize()
+            ret.append(start_event.elapsed_time(end_event) / n_repeat)
+        del graph
+        return _tt._summarize_statistics(ret, quantiles, return_mode)
+
+
 def _patched_do_bench(fn, warmup=25, rep=100, grad_to_none=None,
                       quantiles=None, return_mode="mean"):
-    """Drop-in replacement for triton.testing.do_bench using cudagraph capture."""
+    """Drop-in replacement for triton.testing.do_bench using graph capture."""
     op = os.environ.get("FLAGGEMS_PERF_CURRENT_OP", "<unknown>")
-    phase = "sync"  # which step rejected cudagraph, for the fallback marker
+    phase = "sync"  # which step rejected graph capture, for the fallback marker
     try:
-        # do_bench_cudagraph runs fn on a fresh side stream without syncing
+        di = _device_interface()
-        # with the default stream first; input tensors produced just before
+        # The graph path runs fn on a fresh side stream without syncing with the
-        # (quantize/topk in the input iter) may still be in flight, and racing
+        # default stream first; input tensors produced just before (quantize/topk
-        # on them intermittently kills the context (illegal instruction from a
+        # in the input iter) may still be in flight, and racing on them
-        # device trap on garbage indices). Sync before switching streams.
+        # intermittently kills the context (illegal instruction from a device
-        import torch
+        # trap on garbage indices). Sync before switching streams.
-        torch.cuda.synchronize()
+        di.synchronize()
         phase = "warmup"
         # Settle autotune/JIT before capture (rationale: _warmup_before_capture).
-        _warmup_before_capture(fn, warmup, grad_to_none)
+        _warmup_before_capture(di, fn, warmup, grad_to_none)
         phase = "capture"
-        return _capture_with_scratch_rng(lambda: _tt.do_bench_cudagraph(
+        if _is_cuda(di):
+            return _capture_with_scratch_rng(di, lambda: _tt.do_bench_cudagraph(
+                fn,
+                rep=rep,
+                grad_to_none=grad_to_none,
+                quantiles=quantiles,
+                return_mode=return_mode,
+            ))
+        return _capture_with_scratch_rng(di, lambda: _do_bench_graph(
+            di,
             fn,
             rep=rep,
             grad_to_none=grad_to_none,
@@ -145,7 +251,7 @@ def _patched_do_bench(fn, warmup=25, rep=100, grad_to_none=None,
     except Exception as exc:
         # Re-raise genuine device failures; anything else falls back to plain
         # do_bench, with the failing phase (sync/warmup/capture) in the marker.
-        if _is_fatal_cuda_error(exc):
+        if _is_fatal_device_error(exc):
             raise
         reason = type(exc).__name__
         detail = " ".join(str(exc).split())[:160]
@@ -168,5 +274,10 @@ def _patched_do_bench(fn, warmup=25, rep=100, grad_to_none=None,
 def pytest_configure(config):
     from _term_style import tag
     _tt.do_bench = _patched_do_bench
-    print(f"{tag('[cudagraph-plugin]')} triton.testing.do_bench → do_bench_cudagraph",
+    try:
+        backend = getattr(_device_interface(), "__name__", "?")
+    except Exception:  # pragma: no cover - reported, never fatal at configure
+        backend = "?"
+    print(f"{tag('[cudagraph-plugin]')} triton.testing.do_bench → graph capture "
+          f"via {backend}",
           file=sys.stderr, flush=True)

ab_rlc_test.sh

@@ -0,0 +1,68 @@
+#!/bin/bash
+# A/B 对比：FlagTree RemoveLayoutConversions 增强关 vs 开，逐算子两轮。
+#
+# 口径：每个算子先跑关（record autotune config），再用 REPLAY_FROM 回放同一套
+# config 跑开，两侧只差 FLAGTREE_<VENDOR>_RLC_ENHANCE 这一个开关。产物沿用
+# run_pytest.sh 的目录结构，所以每个算子恰好留下两份 run.log：
+#
+#   runs/<op>_<时间戳>/run.log   关（before）
+#   runs/<op>_<时间戳>/run.log   开（after）
+#
+# 跑完用 batch_summary.py --ab <before> <after> 得到该算子的 speedup 区间。
+#
+# 用法：
+#   OPS="softmax gelu_backward" bash ab_rlc_test.sh
+#   OPS="sort:perf_sort topk" bash ab_rlc_test.sh     # 冒号分隔 OP:OP_FILE
+#
+# 注意：RLC 开关名带厂商前缀，用 VENDOR 指定（默认 MUSA）。套用前先确认开关有效：
+# grep 该开关名于 FlagTree 源码应有命中，否则两侧跑的是同一份代码。
+set -uo pipefail
+
+cd "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+VENDOR="${VENDOR:-MUSA}"
+ENHANCE_VAR="FLAGTREE_${VENDOR}_RLC_ENHANCE"
+
+# 空格分隔的算子表。元素为 OP 或 OP:OP_FILE（省略时 OP_FILE=OP）。
+OPS="${OPS:-softmax}"
+
+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
+
+{
+  echo ">>> vendor=$VENDOR  switch=$ENHANCE_VAR  ops=$OPS"
+  for entry in $OPS; do
+    op="${entry%%:*}"
+    op_file="${entry#*:}"
+    [[ "$op_file" == "$entry" ]] && op_file="$op"
+
+    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)
+
+    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)
+
+    echo "########## $op | summary ##########"
+    python batch_summary.py --ab "$BEFORE/run.log" "$AFTER/run.log" --label "$op" \
+      || echo "!!! summary failed: $op"
+  done
+  echo "########## ALL DONE ##########"
+} 2>&1 | tee "$LOG"
+
+# 去掉 ANSI 转义以便 grep/diff（终端输出保留颜色）
+sed -i -E $'s/\x1b\\[[0-9;]*[A-Za-z]//g' "$LOG"
+
+echo ">>> 总 log: $(pwd)/$LOG"
+echo ">>> 每算子两份 run.log 在 runs/<op>_<时间戳>/ 下，按时间先后即 before / after"

batch_summary.py

@@ -165,12 +165,82 @@ def summarize(batch_dir: Path, out_path: Optional[Path] = None) -> Path:
     return out_path
 
 
+def ab_rows(before_text: str, after_text: str) -> List[Dict[str, str]]:
+    """把两次 run.log 的测量行按 (op, dtype, size_detail) 配对。
+
+    speedup 列是 gems 对 torch，不是前后对比，所以这里用 gems_lat 自己算
+    before/after - 1。只有两侧都 SUCCESS 且延迟为正的行才计入；对不上的行
+    单独返回，避免用不完整的配对报一个看起来完整的区间。
+    """
+    def index(text: str) -> Dict[Tuple[str, str, str], float]:
+        out: Dict[Tuple[str, str, str], float] = {}
+        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:
+                out[(row.get("fg_op_name", ""), row.get("dtype", ""),
+                     row.get("size_detail", ""))] = lat
+        return out
+
+    before, after = index(before_text), index(after_text)
+    paired: List[Dict[str, str]] = []
+    for key in sorted(before.keys() & after.keys()):
+        op, dtype, size = key
+        b, a = before[key], after[key]
+        paired.append({
+            "fg_op_name": op, "dtype": dtype, "size_detail": size,
+            "before_lat": f"{b:.6f}", "after_lat": f"{a:.6f}",
+            "speedup_percent": f"{(b / a - 1.0) * 100.0:.2f}",
+        })
+    return paired
+
+
+def report_ab(before: Path, after: Path, label: str = "") -> List[Dict[str, str]]:
+    """打印一个算子的前后对比：逐行 speedup 与整体区间。"""
+    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)
+    name = label or (rows[0]["fg_op_name"] if rows else "?")
+    if not rows:
+        print(f"[ab] {name}: 没有可配对的测量行"
+              f"（before={before} after={after}）")
+        return rows
+    values = [float(r["speedup_percent"]) for r in rows]
+    for r in rows:
+        print(f"  {r['dtype']:<16} {float(r['speedup_percent']):+7.2f}%  "
+              f"{r['before_lat']} -> {r['after_lat']}  {r['size_detail']}")
+    fallback = ""
+    if DIRECT_NO_CUDAGRAPH_RE.search(before_text) or \
+            DIRECT_NO_CUDAGRAPH_RE.search(after_text):
+        # 回退行用的是 do_bench 口径，与 cudagraph 行不同表可比性
+        fallback = "  [含 BENCHMARK_DIRECT_NO_CUDAGRAPH 回退行]"
+    print(f"[ab] {name}: {len(rows)} 行配对，speedup "
+          f"{min(values):+.2f}% ~ {max(values):+.2f}%{fallback}")
+    return rows
+
+
 def main() -> None:
     p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
-    p.add_argument("batch_dir", type=Path, help="run_batch.py 的产物目录")
+    p.add_argument("batch_dir", type=Path, nargs="?",
+                   help="run_batch.py 的产物目录")
     p.add_argument("-o", "--output", type=Path, default=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 报告里的算子名")
     args = p.parse_args()
+    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)
+        return
+    if args.batch_dir is None:
+        raise SystemExit("需要 batch_dir，或用 --ab BEFORE AFTER")
     if not args.batch_dir.is_dir():
         raise SystemExit(f"目录不存在: {args.batch_dir}")
     out = summarize(args.batch_dir, args.output)

campaigns/musa_rlc_product/README.md

@@ -1,103 +0,0 @@
-# S5000 MUSA RLC 产品验证 campaign
-
-在 Moore Threads S5000 上，经由 FlagGems public dispatch，测量 FlagTree
-`RemoveLayoutConversions`（RLC）增强对 32 个冻结 workload 的影响。
-
-本目录只提供 campaign 层：契约、冻结、执行、判定、重记账。单算子插件、shape、
-inventory 仍然复用仓库根目录的既有能力。
-
-## 组成
-
-| 文件 | 作用 |
-|---|---|
-| `manifest/musa-rlc-32-product-v1.json` | 冻结契约：源码提交、设备、cell、计时口径、32 个 workload |
-| `prepare_musa_rlc_product.py` | 校验 manifest 与 `ops/inventory_main.csv`，冻结不可变 task matrix |
-| `run_musa_rlc_product.py` | 串行执行 task matrix，产出逐 cell、逐 task、campaign 三级收据 |
-| `run_pytest_musa_rlc_product.sh` | 单个算子单个 cell 的 worker，挂载根目录插件 |
-| `reaccount_musa_rlc_topology.py` | 对已封存 run 按新口径重记账，不占用设备 |
-| `tests/` | 上述四者的单元测试 |
-
-根目录的 `_musa_benchmark_compat_plugin.py` 属于插件族，不在本目录。
-
-## 三个 cell
-
-| cell | `FLAGTREE_MUSA_RLC_ENHANCE` | phase mask | 含义 |
-|---|---|---|---|
-| `off` | 0 | — | 增强关闭的基线 |
-| `mask0` | 1 | 0 | 增强开启但所有 phase 关闭；它的 topology 必须等于 `off`，否则控制面本身有问题 |
-| `candidate` | 1 | 13 | 1a、2、3 启用，1b 未启用 |
-
-`off` 记录 autotune config，`mask0` 与 `candidate` 回放同一套 config。回放回退即判定失败，
-这样 A/B 之间的差异才收敛到编译器改动这一个变量。
-
-## 跑法
-
-```bash
-python campaigns/musa_rlc_product/prepare_musa_rlc_product.py \
-  campaigns/musa_rlc_product/manifest/musa-rlc-32-product-v1.json \
-  --inventory-dir ops --selection canary --dry-run
-```
-
-`--dry-run` 只做校验与预算估算。去掉它并给出 `--output-root` 才会冻结 task matrix。
-`--selection all` 需要 `--canary-receipt` 指向一份通过的 canary 收据。
-
-执行：
-
-```bash
-python campaigns/musa_rlc_product/run_musa_rlc_product.py \
-  --manifest <frozen>/manifest.json \
-  --task-matrix <frozen>/task-matrix.json \
-  --build-root <flagtree build root> \
-  --flaggems <flaggems> --zl-bench <this repo> \
-  --worker campaigns/musa_rlc_product/run_pytest_musa_rlc_product.sh \
-  --output-root <run root>
-```
-
-runner 在开跑前校验：manifest 与 matrix 的哈希绑定、FlagTree/FlagGems/zl_bench 三处源码身份、
-Triton 与 FlagGems 确实从冻结路径导入、设备可见数为 1、所有 benchmark 与 correctness 目标
-在真实源码树里存在、设备当前没有其它进程。任何一条不成立就整批拒绝。
-
-单元测试直接跑文件：
-
-```bash
-python campaigns/musa_rlc_product/tests/test_musa_rlc_product.py
-python campaigns/musa_rlc_product/tests/test_reaccount_musa_rlc_topology.py
-```
-
-## 判定
-
-| 类 | 含义 |
-|---|---|
-| G | 正确，convert topology 下降，且双执行顺序下均稳定提速 ≥5% |
-| C | 正确，topology 下降，但收益不足 5% |
-| B | 正确，但 candidate 没有减少被测算子的 convert topology |
-| R | 正确，但双顺序稳定退化 ≤ -3% |
-| K | candidate 特有的正确性失败 |
-| X | 不支持、超时、harness 失败，或同配置回放被破坏 |
-
-只有 topology 相对 incumbent 下降时才追加反向顺序计时。convert 数下降不等于变快，
-这一点在本 campaign 内已有反例。
-
-## 三条口径边界
-
-**convert 计数只算被测算子。** benchmark 调用会先构造一次输入再在计时区内反复执行算子。
-把所有执行过的 kernel 都计入，会把输入构造 kernel 里消掉的 convert 算成收益，而计时器
-根本观测不到它。因此每个 dump 的 TTGIR 都与 `index.tsv` 的 launch 数关联，只有 launch
-数超过 `measured_launch_floor` 的变体才计入。未收敛总数与被排除文件保留在收据里；
-无法与 index 关联的 cell 判为 X，而不是相信一个界定不了的 topology 声明。
-
-**correctness 只跑该 workload 实际测量的变体。** 以整个测试文件为粒度，会让无关变体
-决定被测算子的结论。workload 可以钉 node id 或 `-k` 选择器，两者互斥，且任何收窄都必须
-写明理由。完整 argv 落在收据里。
-
-**目标在开跑前解析。** inventory 可能落后于固定的 FlagGems 提交，例如上游把一个 benchmark
-文件拆成多个。这类任务过去只在 worker 前置检查里静默失败，留下空日志。现在 runner 会
-对真实源码树解析全部目标，workload 可以声明 inventory 里没有的目标但必须给出理由，
-worker 的每个前置检查失败时自报名称。
-
-## 不属于本 campaign 的结论
-
-一次 terminal 的 campaign 只证明：该冻结语料在该设备该提交下的 dispatch 正确性、
-layout topology 响应、以及同契约下的 native-event 延迟。它不证明跨厂商正确性、
-性能对等、CI 状态，也不构成打开生产默认开关的依据。manifest 的
-`production_default_policy` 固定为 `unchanged-off`。

campaigns/musa_rlc_product/manifest/musa-rlc-32-product-v1.json

@@ -1,551 +0,0 @@
-{
-  "schema_version": 1,
-  "campaign_id": "s5000-musa-rlc-32-product-v1-20260902-v3",
-  "experiment_kind": "musa-rlc-product-validation",
-  "permitted_claim": "S5000 MUSA product-dispatch correctness, layout-topology response, and same-contract native-event latency for this frozen corpus only",
-  "production_default_policy": "unchanged-off",
-  "source": {
-    "flagtree_commit": "3059917e2df16f8ea35c5efe1ed1af2d87117396",
-    "flaggems_commit": "46ef8080329190d8da8738340240de2d5a475950",
-    "zl_bench_commit": "da2288564584039366e7be07fe4e94539a560c81"
-  },
-  "device": {
-    "ssh_alias": "S5000",
-    "container": "zixu_flagos_S5000",
-    "backend": "mthreads",
-    "model": "Moore Threads S5000",
-    "physical_device": "7",
-    "logical_device": "0",
-    "visible_device_count": 1
-  },
-  "protocol": {
-    "screen_cells": [
-      "off",
-      "mask0",
-      "candidate"
-    ],
-    "timing_cells": [
-      "off",
-      "candidate"
-    ],
-    "canary_order": [
-      "dropout",
-      "softmax",
-      "instance_norm",
-      "fft",
-      "sort",
-      "batch_norm",
-      "topk",
-      "conv1d"
-    ],
-    "candidate_phase_mask": 13,
-    "candidate_profitability_policy": true,
-    "candidate_product_launch_count": 1,
-    "default_benchmark_dtypes": [
-      "float16"
-    ],
-    "timer": "triton.testing.do_bench via MUSA device-interface Event",
-    "autotune_contract": "off-record then mask0/candidate replay from off",
-    "fresh_process_per_operator_cell": true,
-    "same_physical_device_for_all_cells": true,
-    "serial_measurement": true,
-    "full_campaign_requires_canary_terminal": true,
-    "correctness": "FlagGems public dispatch quick tests under candidate, narrowed to the variant each workload measures and run with the MThreads compatibility plugin, plus benchmark execution for every cell",
-    "classification": {
-      "G": "correct, convert count decreases, and stable speedup >= 5%",
-      "C": "correct and convert count decreases, but latency intervals overlap or speedup < 5%",
-      "B": "correct but candidate does not reduce the executed convert topology",
-      "R": "correct but stable regression <= -3%",
-      "K": "candidate correctness failure",
-      "X": "unsupported, timeout, harness failure, or same-contract replay violation"
-    },
-    "timeout_seconds_by_cost": {
-      "light": 300,
-      "medium": 900,
-      "heavy": 1800
-    },
-    "measured_launch_floor": 1,
-    "supported_benchmark_dtypes": [
-      "float16",
-      "float32",
-      "bfloat16",
-      "int16",
-      "int32",
-      "bool",
-      "complex64"
-    ]
-  },
-  "workloads": [
-    {
-      "id": "batch_norm",
-      "repo": "main",
-      "op": "batch_norm",
-      "op_file": "batch_norm",
-      "shape_key": "batch_norm",
-      "correctness_file": "tests/test_batch_norm.py",
-      "family": "normalization",
-      "nv_evidence": "strong-positive",
-      "nv_gain_percent": 43.1,
-      "prior_s5000": "B 0->0",
-      "cost": "medium",
-      "canary": true
-    },
-    {
-      "id": "replication_pad3d",
-      "repo": "main",
-      "op": "replication_pad3d",
-      "op_file": "replication_pad3d",
-      "shape_key": "replication_pad3d",
-      "correctness_file": "tests/test_replication_pad3d.py",
-      "family": "padding",
-      "nv_evidence": "strong-positive",
-      "nv_gain_percent": 38.6,
-      "prior_s5000": "B 0->0 at author shape",
-      "cost": "medium",
-      "canary": false
-    },
-    {
-      "id": "fft",
-      "repo": "main",
-      "op": "perf_fft",
-      "op_file": "fft",
-      "shape_key": "fft",
-      "correctness_file": "tests/test_fft.py",
-      "family": "fft",
-      "nv_evidence": "strong-positive",
-      "nv_gain_percent": 36.5,
-      "prior_s5000": "G 38->2",
-      "cost": "heavy",
-      "canary": true,
-      "benchmark_dtypes": [
-        "complex64"
-      ]
-    },
-    {
-      "id": "topk",
-      "repo": "main",
-      "op": "topk",
-      "op_file": "topk",
-      "shape_key": "topk",
-      "correctness_file": "tests/test_topk.py",
-      "family": "selection",
-      "nv_evidence": "strong-positive-shared-network-risk",
-      "nv_gain_percent": 20.8,
-      "prior_s5000": "B 0->0 at large author shape",
-      "cost": "medium",
-      "canary": true
-    },
-    {
-      "id": "instance_norm",
-      "repo": "main",
-      "op": "instance_norm",
-      "op_file": "instance_norm",
-      "shape_key": "instance_norm",
-      "correctness_file": "tests/test_instance_norm.py",
-      "family": "normalization",
-      "nv_evidence": "strong-positive",
-      "nv_gain_percent": 17.8,
-      "prior_s5000": "G 1->0",
-      "cost": "medium",
-      "canary": true
-    },
-    {
-      "id": "dropout",
-      "repo": "main",
-      "op": "dropout",
-      "op_file": "dropout",
-      "shape_key": "dropout",
-      "correctness_file": "tests/test_dropout.py",
-      "family": "rng",
-      "nv_evidence": "strong-positive-shared-network-risk",
-      "nv_gain_percent": 15.2,
-      "prior_s5000": "product G 8->0 +9.15%",
-      "cost": "light",
-      "canary": true
-    },
-    {
-      "id": "router_gemm",
-      "repo": "main",
-      "op": "perf_router_gemm",
-      "op_file": "router_gemm",
-      "shape_key": "router_gemm",
-      "correctness_file": "tests/test_router_gemm.py",
-      "family": "mma",
-      "nv_evidence": "strong-positive-mma",
-      "nv_gain_percent": 14.5,
-      "prior_s5000": "public screen available; product v1 pending",
-      "cost": "heavy",
-      "canary": false,
-      "benchmark_dtypes": [
-        "bfloat16"
-      ]
-    },
-    {
-      "id": "fused_recurrent_gated_delta_rule",
-      "repo": "main",
-      "op": "perf_fused_recurrent_gated_delta_rule",
-      "op_file": "FLA/test_fused_recurrent_gated_delta_rule_perf",
-      "shape_key": null,
-      "correctness_file": "tests/test_FLA/test_fused_recurrent_gated_delta_rule.py",
-      "family": "recurrent",
-      "nv_evidence": "strong-positive-recurrent",
-      "nv_gain_percent": 14.2,
-      "prior_s5000": "not product-qualified",
-      "cost": "heavy",
-      "canary": false,
-      "benchmark_dtypes": [
-        "bfloat16"
-      ]
-    },
-    {
-      "id": "top_k_per_row_decode",
-      "repo": "main",
-      "op": "top_k_per_row_decode",
-      "op_file": "top_k_per_row_decode",
-      "shape_key": "top_k_per_row_decode",
-      "correctness_file": "tests/test_top_k_per_row_decode.py",
-      "family": "selection",
-      "nv_evidence": "strong-positive-shared-network-risk",
-      "nv_gain_percent": 13.0,
-      "prior_s5000": "not product-qualified",
-      "cost": "medium",
-      "canary": false
-    },
-    {
-      "id": "sort",
-      "repo": "main",
-      "op": "perf_sort",
-      "op_file": "sort",
-      "shape_key": "sort",
-      "correctness_file": "tests/test_sort.py",
-      "family": "sort",
-      "nv_evidence": "strong-positive-shared-network-risk",
-      "nv_gain_percent": 12.3,
-      "prior_s5000": "G 2->1",
-      "cost": "medium",
-      "canary": true
-    },
-    {
-      "id": "sort_stable",
-      "repo": "main",
-      "op": "perf_sort_stable",
-      "op_file": "sort",
-      "shape_key": "sort_stable",
-      "correctness_file": "tests/test_sort.py",
-      "family": "sort",
-      "nv_evidence": "strong-positive-shared-network-risk",
-      "nv_gain_percent": 12.3,
-      "prior_s5000": "G 2->1",
-      "cost": "medium",
-      "canary": false
-    },
-    {
-      "id": "uniform",
-      "repo": "main",
-      "op": "uniform_",
-      "op_file": "uniform_",
-      "shape_key": "uniform_",
-      "correctness_file": "tests/test_uniform_.py",
-      "family": "rng",
-      "nv_evidence": "strong-positive-rng",
-      "nv_gain_percent": 10.3,
-      "prior_s5000": "public boundary risk; product v1 pending",
-      "cost": "light",
-      "canary": false
-    },
-    {
-      "id": "rand",
-      "repo": "main",
-      "op": "rand",
-      "op_file": "rand",
-      "shape_key": "rand",
-      "correctness_file": "tests/test_rand.py",
-      "family": "rng",
-      "nv_evidence": "strong-positive-rng",
-      "nv_gain_percent": 8.7,
-      "prior_s5000": "product v1 pending",
-      "cost": "light",
-      "canary": false
-    },
-    {
-      "id": "rand_like",
-      "repo": "main",
-      "op": "rand_like",
-      "op_file": "rand_like",
-      "shape_key": "rand_like",
-      "correctness_file": "tests/test_rand_like.py",
-      "family": "rng",
-      "nv_evidence": "strong-positive-rng",
-      "nv_gain_percent": 8.6,
-      "prior_s5000": "product v1 pending",
-      "cost": "light",
-      "canary": false
-    },
-    {
-      "id": "max_pool2d_with_indices",
-      "repo": "main",
-      "op": "max_pool2d_with_indices",
-      "op_file": "max_pool2d",
-      "shape_key": "max_pool2d_with_indices",
-      "correctness_file": "tests/test_max_pool2d.py",
-      "family": "reduction-index",
-      "nv_evidence": "positive-reduction-index",
-      "nv_gain_percent": 6.1,
-      "prior_s5000": "B 0->0",
-      "cost": "medium",
-      "canary": false
-    },
-    {
-      "id": "randperm",
-      "repo": "main",
-      "op": "randperm",
-      "op_file": "randperm",
-      "shape_key": "randperm",
-      "correctness_file": "tests/test_randperm.py",
-      "family": "rng",
-      "nv_evidence": "small-positive-rng",
-      "nv_gain_percent": 4.0,
-      "prior_s5000": "C 7->1 latency overlap",
-      "cost": "medium",
-      "canary": false,
-      "benchmark_dtypes": [
-        "int32"
-      ]
-    },
-    {
-      "id": "randn",
-      "repo": "main",
-      "op": "randn",
-      "op_file": "randn",
-      "shape_key": "randn",
-      "correctness_file": "tests/test_randn.py",
-      "family": "rng",
-      "nv_evidence": "diluted-positive-rng",
-      "nv_gain_percent": 4.0,
-      "prior_s5000": "product v1 pending",
-      "cost": "light",
-      "canary": false
-    },
-    {
-      "id": "randn_like",
-      "repo": "main",
-      "op": "randn_like",
-      "op_file": "randn_like",
-      "shape_key": "randn_like",
-      "correctness_file": "tests/test_randn_like.py",
-      "family": "rng",
-      "nv_evidence": "diluted-positive-rng",
-      "nv_gain_percent": 4.0,
-      "prior_s5000": "product v1 pending",
-      "cost": "light",
-      "canary": false
-    },
-    {
-      "id": "upsample_bicubic2d_aa",
-      "repo": "main",
-      "op": "upsample_bicubic2d_aa",
-      "op_file": "upsample_bicubic2d_aa",
-      "shape_key": "upsample_bicubic2d_aa",
-      "correctness_file": "tests/test_upsample_bicubic2d_aa.py",
-      "family": "interpolation",
-      "nv_evidence": "diluted-positive-cold-path",
-      "nv_gain_percent": 1.7,
-      "prior_s5000": "C 10->0 latency flat",
-      "cost": "medium",
-      "canary": false,
-      "correctness_selector": "not backward",
-      "correctness_scope_reason": "the benchmarked op is the forward path (benchmark/test_upsample_bicubic2d_aa.py::test_upsample_bicubic2d_aa); the backward baseline defect fails identically under off/mask0/candidate and is tracked as a separate MUSA numerics issue"
-    },
-    {
-      "id": "grid_sample",
-      "repo": "main",
-      "op": "grid_sample",
-      "op_file": "grid_sample",
-      "shape_key": "grid_sample",
-      "correctness_file": "tests/test_grid_sample.py",
-      "family": "sampling",
-      "nv_evidence": "structure-change-no-win",
-      "nv_gain_percent": 0.0,
-      "prior_s5000": "B 0->0",
-      "cost": "medium",
-      "canary": false,
-      "benchmark_selector": "dtype0",
-      "benchmark_scope_reason": "benchmark/test_grid_sample.py parametrises over consts.FLOAT_DTYPES and declares dtypes=[dtype] per case, so a campaign-wide --dtypes float16 filter makes the float32 and bfloat16 cases raise; dtype0 is the float16 case this workload measures"
-    },
-    {
-      "id": "max_pool2d_backward",
-      "repo": "main",
-      "op": "max_pool2d_backward",
-      "op_file": "max_pool2d",
-      "shape_key": "max_pool2d_backward",
-      "correctness_file": "tests/test_max_pool2d.py",
-      "family": "reduction-index",
-      "nv_evidence": "structure-change-no-win",
-      "nv_gain_percent": 0.0,
-      "prior_s5000": "B 0->0",
-      "cost": "medium",
-      "canary": false
-    },
-    {
-      "id": "conv1d",
-      "repo": "main",
-      "op": "conv1d",
-      "op_file": "conv1d",
-      "shape_key": "conv1d",
-      "correctness_file": "tests/test_conv1d.py",
-      "family": "convolution",
-      "nv_evidence": "light-regression-review",
-      "nv_gain_percent": -4.2,
-      "prior_s5000": "C 1->0 via phase3, slight slowdown",
-      "cost": "heavy",
-      "canary": true
-    },
-    {
-      "id": "layer_norm_backward",
-      "repo": "main",
-      "op": "layer_norm_backward",
-      "op_file": "layer_norm",
-      "shape_key": "layer_norm_backward",
-      "correctness_file": "tests/test_layer_norm.py",
-      "family": "normalization",
-      "nv_evidence": "light-regression-review",
-      "nv_gain_percent": -1.9,
-      "prior_s5000": "C 4->2 latency flat",
-      "cost": "medium",
-      "canary": false
-    },
-    {
-      "id": "cutlass_scaled_mm",
-      "repo": "main",
-      "op": "cutlass_scaled_mm_benchmark",
-      "op_file": "cutlass_scaled_mm",
-      "shape_key": "cutlass_scaled_mm",
-      "correctness_file": "tests/test_cutlass_scaled_mm.py",
-      "family": "mma",
-      "nv_evidence": "light-regression-mma",
-      "nv_gain_percent": -1.2,
-      "prior_s5000": "likely unsupported CUDA-specific control",
-      "cost": "heavy",
-      "canary": false
-    },
-    {
-      "id": "gelu_backward",
-      "repo": "main",
-      "op": "gelu_backward",
-      "op_file": "gelu",
-      "shape_key": "gelu_backward",
-      "correctness_file": "tests/test_gelu.py",
-      "family": "elementwise-backward",
-      "nv_evidence": "direct-no-cudagraph-noise",
-      "nv_gain_percent": null,
-      "prior_s5000": "B 0->0 with LLVM optimizations",
-      "cost": "medium",
-      "canary": false
-    },
-    {
-      "id": "fused_marlin_moe_mxfp4",
-      "repo": "main",
-      "op": "fused_marlin_moe_w4a16_mxfp4",
-      "op_file": "fused_marlin_moe_w4a16_mxfp4",
-      "shape_key": null,
-      "correctness_file": "tests/test_fused_marlin_moe.py",
-      "family": "quantized-moe",
-      "nv_evidence": "autotune-config-change",
-      "nv_gain_percent": null,
-      "prior_s5000": "likely unsupported CUDA-specific control",
-      "cost": "heavy",
-      "canary": false,
-      "correctness_selector": "mxfp4",
-      "correctness_scope_reason": "tests/test_fused_marlin_moe.py also holds w4a16_int4 and w8a16_int8 variants; only the mxfp4 variant corresponds to this workload, and the int8 failures are a separate baseline defect",
-      "inventory_override_reason": "zl_bench da2288564584039366e7be07fe4e94539a560c81 lists benchmark/test_fused_marlin_moe.py, but FlagGems 46ef8080329190d8da8738340240de2d5a475950 split it into per-quantisation files; the mxfp4 target is benchmark/test_fused_marlin_moe_w4a16_mxfp4.py::test_fused_marlin_moe_w4a16_mxfp4"
-    },
-    {
-      "id": "bincount",
-      "repo": "main",
-      "op": "bincount",
-      "op_file": "bincount",
-      "shape_key": "bincount",
-      "correctness_file": "tests/test_bincount.py",
-      "family": "histogram",
-      "nv_evidence": "autotune-config-change",
-      "nv_gain_percent": null,
-      "prior_s5000": "product v1 pending",
-      "cost": "medium",
-      "canary": false,
-      "benchmark_dtypes": [
-        "float32"
-      ]
-    },
-    {
-      "id": "masked_select",
-      "repo": "main",
-      "op": "masked_select",
-      "op_file": "masked_select",
-      "shape_key": "masked_select",
-      "correctness_file": "tests/test_masked_select.py",
-      "family": "selection",
-      "nv_evidence": "holdout-shared-network",
-      "nv_gain_percent": null,
-      "prior_s5000": "public qualify available",
-      "cost": "medium",
-      "canary": false
-    },
-    {
-      "id": "mm",
-      "repo": "main",
-      "op": "mm",
-      "op_file": "mm",
-      "shape_key": "mm",
-      "correctness_file": "tests/test_mm.py",
-      "family": "mma-control",
-      "nv_evidence": "holdout-mma-control",
-      "nv_gain_percent": null,
-      "prior_s5000": "control only",
-      "cost": "heavy",
-      "canary": false
-    },
-    {
-      "id": "scaled_dot_product_attention",
-      "repo": "main",
-      "op": "scaled_dot_product_attention",
-      "op_file": "scaled_dot_product_attention",
-      "shape_key": "scaled_dot_product_attention",
-      "correctness_file": "tests/test_scaled_dot_product_attention.py",
-      "family": "attention-control",
-      "nv_evidence": "holdout-attention-control",
-      "nv_gain_percent": null,
-      "prior_s5000": "control only",
-      "cost": "heavy",
-      "canary": false
-    },
-    {
-      "id": "softmax",
-      "repo": "main",
-      "op": "softmax",
-      "op_file": "softmax",
-      "shape_key": "softmax",
-      "correctness_file": "tests/test_softmax.py",
-      "family": "reduction-control",
-      "nv_evidence": "holdout-reduction-control",
-      "nv_gain_percent": null,
-      "prior_s5000": "product B 0->0",
-      "cost": "light",
-      "canary": true
-    },
-    {
-      "id": "layer_norm",
-      "repo": "main",
-      "op": "layer_norm",
-      "op_file": "layer_norm",
-      "shape_key": "layer_norm",
-      "correctness_file": "tests/test_layer_norm.py",
-      "family": "normalization-control",
-      "nv_evidence": "holdout-normalization-control",
-      "nv_gain_percent": null,
-      "prior_s5000": "control only",
-      "cost": "medium",
-      "canary": false
-    }
-  ],
-  "predecessor_campaign_id": "s5000-musa-rlc-32-product-v1-20260901-v2"
-}

campaigns/musa_rlc_product/prepare_musa_rlc_product.py

@@ -1,478 +0,0 @@
-#!/usr/bin/env python3
-"""Validate and freeze the single-device MUSA RLC product campaign."""
-
-from __future__ import annotations
-
-import argparse
-import csv
-import hashlib
-import json
-import os
-import shutil
-from pathlib import Path
-from typing import Any
-
-
-COSTS = {"light", "medium", "heavy"}
-CELLS = ["off", "mask0", "candidate"]
-TIMING_CELLS = ["off", "candidate"]
-ESTIMATED_SECONDS = {"light": 180, "medium": 600, "heavy": 1200}
-
-
-class CampaignError(RuntimeError):
-    pass
-
-
-def _canonical(value: object) -> bytes:
-    return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8")
-
-
-def _is_git_sha(value: object) -> bool:
-    return (
-        isinstance(value, str)
-        and len(value) == 40
-        and all(char in "0123456789abcdef" for char in value)
-    )
-
-
-def _correctness_scope_errors(row: dict[str, Any], index: int) -> list[str]:
-    """Check an optional narrowing of a workload's correctness gate.
-
-    Running a whole test file lets an unrelated variant decide the verdict for
-    the operator actually being measured. A workload may therefore pin node ids
-    or a ``-k`` selector, but never both, and narrowing always requires a stated
-    reason so the exclusion stays visible in review.
-    """
-    name = row.get("id", index)
-    selector = row.get("correctness_selector")
-    node_ids = row.get("correctness_node_ids")
-    errors: list[str] = []
-    if selector is not None and node_ids is not None:
-        errors.append(
-            f"workload {name} must not set both correctness_selector and "
-            "correctness_node_ids"
-        )
-    if selector is not None and (not isinstance(selector, str) or not selector.strip()):
-        errors.append(f"workload {name} correctness_selector must be a non-empty string")
-    if node_ids is not None and (
-        not isinstance(node_ids, list)
-        or not node_ids
-        or not all(isinstance(value, str) and value.strip() for value in node_ids)
-    ):
-        errors.append(
-            f"workload {name} correctness_node_ids must be non-empty strings"
-        )
-    if selector is None and node_ids is None:
-        return errors
-    reason = row.get("correctness_scope_reason")
-    if not isinstance(reason, str) or not reason.strip():
-        errors.append(
-            f"workload {name} narrows correctness and must state "
-            "correctness_scope_reason"
-        )
-    return errors
-
-
-def _benchmark_scope_errors(row: dict[str, Any], index: int) -> list[str]:
-    """Check an optional narrowing of a workload's benchmark run.
-
-    Some benchmarks parametrise one dtype per case, so the campaign-wide dtype
-    filter makes every non-matching case raise. A workload may pin the case it
-    measures, but narrowing what gets timed always requires a stated reason.
-    """
-    name = row.get("id", index)
-    selector = row.get("benchmark_selector")
-    if selector is None:
-        return []
-    errors: list[str] = []
-    if not isinstance(selector, str) or not selector.strip():
-        errors.append(f"workload {name} benchmark_selector must be a non-empty string")
-    reason = row.get("benchmark_scope_reason")
-    if not isinstance(reason, str) or not reason.strip():
-        errors.append(
-            f"workload {name} narrows the benchmark and must state "
-            "benchmark_scope_reason"
-        )
-    return errors
-
-
-def _inventory_override_errors(row: dict[str, Any], index: int) -> list[str]:
-    """Check an explicit departure from the pinned zl_bench inventory.
-
-    The inventory can lag the pinned FlagGems commit, for example when upstream
-    splits one benchmark file into several. A workload may then name the target
-    that actually exists, but it must say why, so a stale inventory is never
-    silently routed around.
-    """
-    name = row.get("id", index)
-    reason = row.get("inventory_override_reason")
-    if reason is None:
-        return []
-    if not isinstance(reason, str) or not reason.strip():
-        return [
-            f"workload {name} inventory_override_reason must be a non-empty string"
-        ]
-    return []
-
-
-def validate_manifest(manifest: dict[str, Any]) -> list[str]:
-    errors: list[str] = []
-    if manifest.get("schema_version") != 1:
-        errors.append("schema_version must be 1")
-    if manifest.get("experiment_kind") != "musa-rlc-product-validation":
-        errors.append("experiment_kind must be musa-rlc-product-validation")
-    if manifest.get("production_default_policy") != "unchanged-off":
-        errors.append("production_default_policy must be unchanged-off")
-    if "S5000 MUSA product-dispatch" not in manifest.get("permitted_claim", ""):
-        errors.append("permitted_claim must bind S5000 MUSA product dispatch")
-
-    source = manifest.get("source")
-    if not isinstance(source, dict):
-        errors.append("source must be an object")
-    else:
-        for key in ("flagtree_commit", "flaggems_commit", "zl_bench_commit"):
-            if not _is_git_sha(source.get(key)):
-                errors.append(f"source.{key} must be a lowercase 40-character Git SHA")
-
-    device = manifest.get("device")
-    expected_device = {
-        "ssh_alias": "S5000",
-        "container": "zixu_flagos_S5000",
-        "backend": "mthreads",
-        "model": "Moore Threads S5000",
-        "physical_device": "7",
-        "logical_device": "0",
-        "visible_device_count": 1,
-    }
-    if not isinstance(device, dict):
-        errors.append("device must be an object")
-    else:
-        for key, expected in expected_device.items():
-            if device.get(key) != expected:
-                errors.append(f"device.{key} must equal {expected!r}")
-
-    protocol = manifest.get("protocol")
-    if not isinstance(protocol, dict):
-        errors.append("protocol must be an object")
-    else:
-        if protocol.get("screen_cells") != CELLS:
-            errors.append("protocol.screen_cells must be off/mask0/candidate")
-        if protocol.get("timing_cells") != TIMING_CELLS:
-            errors.append("protocol.timing_cells must be off/candidate")
-        if protocol.get("candidate_phase_mask") != 13:
-            errors.append("candidate_phase_mask must be 13")
-        canary_order = protocol.get("canary_order")
-        if not isinstance(canary_order, list) or len(canary_order) != 8:
-            errors.append("protocol.canary_order must contain eight workload ids")
-        for flag in (
-            "candidate_profitability_policy",
-            "fresh_process_per_operator_cell",
-            "same_physical_device_for_all_cells",
-            "serial_measurement",
-            "full_campaign_requires_canary_terminal",
-        ):
-            if protocol.get(flag) is not True:
-                errors.append(f"protocol.{flag} must be true")
-        if protocol.get("timer") != "triton.testing.do_bench via MUSA device-interface Event":
-            errors.append("protocol.timer must bind the MUSA native-event path")
-        default_dtypes = protocol.get("default_benchmark_dtypes")
-        if not isinstance(default_dtypes, list) or not default_dtypes or not all(
-            isinstance(value, str) and value for value in default_dtypes
-        ):
-            errors.append("protocol.default_benchmark_dtypes must be non-empty strings")
-        supported = protocol.get("supported_benchmark_dtypes")
-        if not isinstance(supported, list) or not supported or not all(
-            isinstance(value, str) and value for value in supported
-        ):
-            errors.append(
-                "protocol.supported_benchmark_dtypes must be non-empty strings"
-            )
-        timeouts = protocol.get("timeout_seconds_by_cost")
-        if not isinstance(timeouts, dict) or set(timeouts) != COSTS:
-            errors.append("timeout_seconds_by_cost must define light/medium/heavy")
-        elif not all(isinstance(value, int) and value > 0 for value in timeouts.values()):
-            errors.append("timeout_seconds_by_cost values must be positive integers")
-        floor = protocol.get("measured_launch_floor")
-        if not isinstance(floor, int) or isinstance(floor, bool) or floor < 0:
-            errors.append(
-                "protocol.measured_launch_floor must be a non-negative integer"
-            )
-
-    workloads = manifest.get("workloads")
-    if not isinstance(workloads, list) or len(workloads) != 32:
-        errors.append("workloads must contain exactly 32 entries")
-        return errors
-
-    ids: list[str] = []
-    nodes: list[tuple[str, str, str]] = []
-    canaries = 0
-    for index, row in enumerate(workloads):
-        if not isinstance(row, dict):
-            errors.append(f"workloads[{index}] must be an object")
-            continue
-        for key in (
-            "id", "repo", "op", "op_file", "correctness_file", "family",
-            "nv_evidence", "prior_s5000", "cost",
-        ):
-            if not isinstance(row.get(key), str) or not row[key]:
-                errors.append(f"workloads[{index}] missing {key}")
-        if row.get("repo") != "main":
-            errors.append(f"workload {row.get('id', index)} must use the main FlagGems repo")
-        if row.get("cost") not in COSTS:
-            errors.append(f"workload {row.get('id', index)} has invalid cost")
-        if not isinstance(row.get("canary"), bool):
-            errors.append(f"workload {row.get('id', index)} canary must be boolean")
-        benchmark_dtypes = row.get(
-            "benchmark_dtypes", protocol.get("default_benchmark_dtypes", [])
-        )
-        if not isinstance(benchmark_dtypes, list) or not benchmark_dtypes or not all(
-            isinstance(value, str) and value for value in benchmark_dtypes
-        ):
-            errors.append(
-                f"workload {row.get('id', index)} benchmark_dtypes must be non-empty strings"
-            )
-        else:
-            # The benchmark CLI rejects a dtype it does not define, and that
-            # only surfaced as an argparse error inside a device run.
-            vocabulary = protocol.get("supported_benchmark_dtypes")
-            if isinstance(vocabulary, list):
-                unknown = [
-                    value for value in benchmark_dtypes if value not in vocabulary
-                ]
-                if unknown:
-                    errors.append(
-                        f"workload {row.get('id', index)} declares benchmark dtype(s) "
-                        f"the benchmark CLI does not accept: {', '.join(unknown)}"
-                    )
-        errors.extend(_correctness_scope_errors(row, index))
-        errors.extend(_benchmark_scope_errors(row, index))
-        errors.extend(_inventory_override_errors(row, index))
-        canaries += row.get("canary") is True
-        ids.append(row.get("id", ""))
-        nodes.append((row.get("repo", ""), row.get("op", ""), row.get("op_file", "")))
-    if len(ids) != len(set(ids)):
-        errors.append("workload ids must be unique")
-    if len(nodes) != len(set(nodes)):
-        errors.append("workload repo/op/op_file triples must be unique")
-    if canaries != 8:
-        errors.append("exactly eight workloads must be canaries")
-    elif set(manifest["protocol"].get("canary_order", [])) != {
-        row["id"] for row in workloads if row["canary"]
-    }:
-        errors.append("protocol.canary_order must contain every canary exactly once")
-    return errors
-
-
-def _load_inventory(inventory_dir: Path) -> dict[tuple[str, str, str], dict[str, str]]:
-    path = inventory_dir / "inventory_main.csv"
-    if not path.is_file():
-        raise CampaignError(f"missing inventory: {path}")
-    rows: dict[tuple[str, str, str], dict[str, str]] = {}
-    with path.open(newline="", encoding="utf-8") as handle:
-        for row in csv.DictReader(handle):
-            key = ("main", row.get("op", ""), row.get("op_file", ""))
-            if key in rows:
-                raise CampaignError(f"duplicate inventory row: {key}")
-            rows[key] = row
-    return rows
-
-
-def validate_inventory(workloads: list[dict[str, Any]], inventory_dir: Path) -> list[str]:
-    inventory = _load_inventory(inventory_dir)
-    errors: list[str] = []
-    for row in workloads:
-        key = (row["repo"], row["op"], row["op_file"])
-        found = inventory.get(key)
-        if found is None:
-            # A stated override admits a target the pinned inventory does not
-            # know about. The runner still resolves it against the real tree.
-            if row.get("inventory_override_reason"):
-                continue
-            errors.append(f"inventory missing {key}")
-            continue
-        if found.get("collected") == "no":
-            errors.append(f"inventory marks {key} collected=no")
-        if found.get("marked_skip") == "yes":
-            errors.append(f"inventory marks {key} skipped")
-    return errors
-
-
-def build_plan(manifest: dict[str, Any], selection: str) -> tuple[dict[str, Any], dict[str, Any]]:
-    workloads = manifest["workloads"]
-    if selection == "canary":
-        by_id = {row["id"]: row for row in workloads if row["canary"]}
-        workloads = [by_id[task_id] for task_id in manifest["protocol"]["canary_order"]]
-    elif selection != "all":
-        raise CampaignError(f"unsupported selection: {selection}")
-
-    timeouts = manifest["protocol"]["timeout_seconds_by_cost"]
-    tasks = []
-    ordinal = 0
-    for row in workloads:
-        ordinal += 1
-        tasks.append(
-            {
-                "ordinal": ordinal,
-                "task_id": row["id"],
-                "repo": row["repo"],
-                "op": row["op"],
-                "op_file": row["op_file"],
-                "shape_key": row.get("shape_key"),
-                "correctness_file": row["correctness_file"],
-                "correctness_selector": row.get("correctness_selector"),
-                "correctness_node_ids": row.get("correctness_node_ids"),
-                "correctness_scope_reason": row.get("correctness_scope_reason"),
-                "benchmark_selector": row.get("benchmark_selector"),
-                "benchmark_scope_reason": row.get("benchmark_scope_reason"),
-                "inventory_override_reason": row.get("inventory_override_reason"),
-                "measured_launch_floor": manifest["protocol"]["measured_launch_floor"],
-                "family": row["family"],
-                "nv_evidence": row["nv_evidence"],
-                "nv_gain_percent": row.get("nv_gain_percent"),
-                "prior_s5000": row["prior_s5000"],
-                "cost": row["cost"],
-                "benchmark_dtypes": row.get(
-                    "benchmark_dtypes",
-                    manifest["protocol"]["default_benchmark_dtypes"],
-                ),
-                "timeout_seconds": timeouts[row["cost"]],
-                "estimated_seconds_per_cell": ESTIMATED_SECONDS[row["cost"]],
-                "cells": CELLS,
-                "status": "PENDING",
-            }
-        )
-    estimate = sum(task["estimated_seconds_per_cell"] * len(CELLS) for task in tasks)
-    matrix = {
-        "schema_version": 1,
-        "campaign_id": manifest["campaign_id"],
-        "selection": selection,
-        "device": manifest["device"],
-        "screen_cells": CELLS,
-        "timing_cells": TIMING_CELLS,
-        "tasks": tasks,
-    }
-    budget = {
-        "schema_version": 1,
-        "campaign_id": manifest["campaign_id"],
-        "selection": selection,
-        "task_count": len(tasks),
-        "gpu_count": 1,
-        "serial_measurement": True,
-        "estimated_wall_seconds": estimate,
-        "estimated_gpu_hours": round(estimate / 3600.0, 4),
-        "note": "Scheduling estimate only; it is not performance evidence.",
-    }
-    return matrix, budget
-
-
-def validate_canary_receipt(
-    manifest: dict[str, Any], selection: str, canary_receipt: Path | None
-) -> list[str]:
-    if selection != "all":
-        return []
-    if canary_receipt is None:
-        return ["full campaign requires --canary-receipt"]
-    try:
-        receipt = json.loads(canary_receipt.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError) as exc:
-        return [f"cannot read canary receipt: {exc}"]
-    errors = []
-    if receipt.get("campaign_id") != manifest.get("campaign_id"):
-        errors.append("canary receipt campaign_id mismatch")
-    if receipt.get("selection") != "canary" or receipt.get("status") != "TERMINAL":
-        errors.append("canary receipt must be a terminal canary")
-    if receipt.get("canary_gate_pass") is not True:
-        errors.append("canary receipt does not authorize the full campaign")
-    return errors
-
-
-def prepare(
-    manifest_path: Path,
-    inventory_dir: Path,
-    output_root: Path,
-    selection: str,
-    canary_receipt: Path | None = None,
-) -> Path:
-    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
-    if not isinstance(manifest, dict):
-        raise CampaignError("manifest root must be an object")
-    errors = validate_manifest(manifest)
-    if not errors:
-        errors.extend(validate_inventory(manifest["workloads"], inventory_dir))
-    if not errors:
-        errors.extend(validate_canary_receipt(manifest, selection, canary_receipt))
-    if errors:
-        raise CampaignError("campaign validation failed:\n- " + "\n- ".join(errors))
-
-    matrix, budget = build_plan(manifest, selection)
-    output = output_root.resolve() / f"{manifest['campaign_id']}--{selection}"
-    if output.exists():
-        raise CampaignError(f"campaign output already exists and is immutable: {output}")
-    output.parent.mkdir(parents=True, exist_ok=True)
-    stage = output.parent / f".{output.name}.staging-{os.getpid()}"
-    if stage.exists():
-        raise CampaignError(f"staging directory already exists: {stage}")
-    try:
-        stage.mkdir()
-        shutil.copy2(manifest_path, stage / "manifest.json")
-        manifest_sha = hashlib.sha256((stage / "manifest.json").read_bytes()).hexdigest()
-        matrix["manifest_sha256"] = manifest_sha
-        budget["manifest_sha256"] = manifest_sha
-        (stage / "manifest.sha256").write_text(
-            f"{manifest_sha}  manifest.json\n", encoding="utf-8", newline="\n"
-        )
-        (stage / "task-matrix.json").write_bytes(_canonical(matrix))
-        (stage / "budget.json").write_bytes(_canonical(budget))
-        (stage / "ops.txt").write_text(
-            "".join(f"{task['repo']}:{task['op']}@{task['op_file']}\n" for task in matrix["tasks"]),
-            encoding="utf-8",
-            newline="\n",
-        )
-        stage.replace(output)
-    except Exception:
-        if stage.exists():
-            shutil.rmtree(stage)
-        raise
-    return output
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser()
-    parser.add_argument("manifest", type=Path)
-    parser.add_argument("--inventory-dir", type=Path, required=True)
-    parser.add_argument("--selection", choices=("canary", "all"), default="canary")
-    parser.add_argument("--output-root", type=Path)
-    parser.add_argument("--canary-receipt", type=Path)
-    parser.add_argument("--dry-run", action="store_true")
-    args = parser.parse_args()
-    try:
-        manifest = json.loads(args.manifest.read_text(encoding="utf-8"))
-        errors = validate_manifest(manifest) if isinstance(manifest, dict) else ["manifest root must be an object"]
-        if not errors:
-            errors.extend(validate_inventory(manifest["workloads"], args.inventory_dir))
-        if not errors and not args.dry_run:
-            errors.extend(validate_canary_receipt(manifest, args.selection, args.canary_receipt))
-        if errors:
-            raise CampaignError("campaign validation failed:\n- " + "\n- ".join(errors))
-        matrix, budget = build_plan(manifest, args.selection)
-        if args.dry_run:
-            print(json.dumps({"status": "DRY_RUN", "budget": budget, "tasks": matrix["tasks"]}, indent=2, sort_keys=True))
-            return 0
-        if args.output_root is None:
-            raise CampaignError("--output-root is required unless --dry-run is used")
-        output = prepare(
-            args.manifest,
-            args.inventory_dir,
-            args.output_root,
-            args.selection,
-            args.canary_receipt,
-        )
-    except (OSError, json.JSONDecodeError, CampaignError) as exc:
-        print(str(exc))
-        return 2
-    print(json.dumps({"status": "PREPARED", "output": str(output)}, indent=2))
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

campaigns/musa_rlc_product/reaccount_musa_rlc_topology.py

@@ -1,223 +0,0 @@
-#!/usr/bin/env python3
-"""Re-account a sealed MUSA RLC run under measured-operator convert scoping.
-
-The sealed campaign counted every executed kernel's ``ttg.convert_layout`` ops,
-including the kernels a benchmark launches once to build its inputs. Those sit
-outside the timed region, so a candidate could be credited with a topology
-reduction that no timer could ever observe.
-
-This tool re-reads the sealed TTGIR dumps and their launch index, recomputes the
-convert counts scoped to repeatedly launched kernels, and re-runs the campaign
-classifier. It never writes inside the sealed run.
-
-A re-accounted verdict can only demote a topology claim. Promoting a task to G
-or C needs reversed-order timing, which a sealed run only holds for tasks that
-already showed a reduction, so a newly appearing reduction is reported as X for
-missing round-b evidence rather than guessed.
-"""
-
-from __future__ import annotations
-
-import argparse
-import csv
-import importlib.util
-import json
-import sys
-from pathlib import Path
-from typing import Any
-
-
-HERE = Path(__file__).resolve().parent
-
-
-class ReaccountError(RuntimeError):
-    pass
-
-
-def _load_runner():
-    spec = importlib.util.spec_from_file_location(
-        "run_musa_rlc_product", HERE / "run_musa_rlc_product.py"
-    )
-    if spec is None or spec.loader is None:
-        raise ReaccountError("cannot load run_musa_rlc_product.py")
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    return module
-
-
-def _canonical(value: object) -> bytes:
-    return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8")
-
-
-def reaccount_task(
-    runner,
-    task_root: Path,
-    receipt: dict[str, Any],
-    measured_launch_floor: int,
-) -> dict[str, Any]:
-    """Recompute one sealed task receipt under the scoped convert count."""
-    cells: dict[str, dict[str, Any]] = {}
-    for key, sealed in receipt.get("cells", {}).items():
-        cell = dict(sealed)
-        cell_dir = task_root / str(sealed.get("round")) / str(sealed.get("cell"))
-        if cell_dir.is_dir():
-            scope = runner._convert_count(
-                cell_dir, measured_launch_floor=measured_launch_floor
-            )
-            cell.update(
-                {
-                    "convert_count": scope["measured_convert_count"],
-                    "convert_count_all": scope["total_convert_count"],
-                    "convert_scope_complete": scope["scope_complete"],
-                    "measured_launch_floor": scope["measured_launch_floor"],
-                    "convert_excluded_unmeasured": scope["excluded_unmeasured"],
-                    "convert_unindexed_files": scope["unindexed_files"],
-                }
-            )
-        else:
-            cell["convert_scope_complete"] = False
-        cells[key] = cell
-
-    classification, detail = runner.classify_task(
-        receipt["correctness"], cells, receipt.get("correctness_isolation")
-    )
-    sealed_class = receipt.get("classification")
-    excluded = sorted(
-        {
-            row["file"]
-            for cell in cells.values()
-            for row in cell.get("convert_excluded_unmeasured", [])
-        }
-    )
-    return {
-        "task_id": receipt["task_id"],
-        "op": receipt.get("op"),
-        "family": receipt.get("family"),
-        "sealed_classification": sealed_class,
-        "reaccounted_classification": classification,
-        "changed": classification != sealed_class,
-        "sealed_reason": receipt.get("classification_detail", {}).get("reason"),
-        "reaccounted_reason": detail["reason"],
-        "sealed_convert": {
-            key: receipt["cells"][key].get("convert_count")
-            for key in sorted(receipt.get("cells", {}))
-        },
-        "reaccounted_convert": {
-            key: cells[key].get("convert_count") for key in sorted(cells)
-        },
-        "excluded_unmeasured_files": excluded,
-        "classification_detail": detail,
-    }
-
-
-def reaccount(
-    run_root: Path, output_root: Path, measured_launch_floor: int
-) -> dict[str, Any]:
-    run_root = run_root.resolve()
-    output_root = output_root.resolve()
-    if not (run_root / "campaign-receipt.json").is_file():
-        raise ReaccountError(f"not a sealed campaign run root: {run_root}")
-    if output_root == run_root or run_root in output_root.parents:
-        raise ReaccountError("refusing to write inside the sealed run root")
-
-    runner = _load_runner()
-    sealed = json.loads((run_root / "campaign-receipt.json").read_text(encoding="utf-8"))
-    rows: list[dict[str, Any]] = []
-    for path in sorted((run_root / "tasks").glob("*/task-receipt.json")):
-        receipt = json.loads(path.read_text(encoding="utf-8"))
-        rows.append(
-            reaccount_task(runner, path.parent, receipt, measured_launch_floor)
-        )
-
-    terminal = sorted(runner.TERMINAL)
-    result = {
-        "schema_version": 1,
-        "kind": "musa-rlc-topology-reaccount",
-        "sealed_campaign_id": sealed.get("campaign_id"),
-        "sealed_run_root": str(run_root),
-        "measured_launch_floor": measured_launch_floor,
-        "task_count": len(rows),
-        "sealed_counts": {
-            key: sum(row["sealed_classification"] == key for row in rows)
-            for key in terminal
-        },
-        "reaccounted_counts": {
-            key: sum(row["reaccounted_classification"] == key for row in rows)
-            for key in terminal
-        },
-        "changed_task_ids": [row["task_id"] for row in rows if row["changed"]],
-        "note": (
-            "Derived accounting over sealed artifacts. It supersedes no receipt "
-            "and produces no new device measurement."
-        ),
-        "tasks": rows,
-    }
-
-    output_root.mkdir(parents=True, exist_ok=True)
-    (output_root / "reaccount-receipt.json").write_bytes(_canonical(result))
-    with (output_root / "reaccount-matrix.csv").open(
-        "w", newline="", encoding="utf-8"
-    ) as handle:
-        writer = csv.writer(handle)
-        writer.writerow(
-            [
-                "task_id",
-                "sealed_class",
-                "reaccounted_class",
-                "changed",
-                "sealed_off",
-                "sealed_candidate",
-                "scoped_off",
-                "scoped_candidate",
-                "reaccounted_reason",
-            ]
-        )
-        for row in rows:
-            writer.writerow(
-                [
-                    row["task_id"],
-                    row["sealed_classification"],
-                    row["reaccounted_classification"],
-                    "yes" if row["changed"] else "no",
-                    row["sealed_convert"].get("a_off"),
-                    row["sealed_convert"].get("a_candidate"),
-                    row["reaccounted_convert"].get("a_off"),
-                    row["reaccounted_convert"].get("a_candidate"),
-                    row["reaccounted_reason"],
-                ]
-            )
-    return result
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--run-root", type=Path, required=True)
-    parser.add_argument("--output-root", type=Path, required=True)
-    parser.add_argument("--measured-launch-floor", type=int, default=1)
-    args = parser.parse_args()
-    try:
-        if args.measured_launch_floor < 0:
-            raise ReaccountError("--measured-launch-floor must be non-negative")
-        result = reaccount(
-            args.run_root, args.output_root, args.measured_launch_floor
-        )
-    except (OSError, json.JSONDecodeError, ReaccountError) as exc:
-        print(f"MUSA_RLC_REACCOUNT_ERROR: {exc}", file=sys.stderr)
-        return 2
-    summary = {
-        key: result[key]
-        for key in (
-            "sealed_campaign_id",
-            "measured_launch_floor",
-            "task_count",
-            "sealed_counts",
-            "reaccounted_counts",
-            "changed_task_ids",
-        )
-    }
-    print(json.dumps(summary, indent=2, sort_keys=True))
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

campaigns/musa_rlc_product/run_musa_rlc_product.py

@@ -1,834 +0,0 @@
-#!/usr/bin/env python3
-"""Run the frozen S5000 MUSA RLC product campaign serially."""
-
-from __future__ import annotations
-
-import argparse
-import csv
-import hashlib
-import importlib.util
-import json
-import math
-import os
-import signal
-import statistics
-import subprocess
-import sys
-import time
-from pathlib import Path
-from typing import Any
-
-
-TERMINAL = {"G", "C", "B", "R", "K", "X"}
-RLC_ENV_KEYS = {
-    "FLAGTREE_RLC_ENHANCE",
-    "FLAGTREE_RLC_PHASE_MASK",
-    "FLAGTREE_MUSA_RLC_ENHANCE",
-    "FLAGTREE_MUSA_RLC_PHASE_MASK",
-    "FLAGTREE_MUSA_RLC_PROFITABILITY_POLICY",
-    "FLAGTREE_MUSA_RLC_PRODUCT_LAUNCH_COUNT",
-    "FLAGTREE_MUSA_RLC_MIN_ADJUSTED_SAVED_COST_PER_TENSOR_OP",
-    "FLAGTREE_MUSA_RLC_PHASE3_SAVED_COST_MULTIPLIER",
-    "FLAGTREE_MUSA_RLC_MAX_EXTERNAL_USE_EDGES",
-    "FLAGTREE_MUSA_RLC_MIN_REMOVED_CONVERT_DENSITY_PER_1024_PROPOSAL_VALUES",
-    "FLAGTREE_MUSA_RLC_PRESERVE_INT_TO_FP_CONTIGUITY",
-    "FLAGTREE_MUSA_RLC_PRESERVE_FP_TO_FP_CONTIGUITY",
-}
-
-
-class CampaignError(RuntimeError):
-    pass
-
-
-def _canonical(value: object) -> bytes:
-    return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8")
-
-
-def _sha256(path: Path) -> str:
-    digest = hashlib.sha256()
-    with path.open("rb") as handle:
-        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
-            digest.update(chunk)
-    return digest.hexdigest()
-
-
-def _atomic_json(path: Path, value: object) -> None:
-    temporary = path.with_name(path.name + ".tmp")
-    temporary.write_bytes(_canonical(value))
-    os.replace(temporary, path)
-
-
-def _load_module(path: Path, name: str):
-    spec = importlib.util.spec_from_file_location(name, path)
-    if spec is None or spec.loader is None:
-        raise CampaignError(f"cannot load module: {path}")
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    return module
-
-
-def _identity(path: Path) -> dict[str, str]:
-    rows: dict[str, str] = {}
-    for raw in path.read_text(encoding="utf-8").splitlines():
-        key, separator, value = raw.partition("=")
-        if separator:
-            rows[key] = value
-    return rows
-
-
-def _base_env(site: Path, flaggems: Path) -> dict[str, str]:
-    env = os.environ.copy()
-    for key in list(env):
-        if key.startswith("FLAGTREE_") and "RLC" in key:
-            env.pop(key)
-    env.pop("DISABLE_LLVM_OPT", None)
-    env.update(
-        {
-            "PYTHONNOUSERSITE": "1",
-            "PYTHONDONTWRITEBYTECODE": "1",
-            "PYTHONUNBUFFERED": "1",
-            "GEMS_VENDOR": "mthreads",
-            "FLAGGEMS_VENDOR": "mthreads",
-            "FLAGTREE_BACKEND": "mthreads",
-            "TRITON_BACKENDS_IN_TREE": "1",
-            "MUSA_VISIBLE_DEVICES": "7",
-            "MTHREADS_VISIBLE_DEVICES": "7",
-            "PYTHONPATH": f"{flaggems / 'src'}:{site}",
-            "FLAGTREE_RLC_TRACE_REJECTS": "1",
-        }
-    )
-    return env
-
-
-def _cell_env(base: dict[str, str], cell: str) -> dict[str, str]:
-    env = base.copy()
-    configs = {
-        "off": {"enhance": "0", "mask": "13", "policy": "0", "launches": "0"},
-        "mask0": {"enhance": "1", "mask": "0", "policy": "1", "launches": "1"},
-        "candidate": {"enhance": "1", "mask": "13", "policy": "1", "launches": "1"},
-    }
-    try:
-        row = configs[cell]
-    except KeyError as exc:
-        raise CampaignError(f"unsupported cell: {cell}") from exc
-    env.update(
-        {
-            "FLAGTREE_MUSA_RLC_ENHANCE": row["enhance"],
-            "FLAGTREE_MUSA_RLC_PHASE_MASK": row["mask"],
-            "FLAGTREE_MUSA_RLC_PROFITABILITY_POLICY": row["policy"],
-            "FLAGTREE_MUSA_RLC_PRODUCT_LAUNCH_COUNT": row["launches"],
-            "FLAGTREE_MUSA_RLC_MIN_ADJUSTED_SAVED_COST_PER_TENSOR_OP": "1800",
-            "FLAGTREE_MUSA_RLC_PHASE3_SAVED_COST_MULTIPLIER": "2",
-            "FLAGTREE_MUSA_RLC_MAX_EXTERNAL_USE_EDGES": "8",
-            "FLAGTREE_MUSA_RLC_MIN_REMOVED_CONVERT_DENSITY_PER_1024_PROPOSAL_VALUES": "0",
-            "FLAGTREE_MUSA_RLC_PRESERVE_INT_TO_FP_CONTIGUITY": "1",
-            "FLAGTREE_MUSA_RLC_PRESERVE_FP_TO_FP_CONTIGUITY": "1",
-        }
-    )
-    return env
-
-
-def _run_process(
-    command: list[str], *, cwd: Path, env: dict[str, str], log: Path, timeout: int
-) -> tuple[int, bool, float]:
-    started = time.time()
-    with log.open("w", encoding="utf-8", newline="\n") as handle:
-        process = subprocess.Popen(
-            command,
-            cwd=str(cwd),
-            env=env,
-            stdout=handle,
-            stderr=subprocess.STDOUT,
-            text=True,
-            start_new_session=True,
-        )
-        timed_out = False
-        try:
-            return_code = process.wait(timeout=timeout)
-        except subprocess.TimeoutExpired:
-            timed_out = True
-            try:
-                os.killpg(process.pid, signal.SIGTERM)
-                process.wait(timeout=10)
-            except subprocess.TimeoutExpired:
-                os.killpg(process.pid, signal.SIGKILL)
-                process.wait()
-            return_code = 124
-    return return_code, timed_out, time.time() - started
-
-
-def _indexed_launches(ttgir_root: Path) -> dict[Path, int] | None:
-    """Map each dumped TTGIR file to its executed launch count.
-
-    Returns None when ``index.tsv`` is missing or unreadable, which means the
-    measured-operator scope cannot be established for this cell.
-    """
-    index = ttgir_root / "index.tsv"
-    if not index.is_file():
-        return None
-    try:
-        lines = index.read_text(encoding="utf-8", errors="replace").splitlines()
-    except OSError:
-        return None
-    if not lines:
-        return None
-    header = lines[0].split("\t")
-    try:
-        shape_at = header.index("shape")
-        kernel_at = header.index("kernel")
-        file_at = header.index("file")
-        launch_at = header.index("launches")
-    except ValueError:
-        return None
-    width = max(shape_at, kernel_at, file_at, launch_at)
-    launches: dict[Path, int] = {}
-    for raw in lines[1:]:
-        fields = raw.split("\t")
-        if len(fields) <= width:
-            continue
-        name = fields[file_at].strip()
-        # Autotune losers are indexed for audit but never dumped; they carry a
-        # parenthesised placeholder instead of a file name.
-        if not name or name.startswith("("):
-            continue
-        try:
-            count = int(fields[launch_at])
-        except ValueError:
-            continue
-        path = ttgir_root / fields[shape_at] / fields[kernel_at] / name
-        launches[path] = max(launches.get(path, 0), count)
-    return launches
-
-
-def _convert_count(path: Path, *, measured_launch_floor: int = 1) -> dict[str, Any]:
-    """Count executed convert_layout ops, scoped to the measured operator.
-
-    A benchmark call constructs its inputs once and then runs the operator many
-    times under the timer. Counting every executed kernel therefore credits the
-    candidate with converts removed from input construction, outside the timed
-    region. Only variants launched more than ``measured_launch_floor`` times
-    contribute to the measured count; the unscoped total is preserved too.
-    """
-    ttgir_root = path / "ttgir"
-    files = sorted(ttgir_root.glob("**/*.ttgir"))
-    launches = _indexed_launches(ttgir_root)
-    total = 0
-    measured = 0
-    excluded: list[dict[str, Any]] = []
-    unindexed: list[str] = []
-    for file in files:
-        converts = file.read_text(encoding="utf-8", errors="replace").count(
-            "ttg.convert_layout"
-        )
-        total += converts
-        relative = file.relative_to(ttgir_root).as_posix()
-        if launches is None:
-            measured += converts
-            continue
-        launch_count = launches.get(file)
-        if launch_count is None:
-            # Dumped but absent from the index: keep it and report the gap
-            # rather than silently dropping executed topology.
-            measured += converts
-            unindexed.append(relative)
-            continue
-        if launch_count > measured_launch_floor:
-            measured += converts
-        elif converts:
-            excluded.append(
-                {
-                    "file": relative,
-                    "launches": launch_count,
-                    "convert_count": converts,
-                }
-            )
-    return {
-        "measured_convert_count": measured,
-        "total_convert_count": total,
-        "ttgir_files": len(files),
-        "measured_launch_floor": measured_launch_floor,
-        "scope_complete": launches is not None and not unindexed,
-        "excluded_unmeasured": excluded,
-        "unindexed_files": unindexed,
-    }
-
-
-def _median_gems_latency(rows: list[dict[str, str]]) -> float | None:
-    values = []
-    for row in rows:
-        if row.get("row_status") != "SUCCESS":
-            continue
-        try:
-            value = float(row.get("gems_lat", ""))
-        except ValueError:
-            continue
-        if math.isfinite(value) and value > 0:
-            values.append(value)
-    return statistics.median(values) if values else None
-
-
-def missing_targets(matrix: dict[str, Any], flaggems: Path) -> list[str]:
-    """Resolve every task's benchmark and correctness target against the tree.
-
-    The inventory a matrix is built from can lag the pinned FlagGems commit, for
-    example when upstream splits one benchmark file into several. Before this
-    check, such a task only failed inside the worker's preflight, which exited
-    silently and left a zero-byte log and a cell receipt with nothing in it.
-    """
-    missing: list[str] = []
-    for task in matrix["tasks"]:
-        benchmark = flaggems / "benchmark" / f"test_{task['op_file']}.py"
-        if not benchmark.is_file():
-            missing.append(f"{task['task_id']} benchmark -> {benchmark}")
-        correctness = flaggems / task["correctness_file"]
-        if not correctness.is_file():
-            missing.append(f"{task['task_id']} correctness -> {correctness}")
-    return missing
-
-
-def validate_contract(
-    manifest_path: Path,
-    matrix_path: Path,
-    build_root: Path,
-    flaggems: Path,
-    zl_bench: Path,
-    python: Path,
-) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
-    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
-    matrix = json.loads(matrix_path.read_text(encoding="utf-8"))
-    if not isinstance(manifest, dict) or not isinstance(matrix, dict):
-        raise CampaignError("manifest and task matrix roots must be objects")
-    manifest_sha = _sha256(manifest_path)
-    if matrix.get("manifest_sha256") != manifest_sha:
-        raise CampaignError("task matrix manifest_sha256 does not match manifest")
-    if matrix.get("campaign_id") != manifest.get("campaign_id"):
-        raise CampaignError("campaign id mismatch")
-    if matrix.get("screen_cells") != ["off", "mask0", "candidate"]:
-        raise CampaignError("task matrix cell contract mismatch")
-
-    missing = missing_targets(matrix, flaggems)
-    if missing:
-        raise CampaignError(
-            "targets missing from the frozen FlagGems tree: " + "; ".join(missing)
-        )
-
-    site = build_root / "site"
-    identity = _identity(build_root / "identity.txt")
-    expected = manifest["source"]
-    if identity.get("git_commit") != expected["flagtree_commit"]:
-        raise CampaignError("FlagTree build identity does not match manifest")
-    if not (build_root / "build.done").is_file():
-        raise CampaignError("FlagTree product build is not terminal")
-    if (flaggems / ".source-commit").read_text(encoding="utf-8").strip() != expected["flaggems_commit"]:
-        raise CampaignError("FlagGems source identity does not match manifest")
-    if (zl_bench / ".source-commit").read_text(encoding="utf-8").strip() != expected["zl_bench_commit"]:
-        raise CampaignError("zl_bench source identity does not match manifest")
-
-    probe_env = _base_env(site, flaggems)
-    probe_code = (
-        "import json,torch,torch_musa,triton,flag_gems;"
-        "print(json.dumps({'count':torch.musa.device_count(),"
-        "'device':str(flag_gems.device),'vendor':str(flag_gems.vendor_name),"
-        "'triton':triton.__file__,'flaggems':flag_gems.__file__},sort_keys=True))"
-    )
-    probe = subprocess.run(
-        [str(python), "-B", "-c", probe_code],
-        cwd=str(flaggems),
-        env=probe_env,
-        text=True,
-        capture_output=True,
-        timeout=60,
-    )
-    if probe.returncode:
-        raise CampaignError(f"product import probe failed: {probe.stderr.strip()}")
-    observed = json.loads(probe.stdout.strip().splitlines()[-1])
-    if observed.get("count") != 1 or observed.get("vendor") != "mthreads":
-        raise CampaignError(f"MUSA device/vendor mismatch: {observed}")
-    if not Path(observed["triton"]).is_relative_to(site):
-        raise CampaignError("imported Triton does not come from the frozen product site")
-    if not Path(observed["flaggems"]).is_relative_to(flaggems / "src"):
-        raise CampaignError("imported FlagGems does not come from the frozen product source")
-
-    observed.update(
-        {
-            "flagtree_commit": identity["git_commit"],
-            "libtriton_sha256": _sha256(site / "triton/_C/libtriton.so"),
-            "flaggems_commit": expected["flaggems_commit"],
-            "zl_bench_commit": expected["zl_bench_commit"],
-            "python": str(python),
-        }
-    )
-    return manifest, matrix, observed
-
-
-def run_cell(
-    task: dict[str, Any],
-    cell: str,
-    round_name: str,
-    replay_from: Path | None,
-    *,
-    output_root: Path,
-    worker: Path,
-    python: Path,
-    zl_bench: Path,
-    flaggems: Path,
-    site: Path,
-    shape_file: Path,
-    batch_summary: Any,
-    base_env: dict[str, str],
-) -> dict[str, Any]:
-    out = output_root / "tasks" / task["task_id"] / round_name / cell
-    env = _cell_env(base_env, cell)
-    env.update(
-        {
-            "PYTHON_BIN": str(python),
-            "ZL_BENCH_DIR": str(zl_bench),
-            "FLAGGEMS_DIR": str(flaggems),
-            "FLAGTREE_SITE": str(site),
-            "OP": task["op"],
-            "OP_FILE": task["op_file"],
-            "SHAPE_FILE": str(shape_file),
-            "OUT_DIR": str(out),
-            "DTYPES": " ".join(task["benchmark_dtypes"]),
-        }
-    )
-    if task.get("benchmark_selector"):
-        env["BENCH_SELECTOR"] = str(task["benchmark_selector"])
-    if replay_from is not None:
-        env["REPLAY_FROM"] = str(replay_from)
-    command = ["bash", str(worker)]
-    # The shell worker owns run.log, so use a sibling controller log.
-    controller_log = out.parent / f"{cell}.controller.log"
-    controller_log.parent.mkdir(parents=True, exist_ok=True)
-    return_code, timed_out, elapsed = _run_process(
-        command,
-        cwd=worker.parent,
-        env=env,
-        log=controller_log,
-        timeout=int(task["timeout_seconds"]),
-    )
-    run_log = out / "run.log"
-    text = run_log.read_text(encoding="utf-8", errors="replace") if run_log.is_file() else ""
-    rows = batch_summary.parse_run_log(text)
-    if timed_out:
-        status = "TIMEOUT"
-    else:
-        status = batch_summary.classify(return_code, rows, text)
-    scope = _convert_count(
-        out, measured_launch_floor=int(task.get("measured_launch_floor", 1))
-    )
-    # A worker can fail its preflight before creating OUT_DIR (for example,
-    # when a frozen inventory entry has no backend benchmark file). Preserve
-    # that failure as a cell receipt instead of aborting the whole campaign.
-    out.mkdir(parents=True, exist_ok=True)
-    receipt = {
-        "task_id": task["task_id"],
-        "round": round_name,
-        "cell": cell,
-        "status": status,
-        "return_code": return_code,
-        "timed_out": timed_out,
-        "elapsed_seconds": round(elapsed, 3),
-        "rows": rows,
-        "rows_success": sum(row.get("row_status") == "SUCCESS" for row in rows),
-        "median_gems_latency_ms": _median_gems_latency(rows),
-        "convert_count": scope["measured_convert_count"],
-        "convert_count_all": scope["total_convert_count"],
-        "convert_scope_complete": scope["scope_complete"],
-        "measured_launch_floor": scope["measured_launch_floor"],
-        "convert_excluded_unmeasured": scope["excluded_unmeasured"],
-        "convert_unindexed_files": scope["unindexed_files"],
-        "ttgir_files": scope["ttgir_files"],
-        "replay_fallback_count": text.count("AUTOTUNE_REPLAY_FALLBACK"),
-        "benchmark_selector": task.get("benchmark_selector"),
-        "benchmark_dtypes": list(task["benchmark_dtypes"]),
-        "effective_rlc": {
-            key: env[key]
-            for key in sorted(RLC_ENV_KEYS)
-            if key in env
-        },
-    }
-    _atomic_json(out / "cell-receipt.json", receipt)
-    return receipt
-
-
-def correctness_command(task: dict[str, Any], python: Path) -> list[str]:
-    """Build the correctness argv, narrowed to the variant this task measures.
-
-    A task-level file runs every variant in that file, so an unrelated failing
-    variant can mask or fabricate a verdict for the operator under test. A
-    manifest may therefore pin explicit node ids or a ``-k`` selector; the
-    manifest schema requires a stated reason whenever it does.
-    """
-    command = [str(python), "-B", "-m", "pytest", "-q"]
-    # The same vendor-scoped compatibility plugin the benchmark worker loads.
-    # Without it a stale CUDA-only guard skips the operator here too, and a
-    # SKIP would be read as "not measured" rather than "not gated".
-    command.extend(["-p", "_musa_benchmark_compat_plugin"])
-    node_ids = task.get("correctness_node_ids")
-    if node_ids:
-        command.extend(str(node) for node in node_ids)
-    else:
-        command.append(task["correctness_file"])
-    selector = task.get("correctness_selector")
-    if selector:
-        command.extend(["-k", str(selector)])
-    # -rs records why a variant was skipped, so a SKIP verdict stays auditable
-    # without a second run.
-    command.extend(["-rs", "--quick"])
-    return command
-
-
-def run_correctness(
-    task: dict[str, Any],
-    cell: str,
-    *,
-    output_root: Path,
-    python: Path,
-    flaggems: Path,
-    site: Path,
-    base_env: dict[str, str],
-    plugin_dir: Path,
-) -> dict[str, Any]:
-    out = output_root / "correctness" / task["task_id"] / cell
-    out.mkdir(parents=True, exist_ok=True)
-    env = _cell_env(base_env, cell)
-    env["PYTEST_DISABLE_PLUGIN_AUTOLOAD"] = "1"
-    env["PYTHONPATH"] = f"{plugin_dir}{os.pathsep}{env['PYTHONPATH']}"
-    env["TRITON_CACHE_DIR"] = str(out / "triton-cache")
-    env["FLAGGEMS_CACHE_DIR"] = str(out / "flaggems-cache")
-    (out / "triton-cache").mkdir()
-    (out / "flaggems-cache").mkdir()
-    command = correctness_command(task, python)
-    return_code, timed_out, elapsed = _run_process(
-        command,
-        cwd=flaggems,
-        env=env,
-        log=out / "correctness.log",
-        timeout=int(task["timeout_seconds"]),
-    )
-    log_text = (out / "correctness.log").read_text(encoding="utf-8", errors="replace")
-    if timed_out:
-        status = "TIMEOUT"
-    elif return_code != 0:
-        status = "FAIL"
-    elif " passed" in log_text:
-        status = "PASS"
-    elif " skipped" in log_text:
-        status = "SKIP"
-    else:
-        status = "FAIL"
-    receipt = {
-        "task_id": task["task_id"],
-        "correctness_file": task["correctness_file"],
-        "correctness_node_ids": task.get("correctness_node_ids"),
-        "correctness_selector": task.get("correctness_selector"),
-        "correctness_scope_reason": task.get("correctness_scope_reason"),
-        "command": command,
-        "cell": cell,
-        "status": status,
-        "return_code": return_code,
-        "timed_out": timed_out,
-        "elapsed_seconds": round(elapsed, 3),
-    }
-    _atomic_json(out / "correctness-receipt.json", receipt)
-    return receipt
-
-
-def _speedup(off: dict[str, Any], candidate: dict[str, Any]) -> float | None:
-    left = off.get("median_gems_latency_ms")
-    right = candidate.get("median_gems_latency_ms")
-    if not isinstance(left, (int, float)) or not isinstance(right, (int, float)) or right <= 0:
-        return None
-    return (left / right - 1.0) * 100.0
-
-
-def classify_task(
-    correctness: dict[str, Any],
-    cells: dict[str, dict[str, Any]],
-    correctness_isolation: dict[str, dict[str, Any]] | None = None,
-) -> tuple[str, dict[str, Any]]:
-    detail: dict[str, Any] = {"reason": ""}
-    if correctness["status"] == "FAIL":
-        isolation = correctness_isolation or {}
-        off_status = isolation.get("off", {}).get("status")
-        mask0_status = isolation.get("mask0", {}).get("status")
-        detail.update(
-            {
-                "off_correctness_status": off_status,
-                "mask0_correctness_status": mask0_status,
-                "candidate_correctness_status": correctness["status"],
-            }
-        )
-        if off_status == "FAIL" and mask0_status == "FAIL":
-            detail["reason"] = "baseline correctness fails under off/mask0/candidate"
-            return "X", detail
-        detail["reason"] = "candidate correctness failed"
-        return "K", detail
-    if correctness["status"] != "PASS":
-        detail["reason"] = f"candidate correctness is {correctness['status']}"
-        return "X", detail
-    required = ("a_off", "a_mask0", "a_candidate")
-    if any(cells.get(key, {}).get("status") != "PASS" for key in required):
-        detail["reason"] = "one or more screen cells did not pass"
-        return "X", detail
-    if any(cells[key].get("replay_fallback_count") for key in ("a_mask0", "a_candidate")):
-        detail["reason"] = "same-config replay fell back"
-        return "X", detail
-    # A topology claim is only admissible when every screen cell could be scoped
-    # to the measured operator. Older receipts predate scoping and are trusted.
-    if any(cells[key].get("convert_scope_complete", True) is False for key in required):
-        detail["reason"] = "measured convert topology scope is incomplete"
-        return "X", detail
-    off_count = cells["a_off"]["convert_count"]
-    mask0_count = cells["a_mask0"]["convert_count"]
-    candidate_count = cells["a_candidate"]["convert_count"]
-    detail.update(
-        {
-            "off_convert_count": off_count,
-            "mask0_convert_count": mask0_count,
-            "candidate_convert_count": candidate_count,
-            "off_convert_count_all": cells["a_off"].get("convert_count_all"),
-            "candidate_convert_count_all": cells["a_candidate"].get("convert_count_all"),
-            "round_a_speedup_percent": _speedup(cells["a_off"], cells["a_candidate"]),
-        }
-    )
-    if off_count != mask0_count:
-        detail["reason"] = "mask0 topology differs from off"
-        return "X", detail
-    if candidate_count >= off_count:
-        speedup = detail["round_a_speedup_percent"]
-        if speedup is not None and speedup <= -3.0:
-            detail["reason"] = "no topology reduction and screening regression"
-            return "R", detail
-        detail["reason"] = "candidate did not reduce executed convert topology"
-        return "B", detail
-
-    if cells.get("b_off", {}).get("status") != "PASS" or cells.get("b_candidate", {}).get("status") != "PASS":
-        detail["reason"] = "topology changed but reversed-order timing did not pass"
-        return "X", detail
-    if cells["b_off"].get("replay_fallback_count") or cells["b_candidate"].get("replay_fallback_count"):
-        detail["reason"] = "reversed-order timing replay fell back"
-        return "X", detail
-    round_b = _speedup(cells["b_off"], cells["b_candidate"])
-    round_a = detail["round_a_speedup_percent"]
-    detail["round_b_speedup_percent"] = round_b
-    if round_a is None or round_b is None:
-        detail["reason"] = "missing comparable native-event latency"
-        return "X", detail
-    if round_a >= 5.0 and round_b >= 5.0:
-        detail["reason"] = "topology reduction and stable >=5% speedup"
-        return "G", detail
-    if round_a <= -3.0 and round_b <= -3.0:
-        detail["reason"] = "topology reduction but stable regression"
-        return "R", detail
-    detail["reason"] = "topology reduction without stable >=5% speedup"
-    return "C", detail
-
-
-def run_campaign(
-    manifest: dict[str, Any],
-    matrix: dict[str, Any],
-    observed: dict[str, Any],
-    *,
-    output_root: Path,
-    worker: Path,
-    python: Path,
-    zl_bench: Path,
-    flaggems: Path,
-    build_root: Path,
-    resume: bool,
-) -> dict[str, Any]:
-    if output_root.exists() and not resume:
-        raise CampaignError(f"output exists; use --resume: {output_root}")
-    observed = dict(observed)
-    observed.update(
-        {
-            "benchmark_runner_sha256": _sha256(Path(__file__)),
-            "benchmark_worker_sha256": _sha256(worker),
-            "musa_benchmark_compat_plugin_sha256": _sha256(
-                zl_bench / "_musa_benchmark_compat_plugin.py"
-            ),
-        }
-    )
-    output_root.mkdir(parents=True, exist_ok=True)
-    site = build_root / "site"
-    shape_file = zl_bench / "ops/shapes_single.yaml"
-    batch_summary = _load_module(zl_bench / "batch_summary.py", "zl_batch_summary")
-    base_env = _base_env(site, flaggems)
-    status_csv = output_root / "status.csv"
-    rows: list[dict[str, Any]] = []
-
-    occupancy = subprocess.run(
-        ["mthreads-gmi"], text=True, capture_output=True, timeout=30
-    )
-    if occupancy.returncode:
-        raise CampaignError(f"mthreads-gmi failed: {occupancy.stderr.strip()}")
-    (output_root / "occupancy-before.txt").write_text(occupancy.stdout, encoding="utf-8", newline="\n")
-    if "No running processes found" not in occupancy.stdout:
-        raise CampaignError("S5000 has an active process; campaign not admitted")
-
-    for task in matrix["tasks"]:
-        task_root = output_root / "tasks" / task["task_id"]
-        task_receipt_path = task_root / "task-receipt.json"
-        if resume and task_receipt_path.is_file():
-            receipt = json.loads(task_receipt_path.read_text(encoding="utf-8"))
-            if receipt.get("classification") in TERMINAL:
-                rows.append(receipt)
-                continue
-        cells: dict[str, dict[str, Any]] = {}
-        off_root = task_root / "round-a" / "off"
-        cells["a_off"] = run_cell(
-            task, "off", "round-a", None,
-            output_root=output_root, worker=worker, python=python,
-            zl_bench=zl_bench, flaggems=flaggems, site=site,
-            shape_file=shape_file, batch_summary=batch_summary, base_env=base_env,
-        )
-        cells["a_mask0"] = run_cell(
-            task, "mask0", "round-a", off_root,
-            output_root=output_root, worker=worker, python=python,
-            zl_bench=zl_bench, flaggems=flaggems, site=site,
-            shape_file=shape_file, batch_summary=batch_summary, base_env=base_env,
-        )
-        cells["a_candidate"] = run_cell(
-            task, "candidate", "round-a", off_root,
-            output_root=output_root, worker=worker, python=python,
-            zl_bench=zl_bench, flaggems=flaggems, site=site,
-            shape_file=shape_file, batch_summary=batch_summary, base_env=base_env,
-        )
-        correctness = run_correctness(
-            task, "candidate", output_root=output_root, python=python, flaggems=flaggems,
-            site=site, base_env=base_env, plugin_dir=zl_bench,
-        )
-        correctness_isolation: dict[str, dict[str, Any]] = {
-            "candidate": correctness
-        }
-        if correctness["status"] != "PASS":
-            correctness_isolation["off"] = run_correctness(
-                task, "off", output_root=output_root, python=python,
-                flaggems=flaggems, site=site, base_env=base_env,
-                plugin_dir=zl_bench,
-            )
-            correctness_isolation["mask0"] = run_correctness(
-                task, "mask0", output_root=output_root, python=python,
-                flaggems=flaggems, site=site, base_env=base_env,
-                plugin_dir=zl_bench,
-            )
-        if (
-            cells["a_off"]["status"] == "PASS"
-            and cells["a_candidate"]["status"] == "PASS"
-            and cells["a_candidate"]["convert_count"] < cells["a_off"]["convert_count"]
-        ):
-            cells["b_candidate"] = run_cell(
-                task, "candidate", "round-b", off_root,
-                output_root=output_root, worker=worker, python=python,
-                zl_bench=zl_bench, flaggems=flaggems, site=site,
-                shape_file=shape_file, batch_summary=batch_summary, base_env=base_env,
-            )
-            cells["b_off"] = run_cell(
-                task, "off", "round-b", off_root,
-                output_root=output_root, worker=worker, python=python,
-                zl_bench=zl_bench, flaggems=flaggems, site=site,
-                shape_file=shape_file, batch_summary=batch_summary, base_env=base_env,
-            )
-        classification, detail = classify_task(
-            correctness, cells, correctness_isolation
-        )
-        receipt = {
-            "task_id": task["task_id"],
-            "op": task["op"],
-            "op_file": task["op_file"],
-            "family": task["family"],
-            "nv_evidence": task["nv_evidence"],
-            "nv_gain_percent": task.get("nv_gain_percent"),
-            "prior_s5000": task["prior_s5000"],
-            "classification": classification,
-            "classification_detail": detail,
-            "correctness": correctness,
-            "correctness_isolation": correctness_isolation,
-            "cells": cells,
-        }
-        task_root.mkdir(parents=True, exist_ok=True)
-        _atomic_json(task_receipt_path, receipt)
-        rows.append(receipt)
-        with status_csv.with_name(status_csv.name + ".tmp").open("w", newline="", encoding="utf-8") as handle:
-            writer = csv.DictWriter(
-                handle,
-                fieldnames=["task_id", "op", "family", "classification", "reason"],
-            )
-            writer.writeheader()
-            for row in rows:
-                writer.writerow(
-                    {
-                        "task_id": row["task_id"],
-                        "op": row["op"],
-                        "family": row["family"],
-                        "classification": row["classification"],
-                        "reason": row["classification_detail"]["reason"],
-                    }
-                )
-        os.replace(status_csv.with_name(status_csv.name + ".tmp"), status_csv)
-        print(
-            f"MUSA_RLC_PRODUCT_TASK task={task['task_id']} class={classification} "
-            f"reason={detail['reason']}",
-            flush=True,
-        )
-
-    counts = {key: sum(row["classification"] == key for row in rows) for key in sorted(TERMINAL)}
-    key_rows = {row["task_id"]: row for row in rows}
-    canary_gate = None
-    if matrix.get("selection") == "canary":
-        canary_gate = (
-            key_rows.get("dropout", {}).get("classification") in {"G", "C"}
-            and key_rows.get("softmax", {}).get("classification") == "B"
-            and counts["K"] == 0
-        )
-    final = {
-        "schema_version": 1,
-        "campaign_id": manifest["campaign_id"],
-        "selection": matrix["selection"],
-        "status": "TERMINAL",
-        "task_count": len(rows),
-        "counts": counts,
-        "canary_gate_pass": canary_gate,
-        "source_observed": observed,
-        "production_default_policy": "unchanged-off",
-        "claim_scope": manifest["permitted_claim"],
-    }
-    _atomic_json(output_root / "campaign-receipt.json", final)
-    return final
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--manifest", type=Path, required=True)
-    parser.add_argument("--task-matrix", type=Path, required=True)
-    parser.add_argument("--build-root", type=Path, required=True)
-    parser.add_argument("--flaggems", type=Path, required=True)
-    parser.add_argument("--zl-bench", type=Path, required=True)
-    parser.add_argument("--worker", type=Path, required=True)
-    parser.add_argument("--python", type=Path, default=Path("/usr/bin/python3"))
-    parser.add_argument("--output-root", type=Path, required=True)
-    parser.add_argument("--resume", action="store_true")
-    args = parser.parse_args()
-    try:
-        manifest, matrix, observed = validate_contract(
-            args.manifest.resolve(), args.task_matrix.resolve(), args.build_root.resolve(),
-            args.flaggems.resolve(), args.zl_bench.resolve(), args.python.resolve(),
-        )
-        final = run_campaign(
-            manifest, matrix, observed,
-            output_root=args.output_root.resolve(), worker=args.worker.resolve(),
-            python=args.python.resolve(), zl_bench=args.zl_bench.resolve(),
-            flaggems=args.flaggems.resolve(), build_root=args.build_root.resolve(),
-            resume=args.resume,
-        )
-    except (OSError, json.JSONDecodeError, CampaignError, subprocess.SubprocessError) as exc:
-        print(f"MUSA_RLC_PRODUCT_ERROR: {exc}", file=sys.stderr)
-        return 2
-    print(json.dumps(final, indent=2, sort_keys=True))
-    return 0 if final.get("canary_gate_pass") is not False else 1
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

campaigns/musa_rlc_product/run_pytest_musa_rlc_product.sh

@@ -1,111 +0,0 @@
-#!/usr/bin/env bash
-set -Eeuo pipefail
-
-: "${PYTHON_BIN:?PYTHON_BIN is required}"
-: "${ZL_BENCH_DIR:?ZL_BENCH_DIR is required}"
-: "${FLAGGEMS_DIR:?FLAGGEMS_DIR is required}"
-: "${FLAGTREE_SITE:?FLAGTREE_SITE is required}"
-: "${OP:?OP is required}"
-: "${OP_FILE:?OP_FILE is required}"
-: "${SHAPE_FILE:?SHAPE_FILE is required}"
-: "${OUT_DIR:?OUT_DIR is required}"
-
-worker_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-
-# A silent preflight exit leaves a zero-byte controller log and no run.log, so
-# the campaign records a failed cell with nothing to diagnose it. Name the check
-# that failed instead.
-preflight() {
-  local description="$1"
-  shift
-  if ! "$@"; then
-    printf 'MUSA_RLC_PRODUCT_PREFLIGHT_FAILED op=%s check=%s\n' \
-      "${OP:-?}" "$description" >&2
-    exit 3
-  fi
-}
-
-preflight "python interpreter is executable: $PYTHON_BIN" test -x "$PYTHON_BIN"
-preflight "zl_bench directory exists: $ZL_BENCH_DIR" test -d "$ZL_BENCH_DIR"
-preflight "ir meta plugin exists: $ZL_BENCH_DIR/_ir_meta_plugin.py" \
-  test -f "$ZL_BENCH_DIR/_ir_meta_plugin.py"
-preflight "benchmark file exists: $FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py" \
-  test -f "$FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py"
-preflight "FlagGems source exists: $FLAGGEMS_DIR/src/flag_gems" \
-  test -d "$FLAGGEMS_DIR/src/flag_gems"
-preflight "libtriton is non-empty: $FLAGTREE_SITE/triton/_C/libtriton.so" \
-  test -s "$FLAGTREE_SITE/triton/_C/libtriton.so"
-preflight "shape file exists: $SHAPE_FILE" test -f "$SHAPE_FILE"
-preflight "output directory is fresh: $OUT_DIR" test ! -e "$OUT_DIR"
-mkdir -p "$OUT_DIR/autotune_records" "$OUT_DIR/triton-cache" "$OUT_DIR/flaggems-cache"
-
-export PYTHONNOUSERSITE=1
-export PYTHONDONTWRITEBYTECODE=1
-export PYTHONUNBUFFERED=1
-export GEMS_VENDOR=mthreads
-export FLAGGEMS_VENDOR=mthreads
-export FLAGTREE_BACKEND=mthreads
-export TRITON_BACKENDS_IN_TREE=1
-export MUSA_VISIBLE_DEVICES=7
-export MTHREADS_VISIBLE_DEVICES=7
-export PYTHONPATH="$worker_dir:$ZL_BENCH_DIR:$FLAGGEMS_DIR/src:$FLAGTREE_SITE"
-export FLAGGEMS_PERF_CURRENT_OP="$OP"
-export FLAGGEMS_PERF_COLOR=never
-export NO_COLOR=1
-
-plugins=(
-  -p _musa_benchmark_compat_plugin
-  -p _device_guard_plugin
-  -p _seed_plugin
-  -p _shape_inject_plugin
-  -p _shape_iter_inject_plugin
-  -p _bespoke_shape_plugin
-  -p _autotune_record_plugin
-  -p _pretty_report_plugin
-  -p _ir_meta_plugin
-)
-
-dtype_args=()
-for dtype in ${DTYPES:-float16}; do
-  dtype_args+=(--dtypes "$dtype")
-done
-
-# Some benchmarks parametrise one dtype per case, so a campaign-wide --dtypes
-# filter makes every non-matching case raise. A workload may then pin the case
-# it measures; the manifest requires a stated reason for that narrowing.
-selector_args=()
-if [[ -n "${BENCH_SELECTOR:-}" ]]; then
-  selector_args+=(-k "$BENCH_SELECTOR")
-fi
-
-if [[ -n "${REPLAY_FROM:-}" ]]; then
-  autotune_env="FLAGGEMS_PERF_AUTOTUNE_REPLAY_DIR=$REPLAY_FROM/autotune_records"
-  mode_desc="replay($REPLAY_FROM)"
-else
-  autotune_env="FLAGGEMS_PERF_AUTOTUNE_RECORD_DIR=$OUT_DIR/autotune_records"
-  mode_desc=record
-fi
-
-cp "$SHAPE_FILE" "$OUT_DIR/shapes.yaml"
-test_file="$FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py::test_${OP}"
-status=0
-{
-  printf 'MUSA_RLC_PRODUCT_RUN op=%s mode=%s out=%s\n' "$OP" "$mode_desc" "$OUT_DIR"
-  env \
-    TRITON_CACHE_DIR="$OUT_DIR/triton-cache" \
-    FLAGGEMS_CACHE_DIR="$OUT_DIR/flaggems-cache" \
-    FLAGGEMS_PERF_TTGIR_DUMP_DIR="$OUT_DIR/ttgir" \
-    "$autotune_env" \
-    USE_FLAGTUNE=0 \
-    "$PYTHON_BIN" -u -m pytest -s "$test_file" \
-      "${plugins[@]}" \
-      --color=no \
-      --shape_file "$SHAPE_FILE" \
-      "${dtype_args[@]}" \
-      "${selector_args[@]}" \
-      --level core --mode kernel || status=$?
-  printf 'MUSA_RLC_PRODUCT_RUN_RC=%s\n' "$status"
-  exit "$status"
-} 2>&1 | tee "$OUT_DIR/run.log" || status=$?
-
-exit "$status"

campaigns/musa_rlc_product/tests/test_musa_rlc_product.py

@@ -1,567 +0,0 @@
-from __future__ import annotations
-
-import copy
-import importlib.util
-import json
-import sys
-import tempfile
-import unittest
-from types import SimpleNamespace
-from pathlib import Path
-
-
-HERE = Path(__file__).resolve().parent
-CAMPAIGN = HERE.parent
-REPO_ROOT = CAMPAIGN.parents[1]
-REFERENCE = CAMPAIGN / "manifest" / "musa-rlc-32-product-v1.json"
-SPEC = importlib.util.spec_from_file_location(
-    "prepare_musa_rlc_product", CAMPAIGN / "prepare_musa_rlc_product.py"
-)
-assert SPEC is not None and SPEC.loader is not None
-MODULE = importlib.util.module_from_spec(SPEC)
-SPEC.loader.exec_module(MODULE)
-RUN_SPEC = importlib.util.spec_from_file_location(
-    "run_musa_rlc_product", CAMPAIGN / "run_musa_rlc_product.py"
-)
-assert RUN_SPEC is not None and RUN_SPEC.loader is not None
-RUNNER = importlib.util.module_from_spec(RUN_SPEC)
-RUN_SPEC.loader.exec_module(RUNNER)
-COMPAT_SPEC = importlib.util.spec_from_file_location(
-    "_musa_benchmark_compat_plugin", REPO_ROOT / "_musa_benchmark_compat_plugin.py"
-)
-assert COMPAT_SPEC is not None and COMPAT_SPEC.loader is not None
-COMPAT = importlib.util.module_from_spec(COMPAT_SPEC)
-COMPAT_SPEC.loader.exec_module(COMPAT)
-
-
-def _manifest():
-    return json.loads(REFERENCE.read_text(encoding="utf-8"))
-
-
-class MusaRlcProductTests(unittest.TestCase):
-    def test_musa_fft_compatibility_override_is_narrow(self):
-        stale = SimpleNamespace(
-            name="skipif", args=(True,), kwargs={"reason": "CUDA is not available"}
-        )
-        unrelated = SimpleNamespace(
-            name="skipif", args=(True,), kwargs={"reason": "another reason"}
-        )
-        fft = SimpleNamespace(path=Path("benchmark/test_fft.py"), name="test_perf_fft")
-        other = SimpleNamespace(path=Path("benchmark/test_fft.py"), name="test_other")
-        self.assertTrue(COMPAT._is_stale_fft_cuda_guard(fft, stale))
-        self.assertFalse(COMPAT._is_stale_fft_cuda_guard(fft, unrelated))
-        self.assertFalse(COMPAT._is_stale_fft_cuda_guard(other, stale))
-
-    def test_runner_preserves_preflight_failure_receipts(self):
-        runner = (CAMPAIGN / "run_musa_rlc_product.py").read_text(encoding="utf-8")
-        ensure_parent = "out.mkdir(parents=True, exist_ok=True)"
-        write_receipt = '_atomic_json(out / "cell-receipt.json", receipt)'
-        self.assertIn(ensure_parent, runner)
-        self.assertLess(runner.index(ensure_parent), runner.index(write_receipt))
-
-    def test_worker_dumps_only_executed_ttgir_variants(self):
-        worker = (CAMPAIGN / "run_pytest_musa_rlc_product.sh").read_text(
-            encoding="utf-8"
-        )
-        self.assertIn(
-            'FLAGGEMS_PERF_TTGIR_DUMP_DIR="$OUT_DIR/ttgir"', worker
-        )
-        self.assertIn("-p _musa_benchmark_compat_plugin", worker)
-        self.assertIn(
-            'PYTHONPATH="$worker_dir:$ZL_BENCH_DIR:$FLAGGEMS_DIR/src:$FLAGTREE_SITE"',
-            worker,
-        )
-
-        with tempfile.TemporaryDirectory() as temporary:
-            root = Path(temporary)
-            dumped = root / "ttgir" / "shape" / "kernel" / "used.ttgir"
-            cached = root / "triton-cache" / "hash" / "loser.ttgir"
-            dumped.parent.mkdir(parents=True)
-            cached.parent.mkdir(parents=True)
-            dumped.write_text("ttg.convert_layout\nttg.convert_layout\n", encoding="utf-8")
-            cached.write_text("ttg.convert_layout\n" * 99, encoding="utf-8")
-            scope = RUNNER._convert_count(root)
-            self.assertEqual(scope["total_convert_count"], 2)
-            self.assertEqual(scope["ttgir_files"], 1)
-            # Without an index the measured scope cannot be established, so the
-            # count stays unscoped and the cell is marked incomplete.
-            self.assertEqual(scope["measured_convert_count"], 2)
-            self.assertFalse(scope["scope_complete"])
-
-    @staticmethod
-    def _write_cell(root, rows):
-        """Materialise a cell's ttgir dump plus the index the dumper writes."""
-        ttgir = root / "ttgir"
-        ttgir.mkdir(parents=True)
-        lines = [
-            "\t".join(
-                [
-                    "shape", "kernel", "file", "launches", "sweep_launches",
-                    "num_warps", "num_stages", "constexprs", "cache_hash",
-                ]
-            )
-        ]
-        for shape, kernel, name, launches, converts in rows:
-            if name.startswith("("):
-                lines.append(
-                    "\t".join([shape, kernel, name, str(launches), "0", "4", "3", "{}", "H"])
-                )
-                continue
-            path = ttgir / shape / kernel / name
-            path.parent.mkdir(parents=True, exist_ok=True)
-            path.write_text("ttg.convert_layout\n" * converts, encoding="utf-8")
-            lines.append(
-                "\t".join([shape, kernel, name, str(launches), "0", "4", "3", "{}", "H"])
-            )
-        (ttgir / "index.tsv").write_text("\n".join(lines) + "\n", encoding="utf-8")
-
-    def test_convert_count_excludes_input_construction_kernels(self):
-        # Reproduces the sealed gelu_backward shape: every convert sits in the
-        # once-launched input builder, while the timed kernel has none.
-        with tempfile.TemporaryDirectory() as temporary:
-            root = Path(temporary)
-            self._write_cell(
-                root,
-                [
-                    ("1048576", "randn_kernel", "w16s3.ttgir", 1, 4),
-                    ("1048576", "gelu_none_kernel_rank_1", "w4s3.ttgir", 1, 0),
-                    ("1048576", "gelu_backward_none_kernel_rank_1", "w4s3.ttgir", 3795, 0),
-                ],
-            )
-            scope = RUNNER._convert_count(root)
-            self.assertEqual(scope["total_convert_count"], 4)
-            self.assertEqual(scope["measured_convert_count"], 0)
-            self.assertTrue(scope["scope_complete"])
-            self.assertEqual(
-                [row["file"] for row in scope["excluded_unmeasured"]],
-                ["1048576/randn_kernel/w16s3.ttgir"],
-            )
-
-    def test_convert_count_keeps_repeatedly_launched_kernels(self):
-        # Reproduces the sealed sort shape: converts in the dominant kernel and
-        # in an autotune loser that was indexed but never dumped.
-        with tempfile.TemporaryDirectory() as temporary:
-            root = Path(temporary)
-            self._write_cell(
-                root,
-                [
-                    ("1024x1024", "sweep", "w4s3.ttgir", 280, 3),
-                    ("1024x1024", "broadcast_to_kernel", "w4s3.ttgir", 140, 1),
-                    ("-", "arange_kernel", "(sweep loser, not dumped)", 0, 0),
-                ],
-            )
-            scope = RUNNER._convert_count(root)
-            self.assertEqual(scope["measured_convert_count"], 4)
-            self.assertEqual(scope["total_convert_count"], 4)
-            self.assertTrue(scope["scope_complete"])
-            self.assertEqual(scope["excluded_unmeasured"], [])
-
-    def test_convert_count_reports_dumped_but_unindexed_files(self):
-        with tempfile.TemporaryDirectory() as temporary:
-            root = Path(temporary)
-            self._write_cell(root, [("s", "measured", "w4s3.ttgir", 140, 2)])
-            stray = root / "ttgir" / "s" / "stray" / "w4s3.ttgir"
-            stray.parent.mkdir(parents=True)
-            stray.write_text("ttg.convert_layout\n", encoding="utf-8")
-            scope = RUNNER._convert_count(root)
-            # An unindexed dump is counted rather than dropped, and flagged.
-            self.assertEqual(scope["measured_convert_count"], 3)
-            self.assertFalse(scope["scope_complete"])
-            self.assertEqual(scope["unindexed_files"], ["s/stray/w4s3.ttgir"])
-
-    def test_incomplete_convert_scope_fails_closed(self):
-        def cell(convert_count, latency, complete=True):
-            return {
-                "status": "PASS",
-                "convert_count": convert_count,
-                "convert_scope_complete": complete,
-                "median_gems_latency_ms": latency,
-                "replay_fallback_count": 0,
-            }
-
-        cells = {
-            "a_off": cell(8, 0.14),
-            "a_mask0": cell(8, 0.14),
-            "a_candidate": cell(0, 0.12, complete=False),
-            "b_candidate": cell(0, 0.121),
-            "b_off": cell(8, 0.141),
-        }
-        classification, detail = RUNNER.classify_task({"status": "PASS"}, cells)
-        self.assertEqual(classification, "X")
-        self.assertEqual(
-            detail["reason"], "measured convert topology scope is incomplete"
-        )
-
-    def test_correctness_command_narrows_to_the_measured_variant(self):
-        python = Path("/usr/bin/python3")
-        plain = RUNNER.correctness_command(
-            {"correctness_file": "tests/test_sort.py"}, python
-        )
-        self.assertEqual(plain[-3:], ["tests/test_sort.py", "-rs", "--quick"])
-        self.assertNotIn("-k", plain)
-
-        selected = RUNNER.correctness_command(
-            {
-                "correctness_file": "tests/test_upsample_bicubic2d_aa.py",
-                "correctness_selector": "not backward",
-            },
-            python,
-        )
-        self.assertIn("tests/test_upsample_bicubic2d_aa.py", selected)
-        self.assertEqual(selected[selected.index("-k") + 1], "not backward")
-
-        pinned = RUNNER.correctness_command(
-            {
-                "correctness_file": "tests/test_mm.py",
-                "correctness_node_ids": ["tests/test_mm.py::test_mm"],
-            },
-            python,
-        )
-        self.assertIn("tests/test_mm.py::test_mm", pinned)
-        self.assertNotIn("tests/test_mm.py", pinned)
-
-    def test_narrowed_correctness_requires_a_stated_reason(self):
-        manifest = copy.deepcopy(_manifest())
-        target = next(
-            row for row in manifest["workloads"] if row["id"] == "upsample_bicubic2d_aa"
-        )
-        self.assertEqual(target["correctness_selector"], "not backward")
-        target.pop("correctness_scope_reason")
-        errors = MODULE.validate_manifest(manifest)
-        self.assertIn(
-            "workload upsample_bicubic2d_aa narrows correctness and must state "
-            "correctness_scope_reason",
-            errors,
-        )
-
-        both = copy.deepcopy(_manifest())
-        row = next(item for item in both["workloads"] if item["id"] == "sort")
-        row["correctness_selector"] = "sort"
-        row["correctness_node_ids"] = ["tests/test_sort.py::test_sort"]
-        row["correctness_scope_reason"] = "reason"
-        self.assertIn(
-            "workload sort must not set both correctness_selector and "
-            "correctness_node_ids",
-            MODULE.validate_manifest(both),
-        )
-
-    def test_measured_launch_floor_is_a_frozen_protocol_field(self):
-        manifest = copy.deepcopy(_manifest())
-        self.assertEqual(manifest["protocol"]["measured_launch_floor"], 1)
-        plan, _ = MODULE.build_plan(manifest, "all")
-        self.assertTrue(
-            all(task["measured_launch_floor"] == 1 for task in plan["tasks"])
-        )
-        scoped = next(
-            task for task in plan["tasks"] if task["task_id"] == "fused_marlin_moe_mxfp4"
-        )
-        self.assertEqual(scoped["correctness_selector"], "mxfp4")
-        self.assertTrue(scoped["correctness_scope_reason"])
-
-        manifest["protocol"]["measured_launch_floor"] = -1
-        self.assertIn(
-            "protocol.measured_launch_floor must be a non-negative integer",
-            MODULE.validate_manifest(manifest),
-        )
-
-    def test_compat_plugin_only_overrides_verified_stale_guards(self):
-        def mark(reason, name="skipif"):
-            return SimpleNamespace(name=name, args=(True,), kwargs={"reason": reason})
-
-        def item(file_name, test_name):
-            return SimpleNamespace(path=Path("tests") / file_name, name=test_name)
-
-        # Verified on S5000: unskipped, these pass.
-        self.assertTrue(
-            COMPAT._is_stale_cuda_guard(
-                item("test_top_k_per_row_decode.py", "test_accuracy_top_k"),
-                mark("CUDA device required"),
-            )
-        )
-        self.assertTrue(
-            COMPAT._is_stale_fft_cuda_guard(
-                item("test_fft.py", "test_perf_fft"), mark("CUDA is not available")
-            )
-        )
-        # The fft guard stays pinned to the one test it was verified against.
-        self.assertFalse(
-            COMPAT._is_stale_cuda_guard(
-                item("test_fft.py", "test_other"), mark("CUDA is not available")
-            )
-        )
-        # Probed and rejected: grid_sample hardcodes device="cuda" in its own
-        # helper, and cutlass_scaled_mm gates on vLLM as well as CUDA.
-        self.assertFalse(
-            COMPAT._is_stale_cuda_guard(
-                item("test_grid_sample.py", "test_nearest_zeros_4d_small"),
-                mark("CUDA is required."),
-            )
-        )
-        self.assertFalse(
-            COMPAT._is_stale_cuda_guard(
-                item("test_cutlass_scaled_mm.py", "test_cutlass_scaled_mm"),
-                mark("Feature requires vLLM and NVIDIA Hopper architecture"),
-            )
-        )
-        # Only skipif marks, and only the exact reason string.
-        self.assertFalse(
-            COMPAT._is_stale_cuda_guard(
-                item("test_top_k_per_row_decode.py", "test_accuracy_top_k"),
-                mark("CUDA device required", name="xfail"),
-            )
-        )
-        self.assertFalse(
-            COMPAT._is_stale_cuda_guard(
-                item("test_top_k_per_row_decode.py", "test_accuracy_top_k"),
-                mark("some other reason"),
-            )
-        )
-
-    def test_compat_plugin_strips_module_level_guards(self):
-        # top_k_per_row_decode carries its guard as a module-level pytestmark,
-        # which resolves through the node chain rather than the item's own marks.
-        stale = SimpleNamespace(
-            name="skipif", args=(True,), kwargs={"reason": "CUDA device required"}
-        )
-        keep = SimpleNamespace(
-            name="skipif", args=(True,), kwargs={"reason": "vLLM not installed"}
-        )
-        module = SimpleNamespace(own_markers=[stale, keep])
-        function = SimpleNamespace(
-            path=Path("tests/test_top_k_per_row_decode.py"),
-            name="test_accuracy_top_k",
-            own_markers=[],
-        )
-        function.listchain = lambda: [module, function]
-
-        sys.modules.setdefault(
-            "flag_gems", SimpleNamespace(vendor_name="mthreads")
-        )
-        COMPAT.pytest_collection_modifyitems([function])
-        self.assertEqual(module.own_markers, [keep])
-
-    def test_benchmark_selector_requires_a_stated_reason(self):
-        manifest = copy.deepcopy(_manifest())
-        target = next(
-            row for row in manifest["workloads"] if row["id"] == "grid_sample"
-        )
-        self.assertEqual(target["benchmark_selector"], "dtype0")
-        target.pop("benchmark_scope_reason")
-        self.assertIn(
-            "workload grid_sample narrows the benchmark and must state "
-            "benchmark_scope_reason",
-            MODULE.validate_manifest(manifest),
-        )
-
-    def test_declared_benchmark_dtypes_must_exist_in_the_cli_vocabulary(self):
-        manifest = copy.deepcopy(_manifest())
-        self.assertEqual(
-            next(row for row in manifest["workloads"] if row["id"] == "randperm")[
-                "benchmark_dtypes"
-            ],
-            ["int32"],
-        )
-        # int64 is not a --dtypes choice; the sealed run only found that out on
-        # the device, as an argparse error inside a screen cell.
-        next(row for row in manifest["workloads"] if row["id"] == "randperm")[
-            "benchmark_dtypes"
-        ] = ["int32", "int64"]
-        self.assertIn(
-            "workload randperm declares benchmark dtype(s) the benchmark CLI "
-            "does not accept: int64",
-            MODULE.validate_manifest(manifest),
-        )
-
-    def test_stale_inventory_needs_an_override_reason(self):
-        manifest = copy.deepcopy(_manifest())
-        marlin = next(
-            row
-            for row in manifest["workloads"]
-            if row["id"] == "fused_marlin_moe_mxfp4"
-        )
-        self.assertEqual(marlin["op_file"], "fused_marlin_moe_w4a16_mxfp4")
-        self.assertTrue(marlin["inventory_override_reason"])
-
-        with tempfile.TemporaryDirectory() as temporary:
-            inventory = Path(temporary)
-            (inventory / "inventory_main.csv").write_text(
-                "repo,op,op_file,file,marked_skip,collected,op_names\n"
-                + "".join(
-                    f"main,{row['op']},{row['op_file']},f.py,,yes,{row['id']}\n"
-                    for row in manifest["workloads"]
-                    if row["id"] != "fused_marlin_moe_mxfp4"
-                ),
-                encoding="utf-8",
-            )
-            # The stated override lets the missing row through.
-            self.assertEqual(
-                MODULE.validate_inventory(manifest["workloads"], inventory), []
-            )
-            marlin.pop("inventory_override_reason")
-            self.assertEqual(
-                MODULE.validate_inventory(manifest["workloads"], inventory),
-                [
-                    "inventory missing ('main', 'fused_marlin_moe_w4a16_mxfp4', "
-                    "'fused_marlin_moe_w4a16_mxfp4')"
-                ],
-            )
-
-    def test_missing_benchmark_target_fails_before_the_campaign_starts(self):
-        matrix = {
-            "tasks": [
-                {
-                    "task_id": "present",
-                    "op_file": "sort",
-                    "correctness_file": "tests/test_sort.py",
-                },
-                {
-                    # The shape of the sealed marlin defect: the inventory named
-                    # a merged benchmark file that upstream had already split.
-                    "task_id": "fused_marlin_moe_mxfp4",
-                    "op_file": "fused_marlin_moe",
-                    "correctness_file": "tests/test_fused_marlin_moe.py",
-                },
-            ]
-        }
-        with tempfile.TemporaryDirectory() as temporary:
-            flaggems = Path(temporary)
-            (flaggems / "benchmark").mkdir()
-            (flaggems / "tests").mkdir()
-            (flaggems / "benchmark" / "test_sort.py").write_text("", encoding="utf-8")
-            (flaggems / "tests" / "test_sort.py").write_text("", encoding="utf-8")
-            (flaggems / "tests" / "test_fused_marlin_moe.py").write_text(
-                "", encoding="utf-8"
-            )
-            missing = RUNNER.missing_targets(matrix, flaggems)
-
-        self.assertEqual(len(missing), 1)
-        self.assertIn("fused_marlin_moe_mxfp4 benchmark", missing[0])
-        self.assertIn("test_fused_marlin_moe.py", missing[0])
-
-    def test_worker_names_the_failed_preflight_check(self):
-        worker = (CAMPAIGN / "run_pytest_musa_rlc_product.sh").read_text(encoding="utf-8")
-        self.assertIn("MUSA_RLC_PRODUCT_PREFLIGHT_FAILED", worker)
-        self.assertIn(
-            'preflight "benchmark file exists: '
-            '$FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py"',
-            worker,
-        )
-        # The benchmark selector must reach pytest.
-        self.assertIn('selector_args+=(-k "$BENCH_SELECTOR")', worker)
-        self.assertIn('"${selector_args[@]}"', worker)
-
-    def test_correctness_runs_with_the_vendor_compat_plugin(self):
-        command = RUNNER.correctness_command(
-            {"correctness_file": "tests/test_top_k_per_row_decode.py"},
-            Path("/usr/bin/python3"),
-        )
-        self.assertEqual(
-            command[command.index("-p") + 1], "_musa_benchmark_compat_plugin"
-        )
-
-    def test_valid_manifest_and_plan_sizes(self):
-        manifest = _manifest()
-        self.assertEqual(MODULE.validate_manifest(manifest), [])
-        canary, canary_budget = MODULE.build_plan(manifest, "canary")
-        full, full_budget = MODULE.build_plan(manifest, "all")
-        self.assertEqual(len(canary["tasks"]), 8)
-        self.assertEqual(len(full["tasks"]), 32)
-        self.assertEqual(canary["tasks"][0]["task_id"], "dropout")
-        self.assertEqual(canary["tasks"][1]["task_id"], "softmax")
-        self.assertEqual(canary["tasks"][0]["benchmark_dtypes"], ["float16"])
-        self.assertEqual(
-            next(task for task in canary["tasks"] if task["task_id"] == "fft")[
-                "benchmark_dtypes"
-            ],
-            ["complex64"],
-        )
-        self.assertEqual(canary_budget["gpu_count"], 1)
-        self.assertEqual(full_budget["gpu_count"], 1)
-        self.assertTrue(canary_budget["serial_measurement"])
-
-    def test_invalid_device_and_canary_count_fail_closed(self):
-        manifest = copy.deepcopy(_manifest())
-        manifest["device"]["physical_device"] = "0"
-        manifest["workloads"][0]["canary"] = False
-        errors = MODULE.validate_manifest(manifest)
-        self.assertIn("device.physical_device must equal '7'", errors)
-        self.assertIn("exactly eight workloads must be canaries", errors)
-
-    def test_invalid_timer_and_default_policy_fail_closed(self):
-        manifest = copy.deepcopy(_manifest())
-        manifest["protocol"]["timer"] = "host wall clock"
-        manifest["production_default_policy"] = "enabled"
-        errors = MODULE.validate_manifest(manifest)
-        self.assertIn("protocol.timer must bind the MUSA native-event path", errors)
-        self.assertIn("production_default_policy must be unchanged-off", errors)
-
-    def test_runtime_classifier_positive_and_fail_closed_paths(self):
-        def cell(convert_count, latency, status="PASS", replay=0):
-            return {
-                "status": status,
-                "convert_count": convert_count,
-                "median_gems_latency_ms": latency,
-                "replay_fallback_count": replay,
-            }
-
-        correct = {"status": "PASS"}
-        cells = {
-            "a_off": cell(8, 0.14),
-            "a_mask0": cell(8, 0.14),
-            "a_candidate": cell(0, 0.12),
-            "b_candidate": cell(0, 0.121),
-            "b_off": cell(8, 0.141),
-        }
-        classification, _ = RUNNER.classify_task(correct, cells)
-        self.assertEqual(classification, "G")
-
-        invalid = copy.deepcopy(cells)
-        invalid["a_mask0"]["convert_count"] = 7
-        classification, detail = RUNNER.classify_task(correct, invalid)
-        self.assertEqual(classification, "X")
-        self.assertEqual(detail["reason"], "mask0 topology differs from off")
-
-        classification, _ = RUNNER.classify_task({"status": "FAIL"}, cells)
-        self.assertEqual(classification, "K")
-
-        classification, detail = RUNNER.classify_task(
-            {"status": "FAIL"},
-            cells,
-            {
-                "candidate": {"status": "FAIL"},
-                "mask0": {"status": "FAIL"},
-                "off": {"status": "FAIL"},
-            },
-        )
-        self.assertEqual(classification, "X")
-        self.assertEqual(
-            detail["reason"],
-            "baseline correctness fails under off/mask0/candidate",
-        )
-
-    def test_full_preparation_requires_passing_canary_receipt(self):
-        manifest = _manifest()
-        self.assertEqual(
-            MODULE.validate_canary_receipt(manifest, "all", None),
-            ["full campaign requires --canary-receipt"],
-        )
-        with tempfile.TemporaryDirectory() as temporary:
-            receipt = Path(temporary) / "campaign-receipt.json"
-            receipt.write_text(
-                json.dumps(
-                    {
-                        "campaign_id": manifest["campaign_id"],
-                        "selection": "canary",
-                        "status": "TERMINAL",
-                        "canary_gate_pass": True,
-                    }
-                ),
-                encoding="utf-8",
-            )
-            self.assertEqual(
-                MODULE.validate_canary_receipt(manifest, "all", receipt), []
-            )
-
-
-if __name__ == "__main__":
-    unittest.main()

campaigns/musa_rlc_product/tests/test_reaccount_musa_rlc_topology.py

@@ -1,157 +0,0 @@
-from __future__ import annotations
-
-import importlib.util
-import json
-import tempfile
-import unittest
-from pathlib import Path
-
-
-HERE = Path(__file__).resolve().parent
-CAMPAIGN = HERE.parent
-SPEC = importlib.util.spec_from_file_location(
-    "reaccount_musa_rlc_topology", CAMPAIGN / "reaccount_musa_rlc_topology.py"
-)
-assert SPEC is not None and SPEC.loader is not None
-TOOL = importlib.util.module_from_spec(SPEC)
-SPEC.loader.exec_module(TOOL)
-
-
-def _write_cell(cell_dir: Path, rows):
-    ttgir = cell_dir / "ttgir"
-    ttgir.mkdir(parents=True)
-    lines = [
-        "\t".join(
-            [
-                "shape", "kernel", "file", "launches", "sweep_launches",
-                "num_warps", "num_stages", "constexprs", "cache_hash",
-            ]
-        )
-    ]
-    for shape, kernel, name, launches, converts in rows:
-        path = ttgir / shape / kernel / name
-        path.parent.mkdir(parents=True, exist_ok=True)
-        path.write_text("ttg.convert_layout\n" * converts, encoding="utf-8")
-        lines.append(
-            "\t".join([shape, kernel, name, str(launches), "0", "4", "3", "{}", "H"])
-        )
-    (ttgir / "index.tsv").write_text("\n".join(lines) + "\n", encoding="utf-8")
-
-
-def _seal_run(root: Path):
-    """A minimal sealed run shaped like the gelu_backward task."""
-    (root / "campaign-receipt.json").write_text(
-        json.dumps({"campaign_id": "sealed-v2"}), encoding="utf-8"
-    )
-    task_root = root / "tasks" / "gelu_backward"
-    cells = {}
-    for key, round_name, cell_name, setup, measured in (
-        ("a_off", "round-a", "off", 4, 0),
-        ("a_mask0", "round-a", "mask0", 4, 0),
-        ("a_candidate", "round-a", "candidate", 0, 0),
-        ("b_off", "round-b", "off", 4, 0),
-        ("b_candidate", "round-b", "candidate", 0, 0),
-    ):
-        cell_dir = task_root / round_name / cell_name
-        _write_cell(
-            cell_dir,
-            [
-                ("s", "randn_kernel", "w16s3.ttgir", 1, setup),
-                ("s", "gelu_backward_kernel", "w4s3.ttgir", 3795, measured),
-            ],
-        )
-        cells[key] = {
-            "round": round_name,
-            "cell": cell_name,
-            "status": "PASS",
-            "convert_count": setup + measured,
-            "median_gems_latency_ms": 0.1168 if cell_name == "off" else 0.1166,
-            "replay_fallback_count": 0,
-        }
-    task_root.mkdir(parents=True, exist_ok=True)
-    (task_root / "task-receipt.json").write_text(
-        json.dumps(
-            {
-                "task_id": "gelu_backward",
-                "op": "gelu_backward",
-                "family": "elementwise-backward",
-                "classification": "C",
-                "classification_detail": {
-                    "reason": "topology reduction without stable >=5% speedup"
-                },
-                "correctness": {"status": "PASS"},
-                "correctness_isolation": {"candidate": {"status": "PASS"}},
-                "cells": cells,
-            }
-        ),
-        encoding="utf-8",
-    )
-
-
-class ReaccountTests(unittest.TestCase):
-    def test_input_construction_reduction_is_demoted_to_b(self):
-        with tempfile.TemporaryDirectory() as temporary:
-            base = Path(temporary)
-            run_root = base / "sealed"
-            run_root.mkdir()
-            _seal_run(run_root)
-            result = TOOL.reaccount(run_root, base / "derived", 1)
-
-        self.assertEqual(result["changed_task_ids"], ["gelu_backward"])
-        task = result["tasks"][0]
-        self.assertEqual(task["sealed_classification"], "C")
-        self.assertEqual(task["reaccounted_classification"], "B")
-        self.assertEqual(task["sealed_convert"]["a_off"], 4)
-        self.assertEqual(task["reaccounted_convert"]["a_off"], 0)
-        self.assertEqual(
-            task["excluded_unmeasured_files"], ["s/randn_kernel/w16s3.ttgir"]
-        )
-
-    def test_a_zero_floor_reproduces_the_sealed_verdict(self):
-        with tempfile.TemporaryDirectory() as temporary:
-            base = Path(temporary)
-            run_root = base / "sealed"
-            run_root.mkdir()
-            _seal_run(run_root)
-            result = TOOL.reaccount(run_root, base / "derived", 0)
-
-        # Floor 0 admits the once-launched builder again, so the derived verdict
-        # must match the sealed one. That equivalence is what makes the scoped
-        # count auditable rather than an unexplained new number.
-        self.assertEqual(result["changed_task_ids"], [])
-        self.assertEqual(result["tasks"][0]["reaccounted_classification"], "C")
-
-    def test_derived_output_is_written_and_never_inside_the_sealed_run(self):
-        with tempfile.TemporaryDirectory() as temporary:
-            base = Path(temporary)
-            run_root = base / "sealed"
-            run_root.mkdir()
-            _seal_run(run_root)
-
-            with self.assertRaises(TOOL.ReaccountError):
-                TOOL.reaccount(run_root, run_root / "derived", 1)
-            with self.assertRaises(TOOL.ReaccountError):
-                TOOL.reaccount(run_root, run_root, 1)
-
-            output = base / "derived"
-            TOOL.reaccount(run_root, output, 1)
-            self.assertTrue((output / "reaccount-receipt.json").is_file())
-            self.assertTrue((output / "reaccount-matrix.csv").is_file())
-            # The sealed receipt must survive untouched.
-            sealed = json.loads(
-                (run_root / "tasks" / "gelu_backward" / "task-receipt.json").read_text(
-                    encoding="utf-8"
-                )
-            )
-            self.assertEqual(sealed["classification"], "C")
-            self.assertEqual(sealed["cells"]["a_off"]["convert_count"], 4)
-
-    def test_a_directory_without_a_campaign_receipt_is_rejected(self):
-        with tempfile.TemporaryDirectory() as temporary:
-            base = Path(temporary)
-            with self.assertRaises(TOOL.ReaccountError):
-                TOOL.reaccount(base, base.parent / "derived", 1)
-
-
-if __name__ == "__main__":
-    unittest.main()