docs: document the MUSA RLC campaign

Covers the three screen cells and why replay locks them to one variable, how to
freeze and run a campaign, what the runner refuses to start on, the six
classifications, and the three scoping rules that decide whether a verdict
means anything.

Also states what a terminal campaign does not prove, since the manifest fixes
production_default_policy to unchanged-off and a passing run is not an argument
for flipping it.

_cudagraph_plugin.py

@@ -1,4 +1,4 @@
-"""pytest plugin: swap `triton.testing.do_bench` for graph-capture timing.
+"""pytest plugin: swap `triton.testing.do_bench` for cudagraph-based 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,18 +8,9 @@ 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 RNG
-state so that an aborted capture cannot poison the process-wide generator
+measurement unstable run-to-run). The capture itself runs under a scratch CUDA
+RNG 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
@@ -33,33 +24,8 @@ import triton.testing as _tt
 _ORIGINAL_DO_BENCH = _tt.do_bench
 
 
-def _device_interface():
-    """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):
+def _is_fatal_cuda_error(exc: Exception) -> bool:
+    """True for CUDA 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.
 
@@ -81,12 +47,8 @@ def _is_fatal_device_error(exc: Exception) -> bool:
     return any(m in msg for m in markers)
 
 
-# Retained under its historical name; the marker list is backend independent.
-_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.
+def _warmup_before_capture(fn, warmup_ms, grad_to_none):
+    """Warm fn up before cudagraph capture.
 
     do_bench_cudagraph only warms up 5 iterations internally — too few for the
     MoE kernels, where the first calls trigger Triton autotune config
@@ -104,6 +66,8 @@ def _warmup_before_capture(di, 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:
@@ -112,23 +76,23 @@ def _warmup_before_capture(di, fn, warmup_ms, grad_to_none):
 
     # First call absorbs autotune/JIT compilation (can take seconds); discard it.
     _run()
-    di.synchronize()
+    torch.cuda.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()
-    di.synchronize()
+    torch.cuda.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()
-    di.synchronize()
+    torch.cuda.synchronize()
 
 
-def _capture_with_scratch_rng(di, call):
-    """Run a graph capture under a throwaway device RNG state.
+def _capture_with_scratch_rng(call):
+    """Run a graph capture under a throwaway CUDA 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
@@ -140,21 +104,12 @@ def _capture_with_scratch_rng(di, 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.
-
-    An interface that does not expose default_generators has no such registration
-    to protect against (torch_musa is one), so the capture simply runs.
-    """
+    is left untouched."""
     import torch
-
-    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()
+    dev = torch.cuda.current_device()
+    gen = torch.cuda.default_generators[dev]
     saved = gen.graphsafe_get_state()
-    scratch = torch.Generator(device=torch.device(gen.device))
+    scratch = torch.Generator(device="cuda")
     scratch.manual_seed(gen.initial_seed())
     gen.graphsafe_set_state(scratch)
     try:
@@ -163,85 +118,24 @@ def _capture_with_scratch_rng(di, 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 graph capture."""
+    """Drop-in replacement for triton.testing.do_bench using cudagraph capture."""
     op = os.environ.get("FLAGGEMS_PERF_CURRENT_OP", "<unknown>")
-    phase = "sync"  # which step rejected graph capture, for the fallback marker
+    phase = "sync"  # which step rejected cudagraph, for the fallback marker
     try:
-        di = _device_interface()
-        # The graph path runs fn on a fresh side stream without syncing with the
-        # default stream first; input tensors produced just before (quantize/topk
-        # in the input iter) may still be in flight, and racing on them
-        # intermittently kills the context (illegal instruction from a device
-        # trap on garbage indices). Sync before switching streams.
-        di.synchronize()
+        # do_bench_cudagraph runs fn on a fresh side stream without syncing
+        # with the default stream first; input tensors produced just before
+        # (quantize/topk in the input iter) may still be in flight, and racing
+        # on them intermittently kills the context (illegal instruction from a
+        # device trap on garbage indices). Sync before switching streams.
+        import torch
+        torch.cuda.synchronize()
         phase = "warmup"
         # Settle autotune/JIT before capture (rationale: _warmup_before_capture).
-        _warmup_before_capture(di, fn, warmup, grad_to_none)
+        _warmup_before_capture(fn, warmup, grad_to_none)
         phase = "capture"
-        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,
+        return _capture_with_scratch_rng(lambda: _tt.do_bench_cudagraph(
             fn,
             rep=rep,
             grad_to_none=grad_to_none,
@@ -251,7 +145,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_device_error(exc):
+        if _is_fatal_cuda_error(exc):
             raise
         reason = type(exc).__name__
         detail = " ".join(str(exc).split())[:160]
@@ -274,10 +168,5 @@ 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
-    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}",
+    print(f"{tag('[cudagraph-plugin]')} triton.testing.do_bench → do_bench_cudagraph",
           file=sys.stderr, flush=True)

ab_rlc_test.sh

@@ -1,68 +0,0 @@
-#!/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,82 +165,12 @@ 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, nargs="?",
-                   help="run_batch.py 的产物目录")
+    p.add_argument("batch_dir", type=Path, 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

@@ -0,0 +1,103 @@
+# 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

@@ -0,0 +1,551 @@
+{
+  "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

@@ -0,0 +1,478 @@
+#!/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

@@ -0,0 +1,223 @@
+#!/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

@@ -0,0 +1,834 @@
+#!/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

@@ -0,0 +1,111 @@
+#!/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

@@ -0,0 +1,567 @@
+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

@@ -0,0 +1,157 @@
+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()