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.

.gitattributes

@@ -0,0 +1,12 @@
+# Everything here runs on Linux accelerator hosts, and campaign receipts record
+# the SHA-256 of the harness files they were produced by. A CRLF checkout on
+# Windows would both break the shell drivers and change every recorded digest,
+# so pin the text formats this repository ships.
+*.py text eol=lf
+*.sh text eol=lf
+*.json text eol=lf
+*.yaml text eol=lf
+*.csv text eol=lf
+*.tsv text eol=lf
+*.txt text eol=lf
+*.md text eol=lf

_musa_benchmark_compat_plugin.py

@@ -0,0 +1,89 @@
+"""Narrow compatibility fixes for the frozen MUSA product corpus.
+
+Several upstream FlagGems tests gate themselves on ``torch.cuda.is_available()``
+even though the code under test runs on any accelerator FlagGems dispatches to.
+On MThreads those guards skip the test instead of exercising the operator, which
+turns a working path into a verdict of "not measured".
+
+Every override here is pinned to one file and one exact skip reason, applies
+only when FlagGems has dispatched to the MThreads vendor, and is only added
+after the unskipped test was observed to pass on the device.
+
+Two files were probed on S5000 and deliberately left out, because their guards
+are not stale:
+
+- ``test_cutlass_scaled_mm.py`` gates on vLLM as well as CUDA, so removing the
+  guard would raise on the missing dependency instead of measuring anything.
+- ``test_grid_sample.py`` builds its inputs through a helper whose device
+  argument defaults to the literal ``"cuda"``. Its guard is consistent with its
+  own body: unskipped, all 90 collected cases fail in ``aten::empty`` on a CUDA
+  backend that does not exist here. Making that file portable means editing
+  upstream test source, which is outside what a compatibility plugin may do.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+
+# file name -> exact skipif reasons that are stale on a dispatching backend.
+_STALE_CUDA_GUARDS: dict[str, frozenset[str]] = {
+    "test_fft.py": frozenset({"CUDA is not available"}),
+    "test_top_k_per_row_decode.py": frozenset({"CUDA device required"}),
+}
+
+# Guards that are only stale for one test in the file stay pinned to it.
+_PINNED_TESTS: dict[str, str] = {
+    "test_fft.py": "test_perf_fft",
+}
+
+
+def _is_stale_cuda_guard(item, mark) -> bool:
+    if mark.name != "skipif":
+        return False
+    name = Path(str(item.path)).name
+    reasons = _STALE_CUDA_GUARDS.get(name)
+    if reasons is None or mark.kwargs.get("reason") not in reasons:
+        return False
+    pinned = _PINNED_TESTS.get(name)
+    return pinned is None or item.name == pinned
+
+
+# Retained under its historical name: the sealed campaign's tests reference it.
+def _is_stale_fft_cuda_guard(item, mark) -> bool:
+    return Path(str(item.path)).name == "test_fft.py" and _is_stale_cuda_guard(
+        item, mark
+    )
+
+
+def pytest_collection_modifyitems(items):
+    import flag_gems
+
+    if flag_gems.vendor_name != "mthreads":
+        return
+    removed: dict[str, int] = {}
+    # A guard can sit on the test, its class, or the module's pytestmark, and
+    # skipif is resolved by walking that whole chain. Strip every level.
+    seen: set[int] = set()
+    for item in items:
+        for node in item.listchain():
+            if id(node) in seen:
+                continue
+            seen.add(id(node))
+            markers = getattr(node, "own_markers", None)
+            if not markers:
+                continue
+            kept = [mark for mark in markers if not _is_stale_cuda_guard(item, mark)]
+            dropped = len(markers) - len(kept)
+            if dropped:
+                markers[:] = kept
+                name = Path(str(item.path)).name
+                removed[name] = removed.get(name, 0) + dropped
+    for name, count in sorted(removed.items()):
+        print(
+            f"[musa-benchmark-compat] removed {count} stale CUDA-only skip "
+            f"guard(s) from {name}",
+            file=sys.stderr,
+            flush=True,
+        )

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()