feat: re-account a sealed MUSA RLC run without re-running it

A sealed run already holds every TTGIR dump and its launch index, so a change
in how convert topology is counted can be applied to it directly instead of
spending accelerator time.

The tool re-reads each cell's dump, recomputes the scoped count, re-runs the
campaign classifier against the sealed correctness verdicts, and writes a
derived receipt and matrix outside the run root. It refuses to write inside a
sealed run and rejects a directory with no campaign receipt.

A derived verdict can only demote a topology claim. Promoting one to G or C
needs reversed-order timing, which a sealed run holds only for tasks that
already showed a reduction, so a reduction appearing for the first time under
scoping is reported as X for missing round-b evidence rather than guessed.

Run against the sealed 32-workload S5000 campaign, a floor of 1 changes exactly
one task: gelu_backward moves from C to B, and the only excluded files in the
whole campaign are its five once-launched randn_kernel dumps. A floor of 0
reproduces the sealed counts with no task changed, which is the equivalence
that makes the scoped number auditable rather than merely different.

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