bench: partition operators across cards by measured cost, and rescore in place

Two pieces the parallel campaign needs.

partition.py assigns screened operators to cards longest-first. run_batch.py's
existing hash partition ignores weight: on this batch it puts the critical path
at 120 minutes and leaves a card idle, where greedy gets 70. The floor is set
by the single heaviest operator — topk alone is 70.6 minutes against an ideal
share of 45.4 — so partitioning cannot beat it, only keep the other cards from
waiting. Weights come from screen.tsv's elapsed_before + elapsed_after, which
is one pair, i.e. 2 cells; N rounds is 2N cells, so the estimate is N times the
weight. Getting that wrong reported topk as 11.8 minutes.

RESCORE=<root> re-judges an existing screening root against the current
fingerprint and rewrites its screen.tsv, keeping the original alongside. The
criterion has already changed twice — index.tsv used to be hashed, and the
switch's own ttg.rlc-* annotations had to be stripped — and the second change
moved 24 operators from 23 IR_CHANGED to 16. Rescoring the artifacts already on
disk takes seconds; re-running that screen takes 79 minutes.

test_backends.py

@@ -659,5 +659,80 @@ class ForeignVendorModuleTests(unittest.TestCase):
         self.assertEqual(foreign_vendor_modules("mthreads", mods), [])
 
 
+class PartitionTests(unittest.TestCase):
+    """按耗时把算子分到各卡。hash 分片在这批活上比它慢 50 分钟。"""
+
+    HEADER = "op\top_file\telapsed_before\telapsed_after\tverdict"
+
+    def _load(self, assignment):
+        return sorted(sum(w for _, w in v) for v in assignment.values())
+
+    def _screen(self, tmp, rows, header=None):
+        p = Path(tmp, "screen.tsv")
+        p.write_text((header or self.HEADER) + "\n" + "\n".join(rows) + "\n",
+                     encoding="utf-8")
+        return p
+
+    def test_heavy_first_keeps_cards_even(self):
+        from partition import plan
+        items = [("a", 10), ("b", 9), ("c", 8), ("d", 7), ("e", 6), ("f", 5)]
+        self.assertEqual(self._load(plan(items, ["0", "1", "2"])), [15, 15, 15])
+
+    def test_one_item_heavier_than_the_ideal_share_sets_the_floor(self):
+        # topk 单个 70 分钟、理想均分 45——加卡不解决，分片只保证别人不空等
+        from partition import imbalance, plan
+        items = [("topk", 70)] + [(f"op{i}", 5) for i in range(10)]
+        a = plan(items, [str(i) for i in range(8)])
+        self.assertEqual(max(sum(w for _, w in v) for v in a.values()), 70)
+        self.assertGreater(imbalance(a), 1.0)
+
+    def test_more_cards_than_operators_leaves_some_empty(self):
+        from partition import plan
+        a = plan([("a", 1), ("b", 1)], ["0", "1", "2", "3"])
+        self.assertEqual(sum(1 for v in a.values() if not v), 2)
+
+    def test_same_input_gives_the_same_plan(self):
+        from partition import plan
+        items = [("a", 5), ("b", 5), ("c", 5), ("d", 5)]
+        first = plan(items, ["0", "1"])
+        for _ in range(5):
+            self.assertEqual(plan(list(reversed(items)), ["0", "1"]), first)
+
+    def test_no_cards_is_the_callers_error(self):
+        from partition import plan
+        with self.assertRaises(ValueError):
+            plan([("a", 1)], [])
+
+    def test_perfect_balance_is_one(self):
+        from partition import imbalance, plan
+        self.assertEqual(imbalance(plan([("a", 5), ("b", 5)], ["0", "1"])), 1.0)
+
+    def test_verdict_filters_and_reports_what_it_dropped(self):
+        from partition import read_screen
+        with tempfile.TemporaryDirectory() as tmp:
+            f = self._screen(tmp, ["a\ta\t10\t10\tIR_CHANGED",
+                                   "b\tbb\t5\t5\tNO_IR_DIFF",
+                                   "c\tc\t1\t1\tNO_TRITON"])
+            items, files, skipped = read_screen(f)
+            self.assertEqual(items, [("a", 20.0)])
+            self.assertEqual(files["a"], "a")
+            self.assertEqual(len(skipped), 2)
+
+    def test_op_file_defaults_to_op_name(self):
+        from partition import read_screen
+        with tempfile.TemporaryDirectory() as tmp:
+            f = self._screen(tmp, ["a\t\t1\t1\tIR_CHANGED"])
+            _, files, _ = read_screen(f)
+            self.assertEqual(files["a"], "a")
+
+    def test_missing_elapsed_falls_back_to_equal_weights(self):
+        from partition import read_screen
+        with tempfile.TemporaryDirectory() as tmp:
+            f = self._screen(tmp, ["a\ta\tIR_CHANGED", "b\tb\tIR_CHANGED"],
+                             header="op\top_file\tverdict")
+            items, _, _ = read_screen(f)
+            self.assertEqual([w for _, w in items], [1.0, 1.0])
+
+
 if __name__ == "__main__":
     unittest.main()