基于AI Agent开发范式的国产GPU大模型推理算子库优化/FlashInfer Benchmark实战：从性能基线到XPU-OJ冒烟提交.md

@@ -366,7 +366,7 @@ pip install pandas
 
 ```bash
 # 检查脚本文件
-python -c "import os; scripts = ['bench_common.py', 'benchmark_result.py', 'summarize_results.py', 'bench_batch_decode.py', 'bench_batch_prefill_paged.py', 'bench_batch_prefill_ragged.py', 'bench_batch_mla.py']; [print(f'✓ {s}') if os.path.exists(s) else print(f'✗ {s} missing') for s in scripts]"
+python -c "import os; scripts = ['bench_common.py', 'bench_batch_decode.py', 'bench_batch_prefill_paged.py', 'bench_batch_prefill_ragged.py', 'bench_batch_mla.py']; [print(f'✓ {s}') if os.path.exists(s) else print(f'✗ {s} missing') for s in scripts]"
 
 # 测试脚本导入
 python -c "from bench_common import setup_workspace, get_csv_path; print('脚本导入正常')"
@@ -376,8 +376,6 @@ python -c "from bench_common import setup_workspace, get_csv_path; print('脚本
 
 ```plaintext
 ✓ bench_common.py
-✓ benchmark_result.py
-✓ summarize_results.py
 ✓ bench_batch_decode.py
 ✓ bench_batch_prefill_paged.py
 ✓ bench_batch_prefill_ragged.py
@@ -416,57 +414,6 @@ Results saved to BatchPrefillWithRaggedKVCacheWrapper_20260626_xxxxxx.csv
 | `out of memory` | 减小 `batch_size` 或 `seq_len` 参数 |
 | 运行时间过长 | 脚本会自动调整重复次数，耐心等待 |
 
-Benchmark 会逐个执行 case。单个 case 出错时，脚本会把 `status=failed`、异常类型和错误信息写入 CSV 后继续执行；只要存在失败 case，CSV 仍会保存，但进程最终返回退出码 `1`。
-
-***
-
-**结构化 CSV 与汇总报告：**
-
-新版 CSV 使用 `schema_version=1`，通用字段如下：
-
-| 字段 | 含义 |
-| --- | --- |
-| `schema_version` | 结果格式版本，当前为 `1` |
-| `api` | FlashInfer wrapper 名称 |
-| case 参数列 | 如 `batch_size`、`seq_len`、`head_dim` 等 |
-| `status` | `ok` 或 `failed` |
-| `time_ms` / `bandwidth_GB_s` / `tflops` | 成功 case 的性能指标 |
-| `error_type` / `error` | 失败 case 的异常类型和错误信息 |
-
-批量汇总一个或多个结果文件：
-
-```bash
-python summarize_results.py results/*.csv \
-  --output flashinfer_benchmark_summary.md
-```
-
-显式比较优化前后的同一组 case：
-
-```bash
-python summarize_results.py \
-  --baseline baseline/*.csv \
-  --candidate candidate/*.csv \
-  --output flashinfer_benchmark_regression.md
-```
-
-默认比较模式只报告变化。CI 中可增加门禁，例如任一匹配 case 的耗时上升或带宽、TFLOPS 下降超过 5% 时返回退出码 `1`：
-
-```bash
-python summarize_results.py \
-  --baseline baseline/*.csv \
-  --candidate candidate/*.csv \
-  --fail-on-regression 5 \
-  --output flashinfer_benchmark_regression.md
-```
-
-| 退出码 | 含义 |
-| --- | --- |
-| `0` | 输入有效，且没有 benchmark/candidate 失败或触发回归门禁 |
-| `1` | 存在失败 case、候选缺失基线 case，或触发回归门禁 |
-| `2` | 参数、CSV 文件或 schema 无效 |
-
-旧版 CSV 没有 `schema_version/status/error` 字段时，只有在表头符合已知 FlashInfer benchmark 格式且性能指标有效的情况下才会被兼容读取；报告会将其标记为 `legacy_inferred_ok`，表示成功状态来自格式推断而非原始记录。
-
 ***
 
 **查看结果命令示例：**
@@ -1037,7 +984,7 @@ FlashInfer 方向包含 **4 个可选算子题目**，均属于同一比赛通
 | --- | --- | --- |
 | Benchmark 运行时间过长 | 参数组合过多， workload 较大 | 耐心等待，脚本会自动调整重复次数 |
 | `KeyError: 'BatchPrefillWithPagedKVCacheKernel'` | profiler 未捕获目标 kernel | 检查 `target_kernels` 配置是否正确 |
-| CSV 文件为空或汇总器返回退出码 `2` | 测试未正常完成或 CSV schema 无效 | 查看汇总报告的 `Input Errors`，检查文件完整性后重新运行 |
+| CSV 文件为空 | 测试未正常完成 | 检查 GPU 显存是否充足，重新运行 |
 
 ### 8.4 代码问题
 
@@ -1503,4 +1450,4 @@ extern "C" void run_kernel(
 }
 ```
 
-[*回退到 Step 8*](#step%208提交%20oj%20冒烟代码)
+[*回退到 Step 8*](#step%208提交%20oj%20冒烟代码)

基于AI Agent开发范式的国产GPU大模型推理算子库优化/operator_task_package/flashinfer_task_package/benchmark/bench_batch_decode.py

@@ -8,16 +8,7 @@ import pandas as pd
 import torch
 
 import flashinfer
-from benchmark_result import execute_benchmark_case, has_failures, STATUS_OK
+from bench_common import dtype, page_block_size, setup_workspace, setup_paged_kv_indptr, run_with_profiler, get_csv_path, compute_reps
-from bench_common import (
-    compute_reps,
-    dtype,
-    get_csv_path,
-    page_block_size,
-    run_with_profiler,
-    setup_paged_kv_indptr,
-    setup_workspace,
-)
 
 target_kernels = ["BatchPrefillWithPagedKVCacheKernel"]
 
@@ -78,39 +69,22 @@ def run_benchmark():
     for idx, (bs, sl_kv, hd) in enumerate(test_cases, 1):
         num_qo_heads = 32
         num_kv_heads = 8 if hd == 64 else 4
-        case = {
+        ms, io, flops = bench_batch_decode(bs, sl_kv, num_qo_heads, num_kv_heads, hd, page_block_size)
+        bw = io / ms / 1e6
+        tflops = flops / ms / 1e9
+        records.append({
+            "api": api_name,
             "batch_size": bs,
             "seq_len_q": 1,
             "seq_len_kv": sl_kv,
             "num_qo_heads": num_qo_heads,
             "num_kv_heads": num_kv_heads,
             "head_dim": hd,
-        }
+            "time_ms": ms,
-        record = execute_benchmark_case(
+            "bandwidth_GB_s": bw,
-            api_name,
+            "tflops": tflops,
-            case,
+        })
-            lambda: bench_batch_decode(
+        print(f"  [{idx}/{total_cases}] bs={bs}, kv_len={sl_kv}, hd={hd}: {ms:.3f}ms, {bw:.2f} GB/s, {tflops:.2f} TFLOPs")
-                bs,
-                sl_kv,
-                num_qo_heads,
-                num_kv_heads,
-                hd,
-                page_block_size,
-            ),
-        )
-        records.append(record)
-        if record["status"] == STATUS_OK:
-            print(
-                f"  [{idx}/{total_cases}] bs={bs}, kv_len={sl_kv}, hd={hd}: "
-                f"{record['time_ms']:.3f}ms, "
-                f"{record['bandwidth_GB_s']:.2f} GB/s, "
-                f"{record['tflops']:.2f} TFLOPs"
-            )
-        else:
-            print(
-                f"  [{idx}/{total_cases}] bs={bs}, kv_len={sl_kv}, hd={hd}: "
-                f"FAILED: {record['error_type']}: {record['error']}"
-            )
 
     return records
 
@@ -124,6 +98,4 @@ if __name__ == "__main__":
     df = pd.DataFrame(records)
     csv_path = get_csv_path("BatchDecodeWithPagedKVCacheWrapper")
     df.to_csv(csv_path, index=False)
-    print(f"\nResults saved to {csv_path}")
+    print(f"\nResults saved to {csv_path}")
-    if has_failures(records):
-        raise SystemExit(1)

基于AI Agent开发范式的国产GPU大模型推理算子库优化/operator_task_package/flashinfer_task_package/benchmark/bench_batch_mla.py

@@ -8,14 +8,7 @@ import pandas as pd
 import torch
 
 import flashinfer
-from benchmark_result import execute_benchmark_case, has_failures, STATUS_OK
+from bench_common import dtype, page_block_size, setup_workspace, run_with_profiler, get_csv_path, compute_reps
-from bench_common import (
-    compute_reps,
-    dtype,
-    get_csv_path,
-    run_with_profiler,
-    setup_workspace,
-)
 
 target_kernels = ["BatchMLAPagedAttentionKernel"]
 
@@ -86,34 +79,21 @@ def run_benchmark():
 
     print(f"[{api_name}] Starting benchmark, total cases: {total_cases}")
     for idx, (num_heads, bs, sl) in enumerate(test_cases, 1):
-        case = {
+        ms, io, flops = bench_batch_mla_paged_attention(bs, sl, num_heads, head_dim_ckv, head_dim_kpe)
+        bw = io / ms / 1e6
+        tflops = flops / ms / 1e9
+        records.append({
+            "api": api_name,
             "batch_size": bs,
             "seq_len": sl,
             "num_heads": num_heads,
             "head_dim_ckv": head_dim_ckv,
             "head_dim_kpe": head_dim_kpe,
-        }
+            "time_ms": ms,
-        record = execute_benchmark_case(
+            "bandwidth_GB_s": bw,
-            api_name,
+            "tflops": tflops,
-            case,
+        })
-            lambda: bench_batch_mla_paged_attention(
+        print(f"  [{idx}/{total_cases}] bs={bs}, sl={sl}, num_heads={num_heads}: {ms:.3f}ms, {bw:.2f} GB/s, {tflops:.2f} TFLOPs")
-                bs, sl, num_heads, head_dim_ckv, head_dim_kpe
-            ),
-        )
-        records.append(record)
-        if record["status"] == STATUS_OK:
-            print(
-                f"  [{idx}/{total_cases}] bs={bs}, sl={sl}, "
-                f"num_heads={num_heads}: {record['time_ms']:.3f}ms, "
-                f"{record['bandwidth_GB_s']:.2f} GB/s, "
-                f"{record['tflops']:.2f} TFLOPs"
-            )
-        else:
-            print(
-                f"  [{idx}/{total_cases}] bs={bs}, sl={sl}, "
-                f"num_heads={num_heads}: FAILED: "
-                f"{record['error_type']}: {record['error']}"
-            )
 
     return records
 
@@ -127,6 +107,4 @@ if __name__ == "__main__":
     df = pd.DataFrame(records)
     csv_path = get_csv_path("BatchMLAPagedAttentionWrapper")
     df.to_csv(csv_path, index=False)
-    print(f"\nResults saved to {csv_path}")
+    print(f"\nResults saved to {csv_path}")
-    if has_failures(records):
-        raise SystemExit(1)

基于AI Agent开发范式的国产GPU大模型推理算子库优化/operator_task_package/flashinfer_task_package/benchmark/bench_batch_prefill_paged.py

@@ -16,7 +16,6 @@ from bench_common import (
     get_csv_path,
     compute_reps,
 )
-from benchmark_result import execute_benchmark_case, has_failures, STATUS_OK
 
 target_kernels = ["BatchPrefillWithPagedKVCacheKernel"]
 
@@ -96,33 +95,27 @@ def run_benchmark():
     for idx, (head_dim, bs, sl) in enumerate(test_cases, 1):
         num_qo_heads = 32
         num_kv_heads = 8 if head_dim == 64 else 4
-        case = {
+        ms, io, flops = bench_batch_prefill_with_paged_kv_cache(
-            "batch_size": bs,
+            bs, sl, num_qo_heads, num_kv_heads, head_dim
-            "seq_len": sl,
+        )
-            "num_qo_heads": num_qo_heads,
+        bw = io / ms / 1e6
-            "num_kv_heads": num_kv_heads,
+        tflops = flops / ms / 1e9
-            "head_dim": head_dim,
+        records.append(
-        }
+            {
-        record = execute_benchmark_case(
+                "api": api_name,
-            api_name,
+                "batch_size": bs,
-            case,
+                "seq_len": sl,
-            lambda: bench_batch_prefill_with_paged_kv_cache(
+                "num_qo_heads": num_qo_heads,
-                bs, sl, num_qo_heads, num_kv_heads, head_dim
+                "num_kv_heads": num_kv_heads,
-            ),
+                "head_dim": head_dim,
+                "time_ms": ms,
+                "bandwidth_GB_s": bw,
+                "tflops": tflops,
+            }
+        )
+        print(
+            f"  [{idx}/{total_cases}] bs={bs}, sl={sl}, hd={head_dim}: {ms:.3f}ms, {bw:.2f} GB/s, {tflops:.2f} TFLOPs"
         )
-        records.append(record)
-        if record["status"] == STATUS_OK:
-            print(
-                f"  [{idx}/{total_cases}] bs={bs}, sl={sl}, hd={head_dim}: "
-                f"{record['time_ms']:.3f}ms, "
-                f"{record['bandwidth_GB_s']:.2f} GB/s, "
-                f"{record['tflops']:.2f} TFLOPs"
-            )
-        else:
-            print(
-                f"  [{idx}/{total_cases}] bs={bs}, sl={sl}, hd={head_dim}: "
-                f"FAILED: {record['error_type']}: {record['error']}"
-            )
 
     return records
 
@@ -138,5 +131,3 @@ if __name__ == "__main__":
     csv_path = get_csv_path("BatchPrefillWithPagedKVCacheWrapper")
     df.to_csv(csv_path, index=False)
     print(f"\nResults saved to {csv_path}")
-    if has_failures(records):
-        raise SystemExit(1)

基于AI Agent开发范式的国产GPU大模型推理算子库优化/operator_task_package/flashinfer_task_package/benchmark/bench_batch_prefill_ragged.py

@@ -15,7 +15,6 @@ from bench_common import (
     get_csv_path,
     compute_reps,
 )
-from benchmark_result import execute_benchmark_case, has_failures, STATUS_OK
 
 target_kernels = [
     "BatchPrefillWithRaggedKVCacheKernel",
@@ -98,41 +97,28 @@ def run_benchmark():
     for idx, ((head_dim_qk, head_dim_vo), bs, sl) in enumerate(test_cases, 1):
         num_qo_heads = 32
         num_kv_heads = 4
-        case = {
+        ms, io, flops = bench_batch_prefill_with_ragged_kv_cache(
-            "batch_size": bs,
+            bs, sl, num_qo_heads, num_kv_heads, head_dim_qk, head_dim_vo
-            "seq_len": sl,
+        )
-            "num_qo_heads": num_qo_heads,
+        bw = io / ms / 1e6
-            "num_kv_heads": num_kv_heads,
+        tflops = flops / ms / 1e9
-            "head_dim_qk": head_dim_qk,
+        records.append(
-            "head_dim_vo": head_dim_vo,
+            {
-        }
+                "api": api_name,
-        record = execute_benchmark_case(
+                "batch_size": bs,
-            api_name,
+                "seq_len": sl,
-            case,
+                "num_qo_heads": num_qo_heads,
-            lambda: bench_batch_prefill_with_ragged_kv_cache(
+                "num_kv_heads": num_kv_heads,
-                bs,
+                "head_dim_qk": head_dim_qk,
-                sl,
+                "head_dim_vo": head_dim_vo,
-                num_qo_heads,
+                "time_ms": ms,
-                num_kv_heads,
+                "bandwidth_GB_s": bw,
-                head_dim_qk,
+                "tflops": tflops,
-                head_dim_vo,
+            }
-            ),
+        )
+        print(
+            f"  [{idx}/{total_cases}] bs={bs}, sl={sl}, hd=[{head_dim_qk},{head_dim_vo}]: {ms:.3f}ms, {bw:.2f} GB/s, {tflops:.2f} TFLOPs"
         )
-        records.append(record)
-        if record["status"] == STATUS_OK:
-            print(
-                f"  [{idx}/{total_cases}] bs={bs}, sl={sl}, "
-                f"hd=[{head_dim_qk},{head_dim_vo}]: "
-                f"{record['time_ms']:.3f}ms, "
-                f"{record['bandwidth_GB_s']:.2f} GB/s, "
-                f"{record['tflops']:.2f} TFLOPs"
-            )
-        else:
-            print(
-                f"  [{idx}/{total_cases}] bs={bs}, sl={sl}, "
-                f"hd=[{head_dim_qk},{head_dim_vo}]: FAILED: "
-                f"{record['error_type']}: {record['error']}"
-            )
 
     return records
 
@@ -148,5 +134,3 @@ if __name__ == "__main__":
     csv_path = get_csv_path("BatchPrefillWithRaggedKVCacheWrapper")
     df.to_csv(csv_path, index=False)
     print(f"\nResults saved to {csv_path}")
-    if has_failures(records):
-        raise SystemExit(1)

基于AI Agent开发范式的国产GPU大模型推理算子库优化/operator_task_package/flashinfer_task_package/benchmark/benchmark_result.py

@@ -1,80 +0,0 @@
-"""Shared result schema and helpers for FlashInfer benchmarks."""
-
-import math
-
-
-SCHEMA_VERSION = "1"
-STATUS_OK = "ok"
-STATUS_FAILED = "failed"
-STATUS_LEGACY_OK = "legacy_inferred_ok"
-
-NUMERIC_COLUMNS = ("time_ms", "bandwidth_GB_s", "tflops")
-RESULT_COLUMNS = (
-    "schema_version",
-    "api",
-    "status",
-    *NUMERIC_COLUMNS,
-    "error_type",
-    "error",
-)
-RESERVED_COLUMNS = frozenset((*RESULT_COLUMNS, "_source"))
-
-
-def _validate_metric(name, value, *, positive=False):
-    number = float(value)
-    if not math.isfinite(number):
-        raise ValueError(f"{name} must be finite, got {value!r}")
-    if positive and number <= 0:
-        raise ValueError(f"{name} must be greater than zero, got {value!r}")
-    if not positive and number < 0:
-        raise ValueError(f"{name} must not be negative, got {value!r}")
-    return number
-
-
-def execute_benchmark_case(api, case, benchmark_fn):
-    """Execute one benchmark case and return a schema-v1 result record."""
-    conflicting = RESERVED_COLUMNS.intersection(case)
-    if conflicting:
-        names = ", ".join(sorted(conflicting))
-        raise ValueError(f"case fields use reserved result columns: {names}")
-
-    record = {
-        "schema_version": SCHEMA_VERSION,
-        "api": api,
-        **case,
-    }
-
-    try:
-        time_ms, io_bytes, flops = benchmark_fn()
-        time_ms = _validate_metric("time_ms", time_ms, positive=True)
-        io_bytes = _validate_metric("io_bytes", io_bytes)
-        flops = _validate_metric("flops", flops)
-        bandwidth = _validate_metric(
-            "bandwidth_GB_s", io_bytes / time_ms / 1e6
-        )
-        tflops = _validate_metric("tflops", flops / time_ms / 1e9)
-    except Exception as exc:
-        return {
-            **record,
-            "status": STATUS_FAILED,
-            "time_ms": "",
-            "bandwidth_GB_s": "",
-            "tflops": "",
-            "error_type": type(exc).__name__,
-            "error": str(exc),
-        }
-
-    return {
-        **record,
-        "status": STATUS_OK,
-        "time_ms": time_ms,
-        "bandwidth_GB_s": bandwidth,
-        "tflops": tflops,
-        "error_type": "",
-        "error": "",
-    }
-
-
-def has_failures(records):
-    """Return whether a collection contains at least one failed case."""
-    return any(record.get("status") == STATUS_FAILED for record in records)

基于AI Agent开发范式的国产GPU大模型推理算子库优化/operator_task_package/flashinfer_task_package/benchmark/summarize_results.py

@@ -1,742 +0,0 @@
-"""Summarize and compare FlashInfer benchmark CSV files."""
-
-import argparse
-import csv
-import math
-from collections import Counter, defaultdict
-from dataclasses import dataclass, field
-from pathlib import Path
-
-from benchmark_result import (
-    NUMERIC_COLUMNS,
-    RESERVED_COLUMNS,
-    SCHEMA_VERSION,
-    STATUS_FAILED,
-    STATUS_LEGACY_OK,
-    STATUS_OK,
-)
-
-
-EXIT_OK = 0
-EXIT_BENCHMARK_FAILURE = 1
-EXIT_INPUT_ERROR = 2
-SUCCESS_STATUSES = frozenset((STATUS_OK, STATUS_LEGACY_OK))
-
-CASE_COLUMN_ORDER = (
-    "batch_size",
-    "seq_len",
-    "seq_len_q",
-    "seq_len_kv",
-    "num_heads",
-    "num_qo_heads",
-    "num_kv_heads",
-    "head_dim",
-    "head_dim_qk",
-    "head_dim_vo",
-    "head_dim_ckv",
-    "head_dim_kpe",
-)
-
-COMMON_LEGACY_COLUMNS = frozenset(("api", *NUMERIC_COLUMNS))
-LEGACY_SCHEMAS = (
-    COMMON_LEGACY_COLUMNS
-    | frozenset(
-        (
-            "batch_size",
-            "seq_len_q",
-            "seq_len_kv",
-            "num_qo_heads",
-            "num_kv_heads",
-            "head_dim",
-        )
-    ),
-    COMMON_LEGACY_COLUMNS
-    | frozenset(
-        (
-            "batch_size",
-            "seq_len",
-            "num_heads",
-            "head_dim_ckv",
-            "head_dim_kpe",
-        )
-    ),
-    COMMON_LEGACY_COLUMNS
-    | frozenset(
-        (
-            "batch_size",
-            "seq_len",
-            "num_qo_heads",
-            "num_kv_heads",
-            "head_dim",
-        )
-    ),
-    COMMON_LEGACY_COLUMNS
-    | frozenset(
-        (
-            "batch_size",
-            "seq_len",
-            "num_qo_heads",
-            "num_kv_heads",
-            "head_dim_qk",
-            "head_dim_vo",
-        )
-    ),
-)
-V1_REQUIRED_COLUMNS = frozenset(
-    (
-        "schema_version",
-        "api",
-        "status",
-        *NUMERIC_COLUMNS,
-        "error_type",
-        "error",
-    )
-)
-INTERNAL_COLUMNS = frozenset(("_source", "_line"))
-
-
-@dataclass
-class InputIssue:
-    source: str
-    message: str
-    group: str = "input"
-
-
-@dataclass
-class LoadResult:
-    group: str
-    paths: list
-    records: list = field(default_factory=list)
-    issues: list = field(default_factory=list)
-
-    @property
-    def valid_sources(self):
-        return {record["_source"] for record in self.records}
-
-
-def parse_float(value):
-    try:
-        number = float(value)
-    except (TypeError, ValueError):
-        return None
-    return number if math.isfinite(number) else None
-
-
-def _metric_error(record, column):
-    value = parse_float(record.get(column))
-    if value is None:
-        return f"{column} must be a finite number"
-    if column == "time_ms" and value <= 0:
-        return "time_ms must be greater than zero"
-    if column != "time_ms" and value < 0:
-        return f"{column} must not be negative"
-    record[column] = value
-    return None
-
-
-def _matches_known_schema(fieldnames):
-    return any(schema.issubset(fieldnames) for schema in LEGACY_SCHEMAS)
-
-
-def _validate_headers(fieldnames):
-    if not fieldnames:
-        return None, "CSV header is missing"
-    if any(name is None or not name for name in fieldnames):
-        return None, "CSV contains an empty header name"
-    if any(name != name.strip() for name in fieldnames):
-        return None, "CSV header names must not contain surrounding whitespace"
-    if len(fieldnames) != len(set(fieldnames)):
-        return None, "CSV contains duplicate header names"
-
-    names = frozenset(fieldnames)
-    has_v1_marker = "schema_version" in names or "status" in names
-    if has_v1_marker:
-        missing = sorted(V1_REQUIRED_COLUMNS - names)
-        if missing:
-            return None, f"schema v1 is missing columns: {', '.join(missing)}"
-        if not _matches_known_schema(names):
-            return None, "CSV does not match a known FlashInfer benchmark schema"
-        return "v1", None
-
-    if not _matches_known_schema(names):
-        return None, "CSV does not match a known legacy FlashInfer schema"
-    return "legacy", None
-
-
-def _validate_row(raw_row, schema, source, line_number):
-    if None in raw_row:
-        return None, "row has more values than the CSV header"
-    if any(value is None for value in raw_row.values()):
-        return None, "row has fewer values than the CSV header"
-
-    record = {key: value.strip() for key, value in raw_row.items()}
-    if not record.get("api"):
-        return None, "api must not be empty"
-
-    if schema == "v1":
-        if record.get("schema_version") != SCHEMA_VERSION:
-            return None, (
-                "unsupported schema_version "
-                f"{record.get('schema_version')!r}; expected {SCHEMA_VERSION!r}"
-            )
-        status = record.get("status")
-        if status not in (STATUS_OK, STATUS_FAILED):
-            return None, f"unsupported status {status!r}"
-        if status == STATUS_OK and (
-            record.get("error_type") or record.get("error")
-        ):
-            return None, "successful rows must not contain error details"
-        if status == STATUS_FAILED and not (
-            record.get("error_type") or record.get("error")
-        ):
-            return None, "failed rows must contain error_type or error"
-    else:
-        record["schema_version"] = ""
-        record["status"] = STATUS_LEGACY_OK
-        record["error_type"] = ""
-        record["error"] = ""
-
-    if record["status"] in SUCCESS_STATUSES:
-        for column in NUMERIC_COLUMNS:
-            error = _metric_error(record, column)
-            if error:
-                return None, error
-
-    record["_source"] = source
-    record["_line"] = line_number
-    return record, None
-
-
-def load_csv(path, group="input"):
-    source = str(path)
-    records = []
-    issues = []
-    try:
-        with path.open("r", encoding="utf-8-sig", newline="") as csv_file:
-            reader = csv.DictReader(csv_file)
-            schema, error = _validate_headers(reader.fieldnames)
-            if error:
-                return [], [InputIssue(source, error, group)]
-
-            for line_number, row in enumerate(reader, 2):
-                record, row_error = _validate_row(
-                    row, schema, source, line_number
-                )
-                if row_error:
-                    issues.append(
-                        InputIssue(
-                            source,
-                            f"line {line_number}: {row_error}",
-                            group,
-                        )
-                    )
-                else:
-                    records.append(record)
-    except (OSError, UnicodeError, csv.Error) as exc:
-        return [], [InputIssue(source, f"unable to read CSV: {exc}", group)]
-
-    if not records and not issues:
-        issues.append(InputIssue(source, "CSV contains no benchmark rows", group))
-    if issues:
-        return [], issues
-    return records, []
-
-
-def load_inputs(paths, group="input"):
-    result = LoadResult(group=group, paths=list(paths))
-    for path in paths:
-        records, issues = load_csv(path, group)
-        result.records.extend(records)
-        result.issues.extend(issues)
-    return result
-
-
-def _case_columns(record):
-    columns = [
-        key
-        for key in record
-        if key not in RESERVED_COLUMNS
-        and key not in INTERNAL_COLUMNS
-        and not key.startswith("_")
-    ]
-    order = {name: index for index, name in enumerate(CASE_COLUMN_ORDER)}
-    return sorted(columns, key=lambda name: (order.get(name, len(order)), name))
-
-
-def describe_case(record):
-    parts = [
-        f"{key}={record[key]}"
-        for key in _case_columns(record)
-        if str(record.get(key, "")).strip()
-    ]
-    return ", ".join(parts) if parts else "n/a"
-
-
-def case_identity(record):
-    values = tuple(
-        (key, str(record.get(key, "")).strip())
-        for key in _case_columns(record)
-    )
-    return record["api"], values
-
-
-def group_by_api(records):
-    grouped = defaultdict(list)
-    for record in records:
-        grouped[record["api"]].append(record)
-    return dict(grouped)
-
-
-def best_record(records, column):
-    candidates = [
-        record for record in records if record["status"] in SUCCESS_STATUSES
-    ]
-    if not candidates:
-        return None
-    key = lambda record: record[column]
-    return min(candidates, key=key) if column == "time_ms" else max(
-        candidates, key=key
-    )
-
-
-def _escape_markdown_fragment(value):
-    text = str(value)
-    for old, new in (
-        ("\\", "\\\\"),
-        ("|", "\\|"),
-        ("`", "\\`"),
-        ("*", "\\*"),
-        ("_", "\\_"),
-        ("#", "\\#"),
-        ("[", "\\["),
-        ("]", "\\]"),
-        ("<", "&lt;"),
-        (">", "&gt;"),
-    ):
-        text = text.replace(old, new)
-    return text
-
-
-def escape_markdown(value):
-    normalized = str(value).replace("\r\n", "\n").replace("\r", "\n")
-    return "<br>".join(
-        _escape_markdown_fragment(part) for part in normalized.split("\n")
-    )
-
-
-def format_metric(value):
-    return f"{value:.6g}"
-
-
-def format_delta(value):
-    if math.isinf(value):
-        return "+∞%" if value > 0 else "-∞%"
-    return f"{value:+.2f}%"
-
-
-def _status_text(records):
-    statuses = Counter(record["status"] for record in records)
-    return ", ".join(
-        f"{key}={value}" for key, value in sorted(statuses.items())
-    )
-
-
-def _input_file_lines(result):
-    records_by_source = defaultdict(list)
-    issues_by_source = defaultdict(list)
-    for record in result.records:
-        records_by_source[record["_source"]].append(record)
-    for issue in result.issues:
-        issues_by_source[issue.source].append(issue)
-
-    lines = ["## Input Files", "", "| File | Rows | Status |", "|---|---:|---|"]
-    for path in result.paths:
-        source = str(path)
-        file_records = records_by_source[source]
-        if issues_by_source[source]:
-            status = "invalid"
-        else:
-            status = _status_text(file_records)
-        lines.append(
-            f"| {escape_markdown(source)} | {len(file_records)} | "
-            f"{escape_markdown(status)} |"
-        )
-    lines.append("")
-    return lines
-
-
-def _input_error_lines(issues):
-    if not issues:
-        return []
-    lines = ["## Input Errors", ""]
-    for issue in issues:
-        lines.append(
-            f"- **{escape_markdown(issue.group)}** — "
-            f"{escape_markdown(issue.source)}: "
-            f"{escape_markdown(issue.message)}"
-        )
-    lines.append("")
-    return lines
-
-
-def _failure_lines(records):
-    failures = [
-        record for record in records if record["status"] == STATUS_FAILED
-    ]
-    if not failures:
-        return []
-    lines = [
-        "## Failure Details",
-        "",
-        "| API | Case | Source | Error |",
-        "|---|---|---|---|",
-    ]
-    for record in failures:
-        error = f"{record.get('error_type', '')}: {record.get('error', '')}"
-        lines.append(
-            f"| {escape_markdown(record['api'])} | "
-            f"{escape_markdown(describe_case(record))} | "
-            f"{escape_markdown(record['_source'])} | "
-            f"{escape_markdown(error.strip(': '))} |"
-        )
-    lines.append("")
-    return lines
-
-
-def _grouped_failure_lines(grouped_records):
-    failures = [
-        (group, record)
-        for group, record in grouped_records
-        if record["status"] == STATUS_FAILED
-    ]
-    if not failures:
-        return []
-    lines = [
-        "## Failure Details",
-        "",
-        "| Run | API | Case | Source | Error |",
-        "|---|---|---|---|---|",
-    ]
-    for group, record in failures:
-        error = f"{record.get('error_type', '')}: {record.get('error', '')}"
-        lines.append(
-            f"| {escape_markdown(group)} | "
-            f"{escape_markdown(record['api'])} | "
-            f"{escape_markdown(describe_case(record))} | "
-            f"{escape_markdown(record['_source'])} | "
-            f"{escape_markdown(error.strip(': '))} |"
-        )
-    lines.append("")
-    return lines
-
-
-def build_summary_markdown(result):
-    lines = [
-        "# FlashInfer Benchmark Summary",
-        "",
-        f"- Input files: {len(result.paths)}",
-        f"- Valid files: {len(result.valid_sources)}",
-        f"- Total rows: {len(result.records)}",
-        "",
-    ]
-    lines.extend(_input_file_lines(result))
-
-    for api, api_records in sorted(group_by_api(result.records).items()):
-        lines.extend(
-            [
-                f"## API: {escape_markdown(api)}",
-                "",
-                f"- Rows: {len(api_records)}",
-                f"- Status: {escape_markdown(_status_text(api_records))}",
-            ]
-        )
-        for column in NUMERIC_COLUMNS:
-            record = best_record(api_records, column)
-            if record is None:
-                continue
-            label = "Minimum" if column == "time_ms" else "Maximum"
-            lines.append(
-                f"- {label} {column}: {format_metric(record[column])} "
-                f"({escape_markdown(describe_case(record))}; "
-                f"source={escape_markdown(record['_source'])})"
-            )
-        lines.append("")
-
-    lines.extend(_failure_lines(result.records))
-    lines.extend(_input_error_lines(result.issues))
-    if any(
-        record["status"] == STATUS_LEGACY_OK for record in result.records
-    ):
-        lines.extend(
-            [
-                "## Compatibility Note",
-                "",
-                "- `legacy_inferred_ok` means the legacy CSV had no explicit "
-                "status column; success was inferred only after validating its "
-                "known FlashInfer schema and performance metrics.",
-                "",
-            ]
-        )
-    return "\n".join(lines).rstrip() + "\n"
-
-
-def build_case_index(records, group):
-    index = {}
-    duplicate_keys = set()
-    issues = []
-    for record in records:
-        key = case_identity(record)
-        if key in index:
-            first = index[key]
-            issues.append(
-                InputIssue(
-                    record["_source"],
-                    "duplicate case also found at "
-                    f"{first['_source']}:{first['_line']}: "
-                    f"{record['api']} ({describe_case(record)})",
-                    group,
-                )
-            )
-            duplicate_keys.add(key)
-        else:
-            index[key] = record
-    for key in duplicate_keys:
-        index.pop(key, None)
-    return index, issues
-
-
-def _percent_delta(baseline, candidate):
-    if baseline == 0:
-        return 0.0 if candidate == 0 else math.inf
-    return (candidate / baseline - 1.0) * 100.0
-
-
-def _is_threshold_regression(deltas, threshold):
-    if threshold is None:
-        return False
-    return (
-        deltas["time_ms"] > threshold
-        or deltas["bandwidth_GB_s"] < -threshold
-        or deltas["tflops"] < -threshold
-    )
-
-
-def build_regression_markdown(baseline, candidate, threshold=None):
-    baseline_index, baseline_duplicates = build_case_index(
-        baseline.records, "baseline"
-    )
-    candidate_index, candidate_duplicates = build_case_index(
-        candidate.records, "candidate"
-    )
-    comparison_issues = baseline_duplicates + candidate_duplicates
-
-    baseline_keys = set(baseline_index)
-    candidate_keys = set(candidate_index)
-    matched_keys = sorted(baseline_keys & candidate_keys)
-    missing_keys = sorted(baseline_keys - candidate_keys)
-    new_keys = sorted(candidate_keys - baseline_keys)
-
-    lines = [
-        "# FlashInfer Benchmark Regression",
-        "",
-        f"- Baseline files: {len(baseline.paths)}",
-        f"- Candidate files: {len(candidate.paths)}",
-        f"- Matched cases: {len(matched_keys)}",
-        f"- Missing candidate cases: {len(missing_keys)}",
-        f"- New candidate cases: {len(new_keys)}",
-        "- Regression gate: "
-        + ("report only" if threshold is None else f"{threshold:g}%"),
-        "",
-        "## Matched Cases",
-        "",
-        "| API | Case | State | Time ms (base → cand, Δ) | "
-        "Bandwidth GB/s (base → cand, Δ) | TFLOPS (base → cand, Δ) | Sources |",
-        "|---|---|---|---|---|---|---|",
-    ]
-
-    threshold_failed = False
-    candidate_failed = False
-    for key in matched_keys:
-        base_record = baseline_index[key]
-        candidate_record = candidate_index[key]
-        base_success = base_record["status"] in SUCCESS_STATUSES
-        candidate_success = candidate_record["status"] in SUCCESS_STATUSES
-        deltas = None
-
-        if base_success and candidate_success:
-            deltas = {
-                column: _percent_delta(
-                    base_record[column], candidate_record[column]
-                )
-                for column in NUMERIC_COLUMNS
-            }
-            regressed = _is_threshold_regression(deltas, threshold)
-            state = "regression" if regressed else "compared"
-            threshold_failed = threshold_failed or regressed
-        elif base_success and not candidate_success:
-            state = "candidate failed"
-            candidate_failed = True
-        elif not base_success and candidate_success:
-            state = "recovered"
-        else:
-            state = "both failed"
-            candidate_failed = True
-
-        if deltas:
-            metric_cells = [
-                f"{format_metric(base_record[column])} → "
-                f"{format_metric(candidate_record[column])}, "
-                f"{format_delta(deltas[column])}"
-                for column in NUMERIC_COLUMNS
-            ]
-        else:
-            metric_cells = ["n/a", "n/a", "n/a"]
-        sources = f"{base_record['_source']} → {candidate_record['_source']}"
-        lines.append(
-            f"| {escape_markdown(key[0])} | "
-            f"{escape_markdown(describe_case(base_record))} | "
-            f"{escape_markdown(state)} | "
-            f"{escape_markdown(metric_cells[0])} | "
-            f"{escape_markdown(metric_cells[1])} | "
-            f"{escape_markdown(metric_cells[2])} | "
-            f"{escape_markdown(sources)} |"
-        )
-    lines.append("")
-
-    if missing_keys:
-        lines.extend(["## Missing Candidate Cases", ""])
-        for key in missing_keys:
-            record = baseline_index[key]
-            lines.append(
-                f"- {escape_markdown(record['api'])}: "
-                f"{escape_markdown(describe_case(record))} "
-                f"(baseline source={escape_markdown(record['_source'])})"
-            )
-        lines.append("")
-
-    if new_keys:
-        lines.extend(["## New Candidate Cases", ""])
-        for key in new_keys:
-            record = candidate_index[key]
-            lines.append(
-                f"- {escape_markdown(record['api'])}: "
-                f"{escape_markdown(describe_case(record))} "
-                f"(candidate source={escape_markdown(record['_source'])})"
-            )
-            if record["status"] == STATUS_FAILED:
-                candidate_failed = True
-        lines.append("")
-
-    grouped_failures = [
-        ("baseline", record)
-        for record in baseline.records
-        if record["status"] == STATUS_FAILED
-    ] + [
-        ("candidate", record)
-        for record in candidate.records
-        if record["status"] == STATUS_FAILED
-    ]
-    if grouped_failures:
-        lines.extend(_grouped_failure_lines(grouped_failures))
-
-    all_issues = baseline.issues + candidate.issues + comparison_issues
-    lines.extend(_input_error_lines(all_issues))
-
-    comparison_failed = (
-        candidate_failed or bool(missing_keys) or threshold_failed
-    )
-    return (
-        "\n".join(lines).rstrip() + "\n",
-        comparison_failed,
-        comparison_issues,
-    )
-
-
-def parse_args(argv=None):
-    parser = argparse.ArgumentParser(
-        prog="summarize_results.py", description=__doc__
-    )
-    parser.add_argument(
-        "csv",
-        nargs="*",
-        type=Path,
-        help="Benchmark CSV files for summary mode.",
-    )
-    parser.add_argument(
-        "--baseline",
-        nargs="+",
-        type=Path,
-        help="Baseline CSV files for regression mode.",
-    )
-    parser.add_argument(
-        "--candidate",
-        nargs="+",
-        type=Path,
-        help="Candidate CSV files for regression mode.",
-    )
-    parser.add_argument(
-        "--fail-on-regression",
-        type=float,
-        metavar="PERCENT",
-        help="Return 1 when a metric regresses by more than this percentage.",
-    )
-    parser.add_argument(
-        "--output",
-        type=Path,
-        default=Path("flashinfer_benchmark_summary.md"),
-        help="Markdown report output path.",
-    )
-    args = parser.parse_args(argv)
-
-    regression_mode = args.baseline is not None or args.candidate is not None
-    if regression_mode:
-        if args.csv:
-            parser.error("positional CSV files cannot be used with regression mode")
-        if args.baseline is None or args.candidate is None:
-            parser.error("regression mode requires both --baseline and --candidate")
-    elif not args.csv:
-        parser.error("provide CSV files or use --baseline with --candidate")
-
-    if args.fail_on_regression is not None:
-        if not regression_mode:
-            parser.error("--fail-on-regression requires regression mode")
-        if not math.isfinite(args.fail_on_regression):
-            parser.error("--fail-on-regression must be finite")
-        if args.fail_on_regression < 0:
-            parser.error("--fail-on-regression must not be negative")
-    return args
-
-
-def _write_report(path, markdown):
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(markdown, encoding="utf-8")
-    print(f"Summary saved to {path}")
-
-
-def main(argv=None):
-    args = parse_args(argv)
-    if args.baseline is not None:
-        baseline = load_inputs(args.baseline, "baseline")
-        candidate = load_inputs(args.candidate, "candidate")
-        markdown, comparison_failed, comparison_issues = (
-            build_regression_markdown(
-                baseline, candidate, args.fail_on_regression
-            )
-        )
-        _write_report(args.output, markdown)
-        if baseline.issues or candidate.issues or comparison_issues:
-            return EXIT_INPUT_ERROR
-        return EXIT_BENCHMARK_FAILURE if comparison_failed else EXIT_OK
-
-    result = load_inputs(args.csv)
-    markdown = build_summary_markdown(result)
-    _write_report(args.output, markdown)
-    if result.issues:
-        return EXIT_INPUT_ERROR
-    if any(record["status"] == STATUS_FAILED for record in result.records):
-        return EXIT_BENCHMARK_FAILURE
-    return EXIT_OK
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

基于AI Agent开发范式的国产GPU大模型推理算子库优化/operator_task_package/flashinfer_task_package/benchmark/test_summarize_results.py

@@ -1,372 +0,0 @@
-import csv
-import io
-import math
-import tempfile
-import unittest
-from contextlib import redirect_stderr, redirect_stdout
-from pathlib import Path
-
-from benchmark_result import (
-    SCHEMA_VERSION,
-    STATUS_FAILED,
-    STATUS_OK,
-    execute_benchmark_case,
-)
-import summarize_results
-
-
-V1_FIELDS = [
-    "schema_version",
-    "api",
-    "batch_size",
-    "seq_len",
-    "num_qo_heads",
-    "num_kv_heads",
-    "head_dim",
-    "status",
-    "time_ms",
-    "bandwidth_GB_s",
-    "tflops",
-    "error_type",
-    "error",
-]
-LEGACY_FIELDS = [
-    "api",
-    "batch_size",
-    "seq_len",
-    "num_qo_heads",
-    "num_kv_heads",
-    "head_dim",
-    "time_ms",
-    "bandwidth_GB_s",
-    "tflops",
-]
-
-
-def success_row(**overrides):
-    row = {
-        "schema_version": SCHEMA_VERSION,
-        "api": "PagedWrapper",
-        "batch_size": "1",
-        "seq_len": "1024",
-        "num_qo_heads": "32",
-        "num_kv_heads": "4",
-        "head_dim": "128",
-        "status": STATUS_OK,
-        "time_ms": "2",
-        "bandwidth_GB_s": "10",
-        "tflops": "20",
-        "error_type": "",
-        "error": "",
-    }
-    row.update({key: str(value) for key, value in overrides.items()})
-    return row
-
-
-def failure_row(**overrides):
-    row = success_row(
-        status=STATUS_FAILED,
-        time_ms="",
-        bandwidth_GB_s="",
-        tflops="",
-        error_type="RuntimeError",
-        error="benchmark failed",
-    )
-    row.update({key: str(value) for key, value in overrides.items()})
-    return row
-
-
-class BenchmarkResultTests(unittest.TestCase):
-    def test_successful_case_computes_metrics(self):
-        record = execute_benchmark_case(
-            "Demo", {"batch_size": 1}, lambda: (2.0, 20_000_000, 4_000_000_000)
-        )
-
-        self.assertEqual(record["status"], STATUS_OK)
-        self.assertEqual(record["time_ms"], 2.0)
-        self.assertEqual(record["bandwidth_GB_s"], 10.0)
-        self.assertEqual(record["tflops"], 2.0)
-
-    def test_exception_and_invalid_metrics_become_failures(self):
-        def raise_error():
-            raise RuntimeError("broken")
-
-        for callback, expected_error in (
-            (raise_error, "RuntimeError"),
-            (lambda: (0.0, 1, 1), "ValueError"),
-            (lambda: (math.nan, 1, 1), "ValueError"),
-            (lambda: (1.0, math.inf, 1), "ValueError"),
-        ):
-            with self.subTest(expected_error=expected_error):
-                record = execute_benchmark_case(
-                    "Demo", {"batch_size": 1}, callback
-                )
-                self.assertEqual(record["status"], STATUS_FAILED)
-                self.assertEqual(record["error_type"], expected_error)
-
-    def test_keyboard_interrupt_is_not_swallowed(self):
-        def interrupt():
-            raise KeyboardInterrupt()
-
-        with self.assertRaises(KeyboardInterrupt):
-            execute_benchmark_case("Demo", {"batch_size": 1}, interrupt)
-
-
-class SummarizeResultsTests(unittest.TestCase):
-    def setUp(self):
-        self.temporary_directory = tempfile.TemporaryDirectory()
-        self.root = Path(self.temporary_directory.name)
-
-    def tearDown(self):
-        self.temporary_directory.cleanup()
-
-    def write_csv(self, name, fieldnames, rows, encoding="utf-8"):
-        path = self.root / name
-        with path.open("w", encoding=encoding, newline="") as csv_file:
-            writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
-            writer.writeheader()
-            writer.writerows(rows)
-        return path
-
-    def run_summary(self, paths, output_name="summary.md"):
-        output = self.root / output_name
-        argv = [*(str(path) for path in paths), "--output", str(output)]
-        with redirect_stdout(io.StringIO()):
-            exit_code = summarize_results.main(argv)
-        return exit_code, output.read_text(encoding="utf-8")
-
-    def run_regression(
-        self, baseline, candidate, threshold=None, output_name="regression.md"
-    ):
-        output = self.root / output_name
-        argv = [
-            "--baseline",
-            *(str(path) for path in baseline),
-            "--candidate",
-            *(str(path) for path in candidate),
-        ]
-        if threshold is not None:
-            argv.extend(("--fail-on-regression", str(threshold)))
-        argv.extend(("--output", str(output)))
-        with redirect_stdout(io.StringIO()):
-            exit_code = summarize_results.main(argv)
-        return exit_code, output.read_text(encoding="utf-8")
-
-    def test_failed_metrics_are_not_selected_and_markdown_is_escaped(self):
-        path = self.write_csv(
-            "mixed.csv",
-            V1_FIELDS,
-            [
-                success_row(),
-                failure_row(
-                    batch_size=2,
-                    time_ms=0,
-                    bandwidth_GB_s=9999,
-                    tflops=9999,
-                    error="bad | value\n## forged section",
-                ),
-            ],
-        )
-
-        exit_code, report = self.run_summary([path])
-
-        self.assertEqual(exit_code, summarize_results.EXIT_BENCHMARK_FAILURE)
-        self.assertIn("Minimum time_ms: 2", report)
-        self.assertNotIn("Minimum time_ms: 0", report)
-        self.assertIn("bad \\| value<br>\\#\\# forged section", report)
-        self.assertIn(summarize_results.escape_markdown(path), report)
-
-    def test_all_failures_are_included(self):
-        rows = [failure_row(batch_size=index, error=f"error-{index}") for index in range(1, 7)]
-        path = self.write_csv("failures.csv", V1_FIELDS, rows)
-
-        exit_code, report = self.run_summary([path])
-
-        self.assertEqual(exit_code, summarize_results.EXIT_BENCHMARK_FAILURE)
-        for index in range(1, 7):
-            self.assertIn(f"error-{index}", report)
-        self.assertNotIn("omitted", report)
-
-    def test_legacy_bom_csv_is_accepted_and_labeled(self):
-        legacy_row = {
-            key: value
-            for key, value in success_row().items()
-            if key in LEGACY_FIELDS
-        }
-        path = self.write_csv(
-            "legacy.csv", LEGACY_FIELDS, [legacy_row], encoding="utf-8-sig"
-        )
-
-        exit_code, report = self.run_summary([path])
-
-        self.assertEqual(exit_code, summarize_results.EXIT_OK)
-        self.assertIn("legacy\\_inferred\\_ok=1", report)
-        self.assertIn("Compatibility Note", report)
-
-    def test_invalid_numeric_empty_and_unrelated_csvs_return_input_error(self):
-        invalid_numeric = self.write_csv(
-            "nan.csv", V1_FIELDS, [success_row(time_ms="NaN")]
-        )
-        empty = self.write_csv("empty.csv", V1_FIELDS, [])
-        unrelated = self.write_csv(
-            "unrelated.csv", ["owner", "note"], [{"owner": "alice", "note": "x"}]
-        )
-
-        exit_code, report = self.run_summary(
-            [invalid_numeric, empty, unrelated]
-        )
-
-        self.assertEqual(exit_code, summarize_results.EXIT_INPUT_ERROR)
-        self.assertIn("Input files: 3", report)
-        self.assertIn("Valid files: 0", report)
-        self.assertIn("must be a finite number", report)
-        self.assertIn("contains no benchmark rows", report)
-        self.assertIn("known legacy FlashInfer schema", report)
-
-    def test_missing_duplicate_headers_and_unknown_schema_are_rejected(self):
-        missing = self.write_csv(
-            "missing.csv",
-            [field for field in V1_FIELDS if field != "error"],
-            [],
-        )
-        duplicate = self.root / "duplicate.csv"
-        duplicate.write_text(
-            "api,api,time_ms,bandwidth_GB_s,tflops\nA,A,1,2,3\n",
-            encoding="utf-8",
-        )
-        unknown = self.write_csv(
-            "unknown.csv", V1_FIELDS, [success_row(schema_version=2)]
-        )
-
-        exit_code, report = self.run_summary([missing, duplicate, unknown])
-
-        self.assertEqual(exit_code, summarize_results.EXIT_INPUT_ERROR)
-        self.assertIn("missing columns: error", report)
-        self.assertIn("duplicate header names", report)
-        self.assertIn("unsupported schema\\_version", report)
-
-    def test_regression_is_report_only_by_default_and_gate_is_optional(self):
-        baseline = self.write_csv(
-            "baseline.csv", V1_FIELDS, [success_row(time_ms=10, bandwidth_GB_s=100, tflops=50)]
-        )
-        candidate = self.write_csv(
-            "candidate.csv", V1_FIELDS, [success_row(time_ms=11, bandwidth_GB_s=95, tflops=45)]
-        )
-
-        exit_code, report = self.run_regression([baseline], [candidate])
-        gated_code, gated_report = self.run_regression(
-            [baseline], [candidate], threshold=5, output_name="gated.md"
-        )
-
-        self.assertEqual(exit_code, summarize_results.EXIT_OK)
-        self.assertIn("report only", report)
-        self.assertIn("+10.00%", report)
-        self.assertEqual(gated_code, summarize_results.EXIT_BENCHMARK_FAILURE)
-        self.assertIn("regression", gated_report)
-
-    def test_zero_baseline_throughput_is_reported_without_division_error(self):
-        baseline = self.write_csv(
-            "baseline.csv",
-            V1_FIELDS,
-            [success_row(bandwidth_GB_s=0, tflops=0)],
-        )
-        candidate = self.write_csv(
-            "candidate.csv",
-            V1_FIELDS,
-            [success_row(bandwidth_GB_s=10, tflops=20)],
-        )
-
-        exit_code, report = self.run_regression(
-            [baseline], [candidate], threshold=5
-        )
-
-        self.assertEqual(exit_code, summarize_results.EXIT_OK)
-        self.assertIn("+∞%", report)
-
-    def test_regression_status_transitions_and_case_sets(self):
-        baseline = self.write_csv(
-            "baseline.csv",
-            V1_FIELDS,
-            [
-                success_row(batch_size=1),
-                failure_row(batch_size=2),
-                success_row(batch_size=3),
-            ],
-        )
-        candidate = self.write_csv(
-            "candidate.csv",
-            V1_FIELDS,
-            [
-                failure_row(batch_size=1),
-                success_row(batch_size=2),
-                success_row(batch_size=4),
-            ],
-        )
-
-        exit_code, report = self.run_regression([baseline], [candidate])
-
-        self.assertEqual(exit_code, summarize_results.EXIT_BENCHMARK_FAILURE)
-        self.assertIn("candidate failed", report)
-        self.assertIn("recovered", report)
-        self.assertIn("Missing Candidate Cases", report)
-        self.assertIn("New Candidate Cases", report)
-
-    def test_both_failed_is_reported_as_candidate_failure(self):
-        baseline = self.write_csv(
-            "baseline.csv", V1_FIELDS, [failure_row(error="old failure")]
-        )
-        candidate = self.write_csv(
-            "candidate.csv", V1_FIELDS, [failure_row(error="new failure")]
-        )
-
-        exit_code, report = self.run_regression([baseline], [candidate])
-
-        self.assertEqual(exit_code, summarize_results.EXIT_BENCHMARK_FAILURE)
-        self.assertIn("both failed", report)
-        self.assertIn("old failure", report)
-        self.assertIn("new failure", report)
-
-    def test_recovery_and_new_successful_case_do_not_fail_comparison(self):
-        baseline = self.write_csv(
-            "baseline.csv", V1_FIELDS, [failure_row(batch_size=1)]
-        )
-        candidate = self.write_csv(
-            "candidate.csv",
-            V1_FIELDS,
-            [success_row(batch_size=1), success_row(batch_size=2)],
-        )
-
-        exit_code, report = self.run_regression([baseline], [candidate])
-
-        self.assertEqual(exit_code, summarize_results.EXIT_OK)
-        self.assertIn("recovered", report)
-        self.assertIn("New Candidate Cases", report)
-
-    def test_duplicate_case_in_a_run_is_an_input_error(self):
-        baseline = self.write_csv(
-            "baseline.csv", V1_FIELDS, [success_row(), success_row()]
-        )
-        candidate = self.write_csv(
-            "candidate.csv", V1_FIELDS, [success_row()]
-        )
-
-        exit_code, report = self.run_regression([baseline], [candidate])
-
-        self.assertEqual(exit_code, summarize_results.EXIT_INPUT_ERROR)
-        self.assertIn("duplicate case", report)
-
-    def test_invalid_cli_combinations_return_exit_two(self):
-        with redirect_stderr(io.StringIO()):
-            with self.assertRaises(SystemExit) as missing_candidate:
-                summarize_results.parse_args(["--baseline", "baseline.csv"])
-            with self.assertRaises(SystemExit) as summary_gate:
-                summarize_results.parse_args(
-                    ["input.csv", "--fail-on-regression", "5"]
-                )
-
-        self.assertEqual(missing_candidate.exception.code, 2)
-        self.assertEqual(summary_gate.exception.code, 2)
-
-
-if __name__ == "__main__":
-    unittest.main()