feat: flashinfer benchmark 结果增加汇总脚本-2

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

@@ -366,7 +366,7 @@ pip install pandas
 
 ```bash
 # 检查脚本文件
-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 "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 "from bench_common import setup_workspace, get_csv_path; print('脚本导入正常')"
@@ -376,6 +376,8 @@ 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
@@ -414,6 +416,57 @@ 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`，表示成功状态来自格式推断而非原始记录。
+
 ***
 
 **查看结果命令示例：**
@@ -984,7 +1037,7 @@ FlashInfer 方向包含 **4 个可选算子题目**，均属于同一比赛通
 | --- | --- | --- |
 | Benchmark 运行时间过长 | 参数组合过多， workload 较大 | 耐心等待，脚本会自动调整重复次数 |
 | `KeyError: 'BatchPrefillWithPagedKVCacheKernel'` | profiler 未捕获目标 kernel | 检查 `target_kernels` 配置是否正确 |
-| CSV 文件为空 | 测试未正常完成 | 检查 GPU 显存是否充足，重新运行 |
+| CSV 文件为空或汇总器返回退出码 `2` | 测试未正常完成或 CSV schema 无效 | 查看汇总报告的 `Input Errors`，检查文件完整性后重新运行 |
 
 ### 8.4 代码问题
 
@@ -1450,4 +1503,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,7 +8,16 @@ import pandas as pd
 import torch
 
 import flashinfer
-from bench_common import dtype, page_block_size, setup_workspace, setup_paged_kv_indptr, run_with_profiler, get_csv_path, compute_reps
+from benchmark_result import execute_benchmark_case, has_failures, STATUS_OK
+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"]
 
@@ -69,22 +78,39 @@ 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
-        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,
+        case = {
             "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,
-            "bandwidth_GB_s": bw,
-            "tflops": tflops,
-        })
-        print(f"  [{idx}/{total_cases}] bs={bs}, kv_len={sl_kv}, hd={hd}: {ms:.3f}ms, {bw:.2f} GB/s, {tflops:.2f} TFLOPs")
+        }
+        record = execute_benchmark_case(
+            api_name,
+            case,
+            lambda: bench_batch_decode(
+                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
 
@@ -98,4 +124,6 @@ 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,7 +8,14 @@ import pandas as pd
 import torch
 
 import flashinfer
-from bench_common import dtype, page_block_size, setup_workspace, run_with_profiler, get_csv_path, compute_reps
+from benchmark_result import execute_benchmark_case, has_failures, STATUS_OK
+from bench_common import (
+    compute_reps,
+    dtype,
+    get_csv_path,
+    run_with_profiler,
+    setup_workspace,
+)
 
 target_kernels = ["BatchMLAPagedAttentionKernel"]
 
@@ -79,21 +86,34 @@ def run_benchmark():
 
     print(f"[{api_name}] Starting benchmark, total cases: {total_cases}")
     for idx, (num_heads, bs, sl) in enumerate(test_cases, 1):
-        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,
+        case = {
             "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,
-            "bandwidth_GB_s": bw,
-            "tflops": tflops,
-        })
-        print(f"  [{idx}/{total_cases}] bs={bs}, sl={sl}, num_heads={num_heads}: {ms:.3f}ms, {bw:.2f} GB/s, {tflops:.2f} TFLOPs")
+        }
+        record = execute_benchmark_case(
+            api_name,
+            case,
+            lambda: bench_batch_mla_paged_attention(
+                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
 
@@ -107,4 +127,6 @@ 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,6 +16,7 @@ from bench_common import (
     get_csv_path,
     compute_reps,
 )
+from benchmark_result import execute_benchmark_case, has_failures, STATUS_OK
 
 target_kernels = ["BatchPrefillWithPagedKVCacheKernel"]
 
@@ -95,27 +96,33 @@ 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
-        ms, io, flops = bench_batch_prefill_with_paged_kv_cache(
-            bs, sl, num_qo_heads, num_kv_heads, head_dim
-        )
-        bw = io / ms / 1e6
-        tflops = flops / ms / 1e9
-        records.append(
-            {
-                "api": api_name,
-                "batch_size": bs,
-                "seq_len": sl,
-                "num_qo_heads": num_qo_heads,
-                "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"
+        case = {
+            "batch_size": bs,
+            "seq_len": sl,
+            "num_qo_heads": num_qo_heads,
+            "num_kv_heads": num_kv_heads,
+            "head_dim": head_dim,
+        }
+        record = execute_benchmark_case(
+            api_name,
+            case,
+            lambda: bench_batch_prefill_with_paged_kv_cache(
+                bs, sl, num_qo_heads, num_kv_heads, head_dim
+            ),
         )
+        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
 
@@ -131,3 +138,5 @@ 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,6 +15,7 @@ from bench_common import (
     get_csv_path,
     compute_reps,
 )
+from benchmark_result import execute_benchmark_case, has_failures, STATUS_OK
 
 target_kernels = [
     "BatchPrefillWithRaggedKVCacheKernel",
@@ -97,28 +98,41 @@ 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
-        ms, io, flops = bench_batch_prefill_with_ragged_kv_cache(
-            bs, sl, num_qo_heads, num_kv_heads, head_dim_qk, head_dim_vo
-        )
-        bw = io / ms / 1e6
-        tflops = flops / ms / 1e9
-        records.append(
-            {
-                "api": api_name,
-                "batch_size": bs,
-                "seq_len": sl,
-                "num_qo_heads": num_qo_heads,
-                "num_kv_heads": num_kv_heads,
-                "head_dim_qk": head_dim_qk,
-                "head_dim_vo": head_dim_vo,
-                "time_ms": ms,
-                "bandwidth_GB_s": bw,
-                "tflops": tflops,
-            }
-        )
-        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"
+        case = {
+            "batch_size": bs,
+            "seq_len": sl,
+            "num_qo_heads": num_qo_heads,
+            "num_kv_heads": num_kv_heads,
+            "head_dim_qk": head_dim_qk,
+            "head_dim_vo": head_dim_vo,
+        }
+        record = execute_benchmark_case(
+            api_name,
+            case,
+            lambda: bench_batch_prefill_with_ragged_kv_cache(
+                bs,
+                sl,
+                num_qo_heads,
+                num_kv_heads,
+                head_dim_qk,
+                head_dim_vo,
+            ),
         )
+        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
 
@@ -134,3 +148,5 @@ 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

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

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

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