Fix report crash on headerless raw CSV

A leftover 0-byte session_<id>_raw.csv (e.g. from an interrupted or
restarted run) was appended to without ever receiving a header, since
_initialize_csv only wrote one when the file was missing. load_session
then read it with csv.DictReader, which treated the first data row as
the header and raised KeyError: 'timestamp', aborting report generation
even though all the data was valid.

- _initialize_csv: write the header when the file is missing OR empty.
- load_session: tolerate a headerless CSV by mapping rows positionally
  against CSV_HEADER (detected via the first cell), padding short rows.
- Add regression tests for both paths.

Author: JasonLeviGoodison

src/benchmarking/metrics/collector.py

@@ -37,7 +37,11 @@ def __init__(self, session_id: str, results_dir: str = "./results"):
         self._initialize_csv()
 
     def _initialize_csv(self) -> None:
-        if not self.csv_path.exists():
+        # Write the header when the file is missing OR exists but is empty.
+        # A leftover 0-byte file (e.g. from an interrupted/restarted run)
+        # would otherwise be appended to without ever getting a header,
+        # producing a headerless CSV that breaks report generation.
+        if not self.csv_path.exists() or self.csv_path.stat().st_size == 0:
             with open(self.csv_path, "w", newline="") as f:
                 csv.writer(f).writerow(CSV_HEADER)
 
@@ -88,37 +92,51 @@ def _f(v: str | None) -> float | None:
                 return None
             return float(v)
 
-        with open(csv_path, "r") as f:
-            for row in csv.DictReader(f):
-                timestamp = datetime.fromisoformat(row["timestamp"])
-                if start_time is None or timestamp < start_time:
-                    start_time = timestamp
-                if end_time is None or timestamp > end_time:
-                    end_time = timestamp
-
-                services.add(row["service_name"])
-                models.add(row.get("model_name", ""))
-                workloads.add(row["workload_type"])
-                iterations_per_workload = max(iterations_per_workload, int(row["iteration"]) + 1)
-
-                metrics.append(
-                    BenchmarkMetric(
-                        session_id=row["session_id"],
-                        service_name=row["service_name"],
-                        model_name=row.get("model_name", ""),
-                        workload_type=row["workload_type"],
-                        iteration=int(row["iteration"]),
-                        ttft_ms=_f(row.get("ttft_ms")),
-                        end_to_end_latency_ms=float(row["end_to_end_latency_ms"]),
-                        tokens_generated=int(row["tokens_generated"]),
-                        prompt_tokens=int(row["prompt_tokens"]),
-                        tokens_per_sec=_f(row.get("tokens_per_sec")),
-                        inter_token_latency_ms=_f(row.get("inter_token_latency_ms")),
-                        attempts=int(row.get("attempts") or 1),
-                        timestamp=timestamp,
-                        error=row["error"] if row["error"] else None,
-                    )
+        with open(csv_path, "r", newline="") as f:
+            data_rows = [r for r in csv.reader(f) if r]
+
+        # Tolerate raw CSVs written without a header row. A genuine header
+        # begins with the literal "session_id" column name; anything else is
+        # treated as data and mapped positionally against CSV_HEADER. This
+        # lets us still produce reports from files an interrupted/restarted
+        # run left headerless.
+        if data_rows and data_rows[0][0] == CSV_HEADER[0]:
+            data_rows = data_rows[1:]
+
+        for raw in data_rows:
+            # Pad short rows so every column is present (mirrors DictReader's
+            # restval behaviour) before zipping into a name->value mapping.
+            padded = list(raw) + [""] * (len(CSV_HEADER) - len(raw))
+            row = dict(zip(CSV_HEADER, padded))
+            timestamp = datetime.fromisoformat(row["timestamp"])
+            if start_time is None or timestamp < start_time:
+                start_time = timestamp
+            if end_time is None or timestamp > end_time:
+                end_time = timestamp
+
+            services.add(row["service_name"])
+            models.add(row.get("model_name", ""))
+            workloads.add(row["workload_type"])
+            iterations_per_workload = max(iterations_per_workload, int(row["iteration"]) + 1)
+
+            metrics.append(
+                BenchmarkMetric(
+                    session_id=row["session_id"],
+                    service_name=row["service_name"],
+                    model_name=row.get("model_name", ""),
+                    workload_type=row["workload_type"],
+                    iteration=int(row["iteration"]),
+                    ttft_ms=_f(row.get("ttft_ms")),
+                    end_to_end_latency_ms=float(row["end_to_end_latency_ms"]),
+                    tokens_generated=int(row["tokens_generated"]),
+                    prompt_tokens=int(row["prompt_tokens"]),
+                    tokens_per_sec=_f(row.get("tokens_per_sec")),
+                    inter_token_latency_ms=_f(row.get("inter_token_latency_ms")),
+                    attempts=int(row.get("attempts") or 1),
+                    timestamp=timestamp,
+                    error=row["error"] if row["error"] else None,
                 )
+            )
 
         return BenchmarkSession(
             session_id=session_id,

tests/test_metrics/test_collector.py

@@ -32,3 +32,32 @@ def test_metrics_collector_round_trips_result(tmp_path: Path) -> None:
     assert session.results[0].ttft_ms == 250.0
     assert session.results[0].end_to_end_latency_ms == 1500.0
     assert session.results[0].attempts == 2
+
+
+def test_load_session_tolerates_headerless_csv(tmp_path: Path) -> None:
+    """A raw CSV left headerless by an interrupted/restarted run must still
+    load (it previously crashed report generation with KeyError: 'timestamp')."""
+    csv_path = tmp_path / "session_legacy_raw.csv"
+    csv_path.write_text(
+        "legacy,Provider A,model-a,ctx_256,0,250.0,1500.0,100,25,80.0,12.5,1,"
+        "2026-06-09T18:15:05.451820+00:00,\n"
+    )
+
+    collector = MetricsCollector("legacy", str(tmp_path))
+    session = collector.load_session("legacy")
+
+    assert len(session.results) == 1
+    assert session.results[0].service_name == "Provider A"
+    assert session.results[0].tokens_generated == 100
+    assert session.results[0].error is None
+
+
+def test_initialize_csv_repairs_empty_file(tmp_path: Path) -> None:
+    """A pre-existing 0-byte session file must get a header written, otherwise
+    every subsequent append produces a headerless CSV."""
+    csv_path = tmp_path / "session_empty_raw.csv"
+    csv_path.write_text("")  # leftover 0-byte file
+
+    MetricsCollector("empty", str(tmp_path))
+
+    assert csv_path.read_text().startswith("session_id,")