Add negative-result archive validator

scripts/validate_negative_archive.py

@@ -0,0 +1,140 @@
+"""Validate negative-result archive CSV entries.
+
+Checks each row against the schema defined in
+docs/evidence/NEGATIVE_RESULT_ARCHIVE.md.
+
+Usage:
+    python scripts/validate_negative_archive.py \
+        --archive outputs/negative_result_archive.csv
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import sys
+from datetime import datetime
+from pathlib import Path
+
+REQUIRED = {"entry_id", "date", "candidate_id", "sequence", "reason_category",
+            "reason_detail", "pipeline_version", "source_batch"}
+
+VALID_CATEGORIES = {"pre_selection_reject", "selected_untested", "lab_inactive",
+                    "lab_toxic", "control_failure", "synthesis_failure"}
+
+CONDITIONAL_CATEGORIES = {"lab_inactive", "lab_toxic"}
+
+DATE_FORMAT = "%Y-%m-%d"
+
+
+def validate_archive(archive_path: str) -> dict:
+    p = Path(archive_path)
+    if not p.exists():
+        return {"error": f"Archive not found: {archive_path}"}
+
+    rows: list[dict] = []
+    with p.open("r", newline="") as f:
+        reader = csv.DictReader(f)
+        for row in reader:
+            rows.append(row)
+
+    if not rows:
+        return {"error": "Archive is empty"}
+
+    errors: list[dict] = []
+    warnings: list[dict] = []
+    valid_count = 0
+
+    for i, row in enumerate(rows):
+        row_errors = []
+        row_warnings = []
+
+        # Required fields
+        for field in REQUIRED:
+            val = row.get(field, "").strip()
+            if not val:
+                row_errors.append(f"Missing required field: {field}")
+
+        # Reason category
+        cat = row.get("reason_category", "").strip()
+        if cat and cat not in VALID_CATEGORIES:
+            row_errors.append(f"Invalid reason_category: {cat}")
+
+        # Date format
+        date_val = row.get("date", "").strip()
+        if date_val:
+            try:
+                datetime.strptime(date_val, DATE_FORMAT)
+            except ValueError:
+                row_errors.append(f"Invalid date format: {date_val} (expected YYYY-MM-DD)")
+
+        # Conditional fields for lab-tested categories
+        if cat in CONDITIONAL_CATEGORIES:
+            for cf in ["assay_type", "assay_result", "assay_unit"]:
+                if not row.get(cf, "").strip():
+                    row_warnings.append(f"Conditional field '{cf}' recommended for {cat}")
+
+        # Superseded_by should reference an existing entry_id
+        superseded = row.get("superseded_by", "").strip()
+        if superseded:
+            found = any(r.get("entry_id", "").strip() == superseded for r in rows)
+            if not found:
+                row_warnings.append(f"superseded_by={superseded} does not match any entry_id")
+
+        # pipeline_version should not be empty if present
+        pv = row.get("pipeline_version", "").strip()
+        if not pv and date_val:
+            row_warnings.append("pipeline_version is recommended")
+
+        if row_errors:
+            errors.append({
+                "row": i + 1,
+                "entry_id": row.get("entry_id", "?"),
+                "errors": row_errors,
+            })
+        else:
+            valid_count += 1
+
+        if row_warnings:
+            warnings.append({
+                "row": i + 1,
+                "entry_id": row.get("entry_id", "?"),
+                "warnings": row_warnings,
+            })
+
+    return {
+        "archive": archive_path,
+        "total_rows": len(rows),
+        "valid": valid_count,
+        "errors": len(errors),
+        "warnings": len(warnings),
+        "per_row_errors": errors,
+        "per_row_warnings": warnings,
+        "all_valid": len(errors) == 0,
+    }
+
+
+def main() -> int:
+    import argparse
+    parser = argparse.ArgumentParser(description="Validate negative-result archive")
+    parser.add_argument("--archive", required=True)
+    parser.add_argument("--out", default=None)
+    args = parser.parse_args()
+
+    result = validate_archive(args.archive)
+    if "error" in result:
+        print(f"Error: {result['error']}", file=sys.stderr)
+        return 2
+
+    print(json.dumps(result, indent=2))
+    print(f"\nValid: {result['valid']}/{result['total_rows']}, "
+          f"Errors: {result['errors']}, Warnings: {result['warnings']}")
+
+    if args.out:
+        Path(args.out).write_text(json.dumps(result, indent=2) + "\n")
+
+    return 0 if result["all_valid"] else 3
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_validate_negative_archive.py

@@ -0,0 +1,151 @@
+"""Tests for the negative-result archive validator."""
+
+import csv
+import json
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from scripts.validate_negative_archive import validate_archive
+
+
+def _make_archive(tmp: Path, rows: list[dict]) -> str:
+    p = tmp / "archive.csv"
+    with p.open("w", newline="") as f:
+        if rows:
+            w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
+            w.writeheader()
+            w.writerows(rows)
+    return str(p)
+
+
+_VALID_ROW = {
+    "entry_id": "1",
+    "date": "2026-08-01",
+    "candidate_id": "WAVE0.5-001",
+    "sequence": "ACDEFGHIKLMNPQRSTVWY",
+    "reason_category": "lab_inactive",
+    "reason_detail": "MIC > 64 ug/mL",
+    "pipeline_version": "v0.5.49",
+    "source_batch": "wave0.5",
+    "assay_type": "MIC",
+    "assay_result": "128",
+    "assay_unit": "ug/mL",
+    "score_activity": "0.85",
+    "score_safety": "0.79",
+    "score_novelty": "0.88",
+    "score_ensemble": "0.83",
+    "recalibration_used": "no",
+    "superseded_by": "",
+    "reviewer_notes": "",
+}
+
+
+def test_valid_entry():
+    with tempfile.TemporaryDirectory() as tmp:
+        result = validate_archive(_make_archive(Path(tmp), [_VALID_ROW]))
+        assert result["all_valid"] is True
+        assert result["valid"] == 1
+
+
+def test_missing_required_field():
+    row = dict(_VALID_ROW)
+    del row["candidate_id"]
+    with tempfile.TemporaryDirectory() as tmp:
+        result = validate_archive(_make_archive(Path(tmp), [row]))
+        assert result["all_valid"] is False
+        assert result["errors"] == 1
+
+
+def test_invalid_category():
+    row = dict(_VALID_ROW, reason_category="invalid_cat")
+    with tempfile.TemporaryDirectory() as tmp:
+        result = validate_archive(_make_archive(Path(tmp), [row]))
+        assert result["all_valid"] is False
+
+
+def test_invalid_date():
+    row = dict(_VALID_ROW, date="not-a-date")
+    with tempfile.TemporaryDirectory() as tmp:
+        result = validate_archive(_make_archive(Path(tmp), [row]))
+        assert result["all_valid"] is False
+
+
+def test_conditional_fields_warning():
+    row = dict(_VALID_ROW, assay_type="", assay_result="", assay_unit="")
+    with tempfile.TemporaryDirectory() as tmp:
+        result = validate_archive(_make_archive(Path(tmp), [row]))
+        assert result["warnings"] >= 1
+
+
+def test_superseded_refers_to_existing():
+    rows = [
+        dict(_VALID_ROW, entry_id="1", superseded_by=""),
+        dict(_VALID_ROW, entry_id="2", superseded_by="1"),
+    ]
+    with tempfile.TemporaryDirectory() as tmp:
+        result = validate_archive(_make_archive(Path(tmp), rows))
+        assert result["all_valid"] is True
+
+
+def test_superseded_refers_to_nonexistent():
+    rows = [
+        dict(_VALID_ROW, entry_id="1", superseded_by="999"),
+    ]
+    with tempfile.TemporaryDirectory() as tmp:
+        result = validate_archive(_make_archive(Path(tmp), rows))
+        assert result["warnings"] >= 1
+
+
+def test_empty_archive():
+    with tempfile.TemporaryDirectory() as tmp:
+        p = Path(tmp) / "empty.csv"
+        # CSV with only header
+        with p.open("w", newline="") as f:
+            w = csv.writer(f)
+            w.writerow(list(_VALID_ROW.keys()))
+        result = validate_archive(str(p))
+        assert "error" in result or result["total_rows"] == 0
+
+
+def test_missing_file():
+    result = validate_archive("/nonexistent.csv")
+    assert "error" in result
+
+
+def test_cli_exit_valid(tmp_path):
+    p = _make_archive(tmp_path, [_VALID_ROW])
+    r = subprocess.run(
+        [sys.executable, "scripts/validate_negative_archive.py",
+         "--archive", p],
+        capture_output=True, text=True,
+    )
+    assert r.returncode == 0
+
+
+def test_cli_exit_invalid(tmp_path):
+    row = dict(_VALID_ROW, reason_category="")
+    p = _make_archive(tmp_path, [row])
+    r = subprocess.run(
+        [sys.executable, "scripts/validate_negative_archive.py",
+         "--archive", p],
+        capture_output=True, text=True,
+    )
+    assert r.returncode == 3
+
+
+def test_cli_writes_output(tmp_path):
+    p = _make_archive(tmp_path, [_VALID_ROW])
+    out = tmp_path / "report.json"
+    subprocess.run(
+        [sys.executable, "scripts/validate_negative_archive.py",
+         "--archive", p, "--out", str(out)],
+        capture_output=True,
+        env={"PYTHONPATH": "src"},
+    )
+    assert out.exists()
+    data = json.loads(out.read_text())
+    assert "valid" in data