Skills › Scientific Agent Skills › peer-review › validate_claim_evidence.py
validate_claim_evidence.py
221 lines · skills/peer-review/scripts/validate_claim_evidence.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate a bounded claim-evidence alignment matrix without echoing prose."""
4from __future__ import annotations
6import argparse
7from collections import Counter
8from typing import Any
10from _common import (
11 ValidationError,
12 error_exit,
13 issue,
14 read_csv_records,
15 require_enum,
16 require_identifier,
17 require_text,
18 require_unique,
19 split_identifiers,
20 write_json_report,
21)
23FIELDS = (
24 "claim_id",
25 "location",
26 "claim_type",
27 "claim_summary",
28 "evidence_ids",
29 "support_level",
30 "alignment_issue",
31 "limitation",
32 "requested_action",
33)
34CLAIM_TYPES = {
35 "primary_outcome",
36 "secondary_outcome",
37 "causal",
38 "mechanistic",
39 "diagnostic",
40 "prediction",
41 "safety",
42 "generalization",
43 "methods",
44 "other",
45}
46SUPPORT_LEVELS = {
47 "supported",
48 "partly_supported",
49 "unsupported",
50 "not_assessed",
51}
52ALIGNMENT_ISSUES = {
53 "none",
54 "direction",
55 "magnitude",
56 "population",
57 "outcome",
58 "timepoint",
59 "causal_language",
60 "scope",
61 "uncertainty",
62 "selective_reporting",
63 "other",
64}
67def load_matrix(raw_path: str) -> list[dict[str, Any]]:
68 rows = read_csv_records(
69 raw_path,
70 required_fields=FIELDS,
71 allowed_fields=FIELDS,
72 )
73 parsed: list[dict[str, Any]] = []
74 claim_ids: list[str] = []
75 for line_number, row in enumerate(rows, start=2):
76 context = f"matrix row {line_number}"
77 claim_id = require_identifier(row["claim_id"], f"{context}.claim_id")
78 claim_ids.append(claim_id)
79 parsed.append(
80 {
81 "claim_id": claim_id,
82 "location": require_text(
83 row["location"], f"{context}.location", maximum=500
84 ),
85 "claim_type": require_enum(
86 row["claim_type"], CLAIM_TYPES, f"{context}.claim_type"
87 ),
88 "claim_summary": require_text(
89 row["claim_summary"],
90 f"{context}.claim_summary",
91 minimum=10,
92 maximum=2_000,
93 ),
94 "evidence_ids": split_identifiers(
95 row["evidence_ids"],
96 f"{context}.evidence_ids",
97 allow_empty=True,
98 ),
99 "support_level": require_enum(
100 row["support_level"],
101 SUPPORT_LEVELS,
102 f"{context}.support_level",
103 ),
104 "alignment_issue": require_enum(
105 row["alignment_issue"],
106 ALIGNMENT_ISSUES,
107 f"{context}.alignment_issue",
108 ),
109 "limitation": require_text(
110 row["limitation"],
111 f"{context}.limitation",
112 allow_empty=True,
113 maximum=2_000,
114 ),
115 "requested_action": require_text(
116 row["requested_action"],
117 f"{context}.requested_action",
118 allow_empty=True,
119 maximum=2_000,
120 ),
121 }
122 )
123 require_unique(claim_ids, "claim-evidence matrix")
124 return parsed
127def validate_matrix(rows: list[dict[str, Any]]) -> dict[str, Any]:
128 errors: list[dict[str, str]] = []
129 warnings: list[dict[str, str]] = []
130 support_counts = Counter()
131 issue_counts = Counter()
132 gap_ids: list[str] = []
133 action_missing_ids: list[str] = []
135 for row in rows:
136 claim_id = row["claim_id"]
137 support = row["support_level"]
138 alignment_issue = row["alignment_issue"]
139 evidence_ids = row["evidence_ids"]
140 support_counts[support] += 1
141 issue_counts[alignment_issue] += 1
143 if support == "supported":
144 if not evidence_ids:
145 errors.append(issue("SUPPORTED_CLAIM_HAS_NO_EVIDENCE", claim_id))
146 if alignment_issue != "none":
147 errors.append(issue("SUPPORTED_CLAIM_HAS_ALIGNMENT_ISSUE", claim_id))
148 elif support == "partly_supported":
149 gap_ids.append(claim_id)
150 if not evidence_ids:
151 errors.append(issue("PARTIAL_CLAIM_HAS_NO_EVIDENCE", claim_id))
152 if alignment_issue == "none":
153 errors.append(issue("PARTIAL_CLAIM_NEEDS_ISSUE_CODE", claim_id))
154 elif support in {"unsupported", "not_assessed"}:
155 gap_ids.append(claim_id)
156 if alignment_issue == "none" and support == "unsupported":
157 errors.append(issue("UNSUPPORTED_CLAIM_NEEDS_ISSUE_CODE", claim_id))
159 if support != "supported" and not row["requested_action"]:
160 action_missing_ids.append(claim_id)
161 warnings.append(issue("REQUESTED_ACTION_MISSING", claim_id))
162 if alignment_issue != "none" and not row["limitation"]:
163 warnings.append(issue("LIMITATION_CONTEXT_MISSING", claim_id))
164 if row["claim_type"] in {"causal", "mechanistic"} and support == "supported":
165 warnings.append(
166 issue("CAUSAL_OR_MECHANISTIC_SUPPORT_REQUIRES_EXPERT_REVIEW", claim_id)
167 )
169 return {
170 "schema_version": "2.0",
171 "valid": not errors,
172 "status": (
173 "INVALID_MATRIX"
174 if errors
175 else "VALID_WITH_ALIGNMENT_GAPS"
176 if gap_ids
177 else "VALID_NO_RECORDED_GAPS"
178 ),
179 "errors": errors,
180 "warnings": warnings,
181 "claim_count": len(rows),
182 "support_counts": dict(sorted(support_counts.items())),
183 "alignment_issue_counts": dict(sorted(issue_counts.items())),
184 "claim_ids_requiring_resolution": sorted(gap_ids),
185 "claim_ids_missing_requested_action": sorted(action_missing_ids),
186 "notice": (
187 "This report checks matrix structure and declared claim-evidence "
188 "alignment only. It does not verify evidence truth, reproduce analyses, "
189 "or determine manuscript merit. Reports contain identifiers, not claim "
190 "or manuscript text."
191 ),
192 }
195def build_parser() -> argparse.ArgumentParser:
196 parser = argparse.ArgumentParser(
197 description=(
198 "Validate a local claim-evidence CSV and emit only identifiers and "
199 "counts; raw claim text is never echoed."
200 )
201 )
202 parser.add_argument("matrix", help="Local claim-evidence matrix CSV")
203 parser.add_argument("-o", "--output", help="Optional local JSON report")
204 parser.add_argument(
205 "--force", action="store_true", help="Replace an existing output file"
206 )
207 return parser
210def main() -> int:
211 args = build_parser().parse_args()
212 try:
213 report = validate_matrix(load_matrix(args.matrix))
214 write_json_report(report, args.output, force=args.force)
215 return 0 if report["valid"] else 1
216 except ValidationError as exc:
217 return error_exit(exc)
220if __name__ == "__main__":
221 raise SystemExit(main())