Skills › Scientific Agent Skills › scholar-evaluation › check_traceability.py
check_traceability.py
233 lines · skills/scholar-evaluation/scripts/check_traceability.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check criterion-to-evidence traceability without copying source content."""
4from __future__ import annotations
6import argparse
7from pathlib import Path
8from typing import Any
10import _common
12SOURCE_TYPES = (
13 "section",
14 "table",
15 "figure",
16 "dataset",
17 "code",
18 "protocol",
19 "registration",
20 "other_local_record",
21)
22ACCESS_STATUSES = ("available", "restricted_authorized", "unavailable")
23VERIFICATION_STATUSES = ("verified", "unverified")
26def validate_manifest(
27 manifest: Any, rubric: dict, evaluation: dict
28) -> tuple[list[_common.Issue], dict[str, dict]]:
29 issues: list[_common.Issue] = []
30 evidence_by_id: dict[str, dict] = {}
31 if not _common.exact_keys(
32 manifest,
33 {
34 "schema_version",
35 "evaluation_id",
36 "work_id",
37 "data_classification",
38 "evidence",
39 },
40 "$",
41 issues,
42 ):
43 return issues, evidence_by_id
44 if manifest.get("schema_version") != _common.SCHEMA_VERSION:
45 issues.append(_common.Issue("SCHEMA_VERSION_UNSUPPORTED", "$.schema_version"))
46 if manifest.get("evaluation_id") != evaluation.get("evaluation_id"):
47 issues.append(_common.Issue("EVALUATION_ID_MISMATCH", "$.evaluation_id"))
48 if manifest.get("work_id") != evaluation.get("work_id"):
49 issues.append(_common.Issue("WORK_ID_MISMATCH", "$.work_id"))
50 if manifest.get("data_classification") != evaluation.get("data_classification"):
51 issues.append(
52 _common.Issue("DATA_CLASSIFICATION_MISMATCH", "$.data_classification")
53 )
54 if manifest.get("data_classification") not in _common.ALLOWED_CLASSIFICATIONS:
55 issues.append(
56 _common.Issue(
57 "DATA_CLASSIFICATION_NOT_ALLOWED", "$.data_classification"
58 )
59 )
60 criterion_ids = {criterion["criterion_id"] for criterion in rubric["criteria"]}
61 evidence = manifest.get("evidence")
62 if not isinstance(evidence, list) or len(evidence) > 5_000:
63 issues.append(_common.Issue("EVIDENCE_LIST_INVALID", "$.evidence"))
64 return issues, evidence_by_id
65 for index, item in enumerate(evidence):
66 path = f"$.evidence[{index}]"
67 if not _common.exact_keys(
68 item,
69 {
70 "evidence_id",
71 "criterion_ids",
72 "source_type",
73 "locator_ref",
74 "claim_ref",
75 "access_status",
76 "verification_status",
77 },
78 path,
79 issues,
80 ):
81 continue
82 evidence_id = item.get("evidence_id")
83 _common.validate_identifier(
84 evidence_id, f"{path}.evidence_id", issues, "EVIDENCE_ID_INVALID"
85 )
86 if isinstance(evidence_id, str):
87 if evidence_id in evidence_by_id:
88 issues.append(
89 _common.Issue("EVIDENCE_ID_DUPLICATE", f"{path}.evidence_id")
90 )
91 evidence_by_id[evidence_id] = item
92 linked = item.get("criterion_ids")
93 if not isinstance(linked, list) or not 1 <= len(linked) <= 50:
94 issues.append(_common.Issue("CRITERION_IDS_INVALID", f"{path}.criterion_ids"))
95 else:
96 linked_text = [
97 criterion_id
98 for criterion_id in linked
99 if isinstance(criterion_id, str)
100 ]
101 if len(linked_text) != len(set(linked_text)):
102 issues.append(
103 _common.Issue(
104 "CRITERION_ID_DUPLICATE", f"{path}.criterion_ids"
105 )
106 )
107 for criterion_index, criterion_id in enumerate(linked):
108 _common.validate_identifier(
109 criterion_id,
110 f"{path}.criterion_ids[{criterion_index}]",
111 issues,
112 )
113 if isinstance(criterion_id, str) and criterion_id not in criterion_ids:
114 issues.append(
115 _common.Issue(
116 "CRITERION_UNKNOWN",
117 f"{path}.criterion_ids[{criterion_index}]",
118 )
119 )
120 if item.get("source_type") not in SOURCE_TYPES:
121 issues.append(_common.Issue("SOURCE_TYPE_INVALID", f"{path}.source_type"))
122 _common.validate_reference(item.get("locator_ref"), f"{path}.locator_ref", issues)
123 _common.validate_reference(item.get("claim_ref"), f"{path}.claim_ref", issues)
124 if item.get("access_status") not in ACCESS_STATUSES:
125 issues.append(_common.Issue("ACCESS_STATUS_INVALID", f"{path}.access_status"))
126 if item.get("access_status") == "unavailable":
127 issues.append(_common.Issue("EVIDENCE_UNAVAILABLE", f"{path}.access_status"))
128 if item.get("verification_status") not in VERIFICATION_STATUSES:
129 issues.append(
130 _common.Issue(
131 "VERIFICATION_STATUS_INVALID", f"{path}.verification_status"
132 )
133 )
134 elif item.get("verification_status") != "verified":
135 issues.append(
136 _common.Issue("EVIDENCE_UNVERIFIED", f"{path}.verification_status")
137 )
138 return issues, evidence_by_id
141def check_traceability(rubric: dict, evaluation: dict, manifest: dict) -> dict:
142 rubric_issues = _common.validate_rubric(rubric)
143 _common.require_valid(rubric_issues)
144 evaluation_issues = _common.validate_evaluation(evaluation, rubric)
145 _common.require_valid(evaluation_issues)
146 issues, evidence_by_id = validate_manifest(manifest, rubric, evaluation)
147 used_ids: set[str] = set()
148 for index, rating in enumerate(evaluation["ratings"]):
149 if rating["status"] != "rated":
150 continue
151 criterion_id = rating["criterion_id"]
152 for evidence_index, evidence_id in enumerate(rating["evidence_ids"]):
153 used_ids.add(evidence_id)
154 path = f"$.ratings[{index}].evidence_ids[{evidence_index}]"
155 evidence = evidence_by_id.get(evidence_id)
156 if evidence is None:
157 issues.append(_common.Issue("EVIDENCE_REFERENCE_UNRESOLVED", path))
158 elif not isinstance(evidence.get("criterion_ids"), list) or criterion_id not in evidence["criterion_ids"]:
159 issues.append(_common.Issue("EVIDENCE_CRITERION_MISMATCH", path))
160 for evidence_id in sorted(set(evidence_by_id) - used_ids):
161 issues.append(
162 _common.Issue(
163 "EVIDENCE_UNUSED",
164 f"$.evidence.{evidence_id}",
165 "warning",
166 )
167 )
168 errors = _common.error_issues(issues)
169 rated_count = sum(
170 rating["status"] == "rated" for rating in evaluation["ratings"]
171 )
172 resolved_count = sum(
173 1
174 for rating in evaluation["ratings"]
175 if rating["status"] == "rated"
176 and rating["evidence_ids"]
177 and all(evidence_id in evidence_by_id for evidence_id in rating["evidence_ids"])
178 )
179 return {
180 "schema_version": _common.SCHEMA_VERSION,
181 "report_type": "evidence_traceability_check",
182 "notice": _common.NOTICE,
183 "status": "fail" if errors else "pass",
184 "evaluation_id": evaluation["evaluation_id"],
185 "work_id": evaluation["work_id"],
186 "rated_criteria": rated_count,
187 "rated_criteria_with_resolved_evidence": resolved_count,
188 "evidence_records": len(evidence_by_id),
189 "error_count": len(errors),
190 "warning_count": len(issues) - len(errors),
191 "issues": [issue.as_dict() for issue in issues],
192 "source_content_copied_to_output": False,
193 }
196def build_parser() -> argparse.ArgumentParser:
197 parser = argparse.ArgumentParser(
198 description=(
199 "Verify local evidence identifiers and criterion mappings. "
200 "The report contains references and counts, never evidence excerpts."
201 )
202 )
203 parser.add_argument("--rubric", required=True, type=Path, help="Local rubric JSON")
204 parser.add_argument(
205 "--evaluation", required=True, type=Path, help="Local evaluation JSON"
206 )
207 parser.add_argument(
208 "--evidence", required=True, type=Path, help="Local evidence-manifest JSON"
209 )
210 parser.add_argument("--output", type=Path, help="Optional local JSON report")
211 parser.add_argument(
212 "--force", action="store_true", help="Replace an existing output JSON file"
213 )
214 return parser
217def main() -> int:
218 args = build_parser().parse_args()
219 try:
220 report = check_traceability(
221 _common.read_json(args.rubric),
222 _common.read_json(args.evaluation),
223 _common.read_json(args.evidence),
224 )
225 _common.write_json(report, args.output, force=args.force)
226 return 0 if report["status"] == "pass" else 2
227 except _common.ValidationError as error:
228 _common.write_json(_common.failure_report(error), None)
229 return 2
232if __name__ == "__main__":
233 raise SystemExit(main())