Skills › Scientific Agent Skills › clinical-reports › provenance_validator.py
provenance_validator.py
266 lines · skills/clinical-reports/scripts/provenance_validator.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate source-fact-to-claim traceability without opening source records."""
4from __future__ import annotations
6import argparse
7import re
8import sys
9from typing import Any
11sys.dont_write_bytecode = True
13from _common import ( # noqa: E402
14 SHA256_RE,
15 ValidationError,
16 error_report,
17 load_json_object,
18 parse_iso_date,
19 require_bool,
20 require_data_class,
21 require_exact_keys,
22 require_identifier,
23 require_string,
24 write_json_report,
25)
27TOOL = "provenance_validator"
28LOCAL_LOCATOR_RE = re.compile(r"^local:[A-Za-z0-9_./:-]{1,280}$")
29SOURCE_KINDS = {
30 "synthetic_record",
31 "deidentified_record",
32 "aggregate_output",
33 "protocol",
34 "statistical_analysis_plan",
35 "published_source",
36 "authorized_system_record",
37}
38FACT_FIELDS = {
39 "fact_id",
40 "source_record_kind",
41 "record_locator",
42 "field_path",
43 "value_hash_sha256",
44 "verification_status",
45 "verified_by_role",
46 "verified_at",
47 "source_version",
48}
49CLAIM_FIELDS = {
50 "claim_id",
51 "artifact_field_path",
52 "fact_ids",
53 "support_status",
54}
55TOP_LEVEL_FIELDS = {
56 "schema_version",
57 "artifact_kind",
58 "manifest_status",
59 "safety_notice",
60 "data_classification",
61 "authorized_purpose",
62 "authorization_verified",
63 "facts",
64 "claims",
65 "review",
66}
69def validate_provenance(data: dict[str, Any]) -> dict[str, Any]:
70 """Validate bounded metadata links while preserving source separation."""
71 errors: list[str] = []
72 warnings: list[str] = []
73 try:
74 require_exact_keys(data, TOP_LEVEL_FIELDS, "manifest")
75 if data.get("schema_version") != "2.0":
76 raise ValidationError("schema_version must be 2.0")
77 if data.get("manifest_status") != "BLOCKED_INCOMPLETE":
78 raise ValidationError("manifest_status must remain BLOCKED_INCOMPLETE")
79 require_string(data.get("safety_notice"), "safety_notice", max_length=1000)
80 except ValidationError as exc:
81 errors.append(str(exc))
82 if data.get("artifact_kind") != "provenance_manifest":
83 errors.append("artifact_kind must be provenance_manifest")
84 try:
85 require_data_class(data.get("data_classification"))
86 require_string(data.get("authorized_purpose"), "authorized_purpose")
87 if not require_bool(data.get("authorization_verified"), "authorization_verified"):
88 errors.append("authorization_verified must be true")
89 except ValidationError as exc:
90 errors.append(str(exc))
92 facts = data.get("facts")
93 claims = data.get("claims")
94 if not isinstance(facts, list) or not facts:
95 errors.append("facts must be a non-empty array")
96 facts = []
97 if not isinstance(claims, list) or not claims:
98 errors.append("claims must be a non-empty array")
99 claims = []
100 if len(facts) > 10_000:
101 errors.append("facts may contain at most 10,000 items")
102 if len(claims) > 10_000:
103 errors.append("claims may contain at most 10,000 items")
105 fact_ids: set[str] = set()
106 for index, fact in enumerate(facts[:10_000]):
107 try:
108 fact = require_exact_keys(fact, FACT_FIELDS, f"facts[{index}]")
109 except ValidationError as exc:
110 errors.append(str(exc))
111 continue
112 try:
113 fact_id = require_identifier(fact.get("fact_id"), f"facts[{index}].fact_id")
114 if fact_id in fact_ids:
115 errors.append(f"duplicate fact_id: {fact_id}")
116 fact_ids.add(fact_id)
117 if (
118 not isinstance(fact.get("source_record_kind"), str)
119 or fact.get("source_record_kind") not in SOURCE_KINDS
120 ):
121 errors.append(f"facts[{index}].source_record_kind is invalid")
122 locator = require_string(
123 fact.get("record_locator"),
124 f"facts[{index}].record_locator",
125 max_length=286,
126 )
127 if not LOCAL_LOCATOR_RE.fullmatch(locator):
128 errors.append(
129 f"facts[{index}].record_locator must be a local: locator"
130 )
131 require_string(
132 fact.get("field_path"),
133 f"facts[{index}].field_path",
134 max_length=300,
135 )
136 value_hash = require_string(
137 fact.get("value_hash_sha256"),
138 f"facts[{index}].value_hash_sha256",
139 max_length=64,
140 )
141 if not SHA256_RE.fullmatch(value_hash):
142 errors.append(f"facts[{index}].value_hash_sha256 is invalid")
143 if fact.get("verification_status") != "verified":
144 errors.append(f"facts[{index}] is not verified")
145 require_string(
146 fact.get("verified_by_role"),
147 f"facts[{index}].verified_by_role",
148 max_length=100,
149 )
150 parse_iso_date(fact.get("verified_at"), f"facts[{index}].verified_at")
151 require_string(
152 fact.get("source_version"),
153 f"facts[{index}].source_version",
154 max_length=100,
155 )
156 except ValidationError as exc:
157 errors.append(str(exc))
159 claim_ids: set[str] = set()
160 referenced_facts: set[str] = set()
161 for index, claim in enumerate(claims[:10_000]):
162 try:
163 claim = require_exact_keys(claim, CLAIM_FIELDS, f"claims[{index}]")
164 except ValidationError as exc:
165 errors.append(str(exc))
166 continue
167 try:
168 claim_id = require_identifier(
169 claim.get("claim_id"),
170 f"claims[{index}].claim_id",
171 )
172 if claim_id in claim_ids:
173 errors.append(f"duplicate claim_id: {claim_id}")
174 claim_ids.add(claim_id)
175 require_string(
176 claim.get("artifact_field_path"),
177 f"claims[{index}].artifact_field_path",
178 max_length=300,
179 )
180 links = claim.get("fact_ids")
181 if not isinstance(links, list) or not links:
182 errors.append(f"claims[{index}].fact_ids must be a non-empty array")
183 links = []
184 for linked in links:
185 linked_id = require_identifier(linked, f"claims[{index}].fact_ids")
186 referenced_facts.add(linked_id)
187 if linked_id not in fact_ids:
188 errors.append(
189 f"claims[{index}] references unknown fact_id {linked_id}"
190 )
191 if claim.get("support_status") != "supported_by_verified_facts":
192 errors.append(f"claims[{index}] is not marked supported_by_verified_facts")
193 except ValidationError as exc:
194 errors.append(str(exc))
196 unused = sorted(fact_ids - referenced_facts)
197 if unused:
198 warnings.append(f"{len(unused)} verified facts are not linked to a claim")
200 review = data.get("review")
201 try:
202 review = require_exact_keys(
203 review,
204 {
205 "source_owner_review",
206 "quality_review",
207 "privacy_review",
208 "release_authorized",
209 },
210 "review",
211 )
212 for field in ("source_owner_review", "quality_review", "privacy_review"):
213 if review.get(field) != "completed":
214 warnings.append(f"review.{field} remains required")
215 if review.get("release_authorized") is not False:
216 errors.append("review.release_authorized must remain false")
217 except ValidationError as exc:
218 errors.append(str(exc))
220 return {
221 "tool": TOOL,
222 "status": "BLOCKED" if errors else "TRACEABILITY_COMPLETE_REVIEW_REQUIRED",
223 "fact_count": len(facts),
224 "claim_count": len(claims),
225 "linked_fact_count": len(referenced_facts),
226 "errors": errors,
227 "warnings": warnings,
228 "limitations": [
229 "Source records were not opened and hash values were not recomputed.",
230 "Traceability metadata does not establish source truth or clinical correctness.",
231 ],
232 "review_required": True,
233 "authorizes_clinical_use_or_submission": False,
234 }
237def build_parser() -> argparse.ArgumentParser:
238 parser = argparse.ArgumentParser(
239 description=(
240 "Check bounded local source-fact traceability metadata. "
241 "Does not open source records or copy clinical content."
242 )
243 )
244 parser.add_argument("input_file", help="Local provenance manifest (.json)")
245 parser.add_argument("-o", "--output", help="Optional local JSON report")
246 parser.add_argument("--overwrite", action="store_true")
247 return parser
250def main(argv: list[str] | None = None) -> int:
251 args = build_parser().parse_args(argv)
252 try:
253 _, data = load_json_object(args.input_file)
254 report = validate_provenance(data)
255 except (OSError, ValidationError) as exc:
256 report = error_report(TOOL, exc)
257 try:
258 write_json_report(report, args.output, overwrite=args.overwrite)
259 except (OSError, ValidationError) as exc:
260 print(str(exc), file=sys.stderr)
261 return 2
262 return 0 if report["status"] == "TRACEABILITY_COMPLETE_REVIEW_REQUIRED" else 1
265if __name__ == "__main__":
266 raise SystemExit(main())