Skills › Scientific Agent Skills › treatment-plans › validate_traceability.py
validate_traceability.py
147 lines · skills/treatment-plans/scripts/validate_traceability.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate source-fact traceability without interpreting source content."""
4from __future__ import annotations
6import argparse
7import sys
9from _common import (
10 Issue,
11 ValidationError,
12 all_fact_references,
13 error_report,
14 load_package,
15 print_report,
16 report_payload,
17 validate_package_structure,
18)
21def validate_traceability(documents: dict[str, dict]) -> dict:
22 issues = validate_package_structure(documents)
23 if issues:
24 return report_payload(
25 "source_fact_traceability",
26 issues,
27 counts={"structural_issues": len(issues)},
28 )
30 manifest = documents["source_fact_manifest"]
31 facts = manifest["facts"]
32 fact_index = {
33 fact["fact_id"]: (index, fact) for index, fact in enumerate(facts)
34 }
35 references = all_fact_references(documents)
36 referenced_ids: set[str] = set()
38 for path, fact_id in references:
39 record = fact_index.get(fact_id)
40 if record is None:
41 issues.append(Issue("UNKNOWN_SOURCE_FACT_REFERENCE", path))
42 continue
43 referenced_ids.add(fact_id)
44 index, fact = record
45 verification = fact["verification"]
46 if verification["status"] != "verified":
47 issues.append(
48 Issue(
49 "REFERENCED_FACT_NOT_VERIFIED",
50 f"source_fact_manifest:$.facts[{index}].verification.status",
51 )
52 )
54 official_types = {
55 "current_fda_labeling",
56 "current_rems_material",
57 "current_official_guidance",
58 "local_policy",
59 }
60 for index, fact in enumerate(facts):
61 base = f"source_fact_manifest:$.facts[{index}]"
62 verification = fact["verification"]
63 if verification["status"] == "verified":
64 if not verification["verified_by_role"]:
65 issues.append(
66 Issue(
67 "VERIFIED_FACT_REVIEWER_ROLE_MISSING",
68 f"{base}.verification.verified_by_role",
69 )
70 )
71 if verification["verified_at"] is None:
72 issues.append(
73 Issue(
74 "VERIFIED_FACT_TIME_MISSING",
75 f"{base}.verification.verified_at",
76 )
77 )
78 source_type = fact["source"]["source_type"]
79 applicability = fact["applicability"]
80 if (
81 source_type in official_types
82 and applicability["status"]
83 != "confirmed_by_authorized_clinician"
84 ):
85 issues.append(
86 Issue(
87 "SOURCE_APPLICABILITY_NOT_CONFIRMED",
88 f"{base}.applicability.status",
89 )
90 )
91 if (
92 applicability["status"]
93 == "confirmed_by_authorized_clinician"
94 and (
95 not applicability["confirmed_by_role"]
96 or applicability["confirmed_at"] is None
97 )
98 ):
99 issues.append(
100 Issue(
101 "SOURCE_APPLICABILITY_CONFIRMATION_INCOMPLETE",
102 f"{base}.applicability",
103 )
104 )
105 if fact["fact_id"] not in referenced_ids:
106 issues.append(Issue("ORPHAN_SOURCE_FACT", base, "warning"))
108 issues.sort(key=lambda item: (item.path, item.code))
109 return report_payload(
110 "source_fact_traceability",
111 issues,
112 counts={
113 "facts": len(facts),
114 "fact_references": len(references),
115 "referenced_facts": len(referenced_ids),
116 },
117 extra={
118 "source_content_interpreted": False,
119 "source_applicability_determined_by_script": False,
120 },
121 )
124def build_parser() -> argparse.ArgumentParser:
125 parser = argparse.ArgumentParser(
126 description=(
127 "Check that every package fact reference exists and points to a "
128 "human-verified source fact. Source content is not interpreted."
129 )
130 )
131 parser.add_argument("package", help="Complete local package directory")
132 return parser
135def main() -> int:
136 args = build_parser().parse_args()
137 try:
138 documents, _ = load_package(args.package)
139 report = validate_traceability(documents)
140 except ValidationError as exc:
141 report = error_report("source_fact_traceability", exc)
142 print_report(report)
143 return 1 if report["status"] == "fail" else 0
146if __name__ == "__main__":
147 sys.exit(main())