Skills › Scientific Agent Skills › clinical-decision-support › validate_cds_artifact.py
validate_cds_artifact.py
269 lines · skills/clinical-decision-support/scripts/validate_cds_artifact.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate intended use and governance fields for research-only CDS artifacts."""
4from __future__ import annotations
6import argparse
7import sys
8from typing import Any
10from _common import (
11 InputError,
12 IssueLog,
13 load_json_object,
14 print_report,
15 require_list,
16 require_nonempty_text,
17 source_ids,
18 write_json,
19)
21ALLOWED_TYPES = {
22 "aggregate_cohort_evaluation",
23 "evidence_profile",
24 "model_biomarker_evaluation",
25 "research_analysis_plan",
26 "governance_traceability",
27 "privacy_process_checklist",
28}
29ALLOWED_STATUSES = {"draft", "evaluation_only", "retired"}
30ALLOWED_DATA_LEVELS = {"aggregate", "synthetic", "aggregate_and_synthetic"}
31REQUIRED_PROHIBITIONS = {
32 "patient-specific",
33 "diagnosis",
34 "treatment recommendation",
35 "dosing",
36 "triage",
37 "autonomous decision",
38 "alarm",
39 "bedside",
40 "live clinical",
41 "regulatory compliance",
42 "hipaa compliance",
43}
44UNSAFE_OUTPUT_KEYS = {
45 "alarm",
46 "care_plan",
47 "clinical_action",
48 "diagnosis",
49 "dose",
50 "dosing",
51 "patient_class",
52 "patient_prediction",
53 "recommendation",
54 "treatment_recommendation",
55 "triage",
56 "urgency",
57}
58PLACEHOLDER_MARKERS = ("REPLACE_", "REQUIRES_", "YYYY-MM-DD")
61def _find_unsafe_keys(value: Any, location: str = "$") -> list[str]:
62 findings: list[str] = []
63 if isinstance(value, dict):
64 for key, nested in value.items():
65 normalized = str(key).strip().lower()
66 if normalized in UNSAFE_OUTPUT_KEYS:
67 findings.append(f"{location}.{key}")
68 findings.extend(_find_unsafe_keys(nested, f"{location}.{key}"))
69 elif isinstance(value, list):
70 for index, nested in enumerate(value):
71 findings.extend(_find_unsafe_keys(nested, f"{location}[{index}]"))
72 return findings
75def _find_placeholders(value: Any, location: str = "$") -> list[str]:
76 findings: list[str] = []
77 if isinstance(value, dict):
78 for key, nested in value.items():
79 findings.extend(_find_placeholders(nested, f"{location}.{key}"))
80 elif isinstance(value, list):
81 for index, nested in enumerate(value):
82 findings.extend(_find_placeholders(nested, f"{location}[{index}]"))
83 elif isinstance(value, str) and any(marker in value for marker in PLACEHOLDER_MARKERS):
84 findings.append(location)
85 return findings
88def validate_artifact(document: dict[str, Any]) -> IssueLog:
89 log = IssueLog()
90 try:
91 require_nonempty_text(document.get("schema_version"), "schema_version", max_length=20)
92 known_sources = source_ids(document)
93 if not known_sources:
94 log.errors.append("At least one source is required")
96 artifact = document.get("artifact")
97 if not isinstance(artifact, dict):
98 raise InputError("artifact must be an object")
99 for field in ("id", "title", "version", "owner", "date", "change_summary"):
100 require_nonempty_text(artifact.get(field), f"artifact.{field}")
101 artifact_type = require_nonempty_text(
102 artifact.get("type"), "artifact.type", max_length=100
103 )
104 if artifact_type not in ALLOWED_TYPES:
105 log.errors.append(f"artifact.type is not allowed: {artifact_type}")
106 status = require_nonempty_text(
107 artifact.get("status"), "artifact.status", max_length=30
108 )
109 if status not in ALLOWED_STATUSES:
110 log.errors.append(
111 "artifact.status must be draft, evaluation_only, or retired"
112 )
114 intended = document.get("intended_use")
115 if not isinstance(intended, dict):
116 raise InputError("intended_use must be an object")
117 require_nonempty_text(intended.get("purpose"), "intended_use.purpose")
118 users = require_list(
119 intended.get("intended_users"), "intended_use.intended_users", maximum=20
120 )
121 if not users:
122 log.errors.append("At least one intended user role is required")
123 for index, user in enumerate(users):
124 require_nonempty_text(user, f"intended_use.intended_users[{index}]")
125 if intended.get("population_scope") != "aggregate_or_synthetic_only":
126 log.errors.append(
127 "intended_use.population_scope must be aggregate_or_synthetic_only"
128 )
129 if intended.get("decision_role") != "research_evaluation_governance_only":
130 log.errors.append(
131 "intended_use.decision_role must be "
132 "research_evaluation_governance_only"
133 )
134 boundary = require_nonempty_text(
135 intended.get("boundary_statement"), "intended_use.boundary_statement"
136 ).lower()
137 if "not for patient care or live clinical use" not in boundary:
138 log.errors.append(
139 "Boundary statement must say: Not for patient care or live clinical use"
140 )
142 prohibited = require_list(
143 document.get("prohibited_uses"), "prohibited_uses", maximum=40
144 )
145 prohibited_text = " ".join(
146 require_nonempty_text(item, f"prohibited_uses[{index}]")
147 for index, item in enumerate(prohibited)
148 ).lower()
149 missing = sorted(
150 phrase for phrase in REQUIRED_PROHIBITIONS if phrase not in prohibited_text
151 )
152 if missing:
153 log.errors.append(
154 "prohibited_uses is missing required concepts: " + ", ".join(missing)
155 )
157 data = document.get("data_governance")
158 if not isinstance(data, dict):
159 raise InputError("data_governance must be an object")
160 if not isinstance(data.get("data_level"), str) or data.get("data_level") not in ALLOWED_DATA_LEVELS:
161 log.errors.append("data_governance.data_level must be aggregate or synthetic")
162 if data.get("phi_supplied") is not False:
163 log.errors.append("data_governance.phi_supplied must be false")
164 if data.get("raw_rows_supplied") is not False:
165 log.errors.append("data_governance.raw_rows_supplied must be false")
166 for field in ("provenance", "data_cut_date", "disclosure_policy"):
167 require_nonempty_text(data.get(field), f"data_governance.{field}")
169 limitations = require_list(
170 document.get("limitations"), "limitations", maximum=40
171 )
172 if len(limitations) < 3:
173 log.errors.append("At least three limitations are required")
174 for index, limitation in enumerate(limitations):
175 require_nonempty_text(limitation, f"limitations[{index}]")
177 review = document.get("human_review")
178 if not isinstance(review, dict):
179 raise InputError("human_review must be an object")
180 if review.get("required") is not True:
181 log.errors.append("human_review.required must be true")
182 roles = require_list(review.get("roles"), "human_review.roles", maximum=20)
183 if not roles:
184 log.errors.append("At least one human review role is required")
185 for index, role in enumerate(roles):
186 require_nonempty_text(role, f"human_review.roles[{index}]")
187 require_nonempty_text(
188 review.get("approval_boundary"), "human_review.approval_boundary"
189 )
190 if review.get("completed") is not True:
191 log.warnings.append("Human review is not recorded as complete")
193 validation = document.get("validation")
194 if not isinstance(validation, dict):
195 raise InputError("validation must be an object")
196 for field in (
197 "external_validation",
198 "calibration",
199 "subgroup_fairness",
200 "uncertainty",
201 "human_factors",
202 ):
203 require_nonempty_text(validation.get(field), f"validation.{field}")
205 lifecycle = document.get("lifecycle")
206 if not isinstance(lifecycle, dict):
207 raise InputError("lifecycle must be an object")
208 for field in (
209 "monitoring_plan",
210 "change_control",
211 "audit_plan",
212 "retirement_criteria",
213 ):
214 require_nonempty_text(lifecycle.get(field), f"lifecycle.{field}")
216 if document.get("compliance_claims") is not False:
217 log.errors.append("compliance_claims must be false")
219 unsafe_locations = _find_unsafe_keys(document)
220 if unsafe_locations:
221 log.errors.append(
222 "Unsafe clinical-output keys are prohibited: "
223 + ", ".join(unsafe_locations[:10])
224 )
225 placeholders = _find_placeholders(document)
226 if placeholders:
227 log.errors.append(
228 "Unresolved template placeholders remain: "
229 + ", ".join(placeholders[:10])
230 )
231 except InputError as exc:
232 log.errors.append(str(exc))
234 if log.ok:
235 log.info.append("Required intended-use and governance fields are complete")
236 return log
239def main() -> int:
240 parser = argparse.ArgumentParser(
241 description=(
242 "Validate a bounded local JSON metadata artifact. Structural checks only; "
243 "never a clinical or compliance determination."
244 )
245 )
246 parser.add_argument("input", help="Local JSON artifact")
247 parser.add_argument("-o", "--output", help="Optional local JSON report")
248 parser.add_argument(
249 "--strict", action="store_true", help="Return failure when warnings are present"
250 )
251 args = parser.parse_args()
253 try:
254 document = load_json_object(args.input)
255 result = validate_artifact(document).as_dict()
256 if args.output:
257 write_json(args.output, result)
258 print_report(result)
259 except InputError as exc:
260 result = IssueLog(errors=[str(exc)]).as_dict()
261 print_report(result)
262 return 2
263 if result["status"] != "pass" or (args.strict and result["warnings"]):
264 return 1
265 return 0
268if __name__ == "__main__":
269 sys.exit(main())