Skills › Scientific Agent Skills › treatment-plans › check_consistency.py
check_consistency.py
386 lines · skills/treatment-plans/scripts/check_consistency.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check cross-record consistency without clinical interpretation."""
4from __future__ import annotations
6import argparse
7import sys
8from datetime import date
9from typing import Any
11from _common import (
12 Issue,
13 ValidationError,
14 load_package,
15 parse_iso_datetime,
16 print_report,
17 record_ids,
18 report_payload,
19 validate_package_structure,
20 error_report,
21)
24def _collect_ids(documents: dict[str, dict]) -> list[tuple[str, str]]:
25 values: list[tuple[str, str]] = []
27 def collect(records: list, field: str, path: str) -> None:
28 for index, record in enumerate(records):
29 if isinstance(record, dict) and isinstance(record.get(field), str):
30 values.append((f"{path}[{index}].{field}", record[field]))
32 facts = documents["source_fact_manifest"]["facts"]
33 collect(facts, "fact_id", "source_fact_manifest:$.facts")
35 interventions = documents[
36 "clinician_authored_intervention_record"
37 ]["interventions"]
38 collect(
39 interventions,
40 "intervention_id",
41 "clinician_authored_intervention_record:$.interventions",
42 )
43 for index, record in enumerate(interventions):
44 collect(
45 record["parameters_as_supplied"],
46 "parameter_id",
47 (
48 "clinician_authored_intervention_record:"
49 f"$.interventions[{index}].parameters_as_supplied"
50 ),
51 )
53 planning = documents["goals_monitoring_checkpoint_record"]
54 collect(
55 planning["goals"],
56 "goal_id",
57 "goals_monitoring_checkpoint_record:$.goals",
58 )
59 collect(
60 planning["monitoring_items"],
61 "monitoring_id",
62 "goals_monitoring_checkpoint_record:$.monitoring_items",
63 )
64 collect(
65 planning["checkpoints"],
66 "checkpoint_id",
67 "goals_monitoring_checkpoint_record:$.checkpoints",
68 )
70 entries = documents[
71 "informed_preference_shared_decision_record"
72 ]["entries"]
73 collect(
74 entries,
75 "decision_id",
76 "informed_preference_shared_decision_record:$.entries",
77 )
78 for index, record in enumerate(entries):
79 collect(
80 record["options_as_documented"],
81 "option_id",
82 (
83 "informed_preference_shared_decision_record:"
84 f"$.entries[{index}].options_as_documented"
85 ),
86 )
88 transition = documents["transition_reconciliation_record"]
89 collect(
90 transition["handoff_items"],
91 "item_id",
92 "transition_reconciliation_record:$.handoff_items",
93 )
94 collect(
95 transition["unresolved_items"],
96 "item_id",
97 "transition_reconciliation_record:$.unresolved_items",
98 )
99 return values
102def _duplicate_reference_issues(
103 value: Any, path: str = "$"
104) -> list[Issue]:
105 issues: list[Issue] = []
106 if isinstance(value, dict):
107 for key, nested in value.items():
108 nested_path = f"{path}.{key}"
109 if key.endswith("_ids") and isinstance(nested, list):
110 if len(nested) != len(set(nested)):
111 issues.append(
112 Issue("DUPLICATE_REFERENCE_IN_LIST", nested_path)
113 )
114 issues.extend(_duplicate_reference_issues(nested, nested_path))
115 elif isinstance(value, list):
116 for index, nested in enumerate(value):
117 issues.extend(
118 _duplicate_reference_issues(nested, f"{path}[{index}]")
119 )
120 return issues
123def check_consistency(documents: dict[str, dict]) -> dict:
124 issues = validate_package_structure(documents)
125 if issues:
126 return report_payload(
127 "cross_record_consistency",
128 issues,
129 counts={"structural_issues": len(issues)},
130 )
132 handoff_document = documents["intended_use_handoff_record"]
133 subject_ref = handoff_document["subject_ref"]
134 classification = handoff_document["data_classification"]
135 for document_type, document in documents.items():
136 if document["subject_ref"] != subject_ref:
137 issues.append(
138 Issue(
139 "SUBJECT_REFERENCE_MISMATCH",
140 f"{document_type}:$.subject_ref",
141 )
142 )
143 if document["data_classification"] != classification:
144 issues.append(
145 Issue(
146 "DATA_CLASSIFICATION_MISMATCH",
147 f"{document_type}:$.data_classification",
148 )
149 )
150 issues.extend(
151 Issue(issue.code, f"{document_type}:{issue.path}", issue.level)
152 for issue in _duplicate_reference_issues(document)
153 )
155 seen: set[str] = set()
156 for path, identifier in _collect_ids(documents):
157 if identifier in seen:
158 issues.append(Issue("GLOBAL_RECORD_ID_COLLISION", path))
159 seen.add(identifier)
161 interventions = documents[
162 "clinician_authored_intervention_record"
163 ]["interventions"]
164 for index, record in enumerate(interventions):
165 start = record["start_date"]
166 end = record["end_date"]
167 if start is not None and end is not None:
168 if date.fromisoformat(start) > date.fromisoformat(end):
169 issues.append(
170 Issue(
171 "INTERVENTION_DATE_ORDER_INVALID",
172 (
173 "clinician_authored_intervention_record:"
174 f"$.interventions[{index}]"
175 ),
176 )
177 )
179 planning = documents["goals_monitoring_checkpoint_record"]
180 goal_ids = record_ids(planning, "goals", "goal_id")
181 intervention_ids = record_ids(
182 documents["clinician_authored_intervention_record"],
183 "interventions",
184 "intervention_id",
185 )
186 for index, checkpoint in enumerate(planning["checkpoints"]):
187 base = (
188 "goals_monitoring_checkpoint_record:"
189 f"$.checkpoints[{index}]"
190 )
191 for reference_index, goal_id in enumerate(
192 checkpoint["linked_goal_ids"]
193 ):
194 if goal_id not in goal_ids:
195 issues.append(
196 Issue(
197 "CHECKPOINT_UNKNOWN_GOAL",
198 f"{base}.linked_goal_ids[{reference_index}]",
199 )
200 )
201 for reference_index, intervention_id in enumerate(
202 checkpoint["linked_intervention_ids"]
203 ):
204 if intervention_id not in intervention_ids:
205 issues.append(
206 Issue(
207 "CHECKPOINT_UNKNOWN_INTERVENTION",
208 (
209 f"{base}.linked_intervention_ids"
210 f"[{reference_index}]"
211 ),
212 )
213 )
215 transition = documents["transition_reconciliation_record"]
216 reconciliation = transition["medication_reconciliation"]
217 if reconciliation["status"] == "pending_authorized_review":
218 if reconciliation["completed_by_role"] or reconciliation[
219 "completed_at"
220 ] is not None:
221 issues.append(
222 Issue(
223 "PENDING_RECONCILIATION_HAS_COMPLETION_DATA",
224 (
225 "transition_reconciliation_record:"
226 "$.medication_reconciliation"
227 ),
228 )
229 )
230 elif reconciliation["status"] == "not_applicable_by_authorized_clinician":
231 if (
232 reconciliation["source_list_fact_ids"]
233 or reconciliation["destination_list_fact_ids"]
234 ):
235 issues.append(
236 Issue(
237 "NOT_APPLICABLE_RECONCILIATION_HAS_LIST_REFERENCES",
238 (
239 "transition_reconciliation_record:"
240 "$.medication_reconciliation"
241 ),
242 )
243 )
244 elif reconciliation["status"] == "completed_by_authorized_clinician":
245 if (
246 not reconciliation["completed_by_role"]
247 or reconciliation["completed_at"] is None
248 ):
249 issues.append(
250 Issue(
251 "COMPLETED_RECONCILIATION_MISSING_REVIEW",
252 (
253 "transition_reconciliation_record:"
254 "$.medication_reconciliation"
255 ),
256 )
257 )
259 privacy = handoff_document["privacy_process"]
260 signoff = handoff_document["clinician_signoff"]
261 release = handoff_document["release_gate"]
262 handoff = handoff_document["handoff"]
264 if privacy["data_classification"] != classification:
265 issues.append(
266 Issue(
267 "PRIVACY_CLASSIFICATION_MISMATCH",
268 (
269 "intended_use_handoff_record:"
270 "$.privacy_process.data_classification"
271 ),
272 )
273 )
275 if release["status"] == "released_for_authorized_documentation_handoff":
276 if signoff["status"] != "signed":
277 issues.append(
278 Issue(
279 "RELEASE_WITHOUT_CLINICIAN_SIGNOFF",
280 "intended_use_handoff_record:$.release_gate.status",
281 )
282 )
283 if release["blocker_codes"]:
284 issues.append(
285 Issue(
286 "RELEASE_WITH_BLOCKERS",
287 (
288 "intended_use_handoff_record:"
289 "$.release_gate.blocker_codes"
290 ),
291 )
292 )
293 if handoff["acknowledgment_status"] != "acknowledged":
294 issues.append(
295 Issue(
296 "RELEASE_WITHOUT_HANDOFF_ACKNOWLEDGMENT",
297 (
298 "intended_use_handoff_record:"
299 "$.handoff.acknowledgment_status"
300 ),
301 )
302 )
304 signed_at = (
305 parse_iso_datetime(signoff["signed_at"])
306 if signoff["signed_at"] is not None
307 else None
308 )
309 released_at = (
310 parse_iso_datetime(release["released_at"])
311 if release["released_at"] is not None
312 else None
313 )
314 sent_at = (
315 parse_iso_datetime(handoff["sent_at"])
316 if handoff["sent_at"] is not None
317 else None
318 )
319 reviewed_at = (
320 parse_iso_datetime(privacy["reviewed_at"])
321 if privacy["reviewed_at"] is not None
322 else None
323 )
325 if signed_at and released_at and signed_at > released_at:
326 issues.append(
327 Issue(
328 "SIGNOFF_AFTER_RELEASE",
329 "intended_use_handoff_record:$.release_gate.released_at",
330 )
331 )
332 if released_at and sent_at and released_at > sent_at:
333 issues.append(
334 Issue(
335 "RELEASE_AFTER_HANDOFF_SENT",
336 "intended_use_handoff_record:$.handoff.sent_at",
337 )
338 )
339 if reviewed_at and released_at and reviewed_at > released_at:
340 issues.append(
341 Issue(
342 "PRIVACY_REVIEW_AFTER_RELEASE",
343 "intended_use_handoff_record:$.privacy_process.reviewed_at",
344 )
345 )
347 issues.sort(key=lambda item: (item.path, item.code))
348 return report_payload(
349 "cross_record_consistency",
350 issues,
351 counts={
352 "global_record_ids": len(seen),
353 "checkpoints": len(planning["checkpoints"]),
354 "interventions": len(interventions),
355 },
356 extra={
357 "clinical_semantics_compared": False,
358 "date_intervals_inferred": False,
359 },
360 )
363def build_parser() -> argparse.ArgumentParser:
364 parser = argparse.ArgumentParser(
365 description=(
366 "Check identifiers, links, classifications, status coherence, and "
367 "explicit date order. Clinical text is not compared or interpreted."
368 )
369 )
370 parser.add_argument("package", help="Complete local package directory")
371 return parser
374def main() -> int:
375 args = build_parser().parse_args()
376 try:
377 documents, _ = load_package(args.package)
378 report = check_consistency(documents)
379 except ValidationError as exc:
380 report = error_report("cross_record_consistency", exc)
381 print_report(report)
382 return 1 if report["status"] == "fail" else 0
385if __name__ == "__main__":
386 sys.exit(main())