Skills › Scientific Agent Skills › treatment-plans › check_completeness.py
check_completeness.py
573 lines · skills/treatment-plans/scripts/check_completeness.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check documentation gates without assessing clinical adequacy."""
4from __future__ import annotations
6import argparse
7import sys
9from _common import (
10 Issue,
11 ValidationError,
12 error_report,
13 load_package,
14 print_report,
15 report_payload,
16 validate_package_structure,
17)
20def _required_text(value: object, path: str, issues: list[Issue]) -> None:
21 if not isinstance(value, str) or not value.strip():
22 issues.append(Issue("REQUIRED_VALUE_MISSING", path))
25def _required_time(value: object, path: str, issues: list[Issue]) -> None:
26 if value is None:
27 issues.append(Issue("REQUIRED_TIME_MISSING", path))
30def check_completeness(documents: dict[str, dict]) -> dict:
31 issues = validate_package_structure(documents)
32 if issues:
33 return report_payload(
34 "documentation_completeness",
35 issues,
36 counts={"structural_issues": len(issues)},
37 )
39 for document_type, document in documents.items():
40 if "TEMPLATE" in document["document_id"]:
41 issues.append(
42 Issue(
43 "TEMPLATE_DOCUMENT_ID_UNRESOLVED",
44 f"{document_type}:$.document_id",
45 )
46 )
47 _required_time(
48 document["created_at"], f"{document_type}:$.created_at", issues
49 )
51 facts = documents["source_fact_manifest"]["facts"]
52 if not facts:
53 issues.append(Issue("SOURCE_FACTS_REQUIRED", "source_fact_manifest:$.facts"))
54 for index, fact in enumerate(facts):
55 base = f"source_fact_manifest:$.facts[{index}]"
56 verification = fact["verification"]
57 if verification["status"] != "verified":
58 issues.append(
59 Issue(
60 "SOURCE_FACT_VERIFICATION_PENDING",
61 f"{base}.verification.status",
62 )
63 )
64 _required_text(
65 verification["verified_by_role"],
66 f"{base}.verification.verified_by_role",
67 issues,
68 )
69 _required_time(
70 verification["verified_at"],
71 f"{base}.verification.verified_at",
72 issues,
73 )
75 interventions = documents[
76 "clinician_authored_intervention_record"
77 ]["interventions"]
78 if not interventions:
79 issues.append(
80 Issue(
81 "CLINICIAN_AUTHORED_INTERVENTION_REQUIRED",
82 "clinician_authored_intervention_record:$.interventions",
83 )
84 )
85 for index, intervention in enumerate(interventions):
86 base = (
87 "clinician_authored_intervention_record:"
88 f"$.interventions[{index}]"
89 )
90 if (
91 intervention["clinical_decision_status"]
92 != "supplied_and_verified_by_authorized_clinician"
93 ):
94 issues.append(
95 Issue(
96 "INTERVENTION_AUTHORIZED_VERIFICATION_PENDING",
97 f"{base}.clinical_decision_status",
98 )
99 )
100 verification = intervention["verification"]
101 if verification["status"] != "verified":
102 issues.append(
103 Issue(
104 "INTERVENTION_VERIFICATION_PENDING",
105 f"{base}.verification.status",
106 )
107 )
108 _required_text(
109 verification["verified_by_role"],
110 f"{base}.verification.verified_by_role",
111 issues,
112 )
113 _required_time(
114 verification["verified_at"],
115 f"{base}.verification.verified_at",
116 issues,
117 )
119 planning = documents["goals_monitoring_checkpoint_record"]
120 for field, code in (
121 ("goals", "GOAL_RECORD_REQUIRED"),
122 ("monitoring_items", "MONITORING_RECORD_REQUIRED"),
123 ("checkpoints", "CHECKPOINT_RECORD_REQUIRED"),
124 ):
125 if not planning[field]:
126 issues.append(
127 Issue(
128 code,
129 f"goals_monitoring_checkpoint_record:$.{field}",
130 )
131 )
133 decisions = documents[
134 "informed_preference_shared_decision_record"
135 ]["entries"]
136 if not decisions:
137 issues.append(
138 Issue(
139 "SHARED_DECISION_RECORD_REQUIRED",
140 "informed_preference_shared_decision_record:$.entries",
141 )
142 )
143 for index, decision in enumerate(decisions):
144 base = (
145 "informed_preference_shared_decision_record:"
146 f"$.entries[{index}]"
147 )
148 if decision["decision_status"] != "documented_by_authorized_clinician":
149 issues.append(
150 Issue(
151 "SHARED_DECISION_DOCUMENTATION_PENDING",
152 f"{base}.decision_status",
153 )
154 )
155 if not decision["benefits_harms_uncertainty_documented"]:
156 issues.append(
157 Issue(
158 "SHARED_DECISION_ELEMENTS_NOT_DOCUMENTED",
159 f"{base}.benefits_harms_uncertainty_documented",
160 )
161 )
162 _required_text(
163 decision["preference_as_documented"],
164 f"{base}.preference_as_documented",
165 issues,
166 )
167 _required_text(
168 decision["outcome_as_documented"],
169 f"{base}.outcome_as_documented",
170 issues,
171 )
172 _required_text(
173 decision["documented_by_role"],
174 f"{base}.documented_by_role",
175 issues,
176 )
177 _required_time(
178 decision["documented_at"],
179 f"{base}.documented_at",
180 issues,
181 )
182 if decision["acknowledgment_status"] == "pending":
183 issues.append(
184 Issue(
185 "SHARED_DECISION_ACKNOWLEDGMENT_PENDING",
186 f"{base}.acknowledgment_status",
187 )
188 )
190 transition_document = documents["transition_reconciliation_record"]
191 transition = transition_document["transition"]
192 for field in (
193 "from_setting_as_documented",
194 "to_setting_as_documented",
195 "responsible_sender_role",
196 "responsible_receiver_role",
197 ):
198 _required_text(
199 transition[field],
200 f"transition_reconciliation_record:$.transition.{field}",
201 issues,
202 )
203 _required_time(
204 transition["handoff_date"],
205 "transition_reconciliation_record:$.transition.handoff_date",
206 issues,
207 )
209 reconciliation = transition_document["medication_reconciliation"]
210 if reconciliation["status"] == "pending_authorized_review":
211 issues.append(
212 Issue(
213 "MEDICATION_RECONCILIATION_PENDING",
214 (
215 "transition_reconciliation_record:"
216 "$.medication_reconciliation.status"
217 ),
218 )
219 )
220 if reconciliation["status"] == "completed_by_authorized_clinician":
221 if not reconciliation["source_list_fact_ids"]:
222 issues.append(
223 Issue(
224 "RECONCILIATION_SOURCE_LIST_FACTS_REQUIRED",
225 (
226 "transition_reconciliation_record:"
227 "$.medication_reconciliation.source_list_fact_ids"
228 ),
229 )
230 )
231 if not reconciliation["destination_list_fact_ids"]:
232 issues.append(
233 Issue(
234 "RECONCILIATION_DESTINATION_LIST_FACTS_REQUIRED",
235 (
236 "transition_reconciliation_record:"
237 "$.medication_reconciliation.destination_list_fact_ids"
238 ),
239 )
240 )
241 if reconciliation["discrepancy_status"] == "not_assessed":
242 issues.append(
243 Issue(
244 "RECONCILIATION_DISCREPANCY_STATUS_PENDING",
245 (
246 "transition_reconciliation_record:"
247 "$.medication_reconciliation.discrepancy_status"
248 ),
249 )
250 )
251 if reconciliation["status"] != "pending_authorized_review":
252 _required_text(
253 reconciliation["completed_by_role"],
254 (
255 "transition_reconciliation_record:"
256 "$.medication_reconciliation.completed_by_role"
257 ),
258 issues,
259 )
260 _required_time(
261 reconciliation["completed_at"],
262 (
263 "transition_reconciliation_record:"
264 "$.medication_reconciliation.completed_at"
265 ),
266 issues,
267 )
269 if not transition_document["handoff_items"]:
270 issues.append(
271 Issue(
272 "HANDOFF_ITEM_REQUIRED",
273 "transition_reconciliation_record:$.handoff_items",
274 )
275 )
276 for index, item in enumerate(transition_document["handoff_items"]):
277 if item["acknowledgment_status"] == "pending":
278 issues.append(
279 Issue(
280 "HANDOFF_ITEM_ACKNOWLEDGMENT_PENDING",
281 (
282 "transition_reconciliation_record:"
283 f"$.handoff_items[{index}].acknowledgment_status"
284 ),
285 )
286 )
287 for index, item in enumerate(transition_document["unresolved_items"]):
288 if item["status"] == "open_unrouted":
289 issues.append(
290 Issue(
291 "UNRESOLVED_ITEM_NOT_ROUTED",
292 (
293 "transition_reconciliation_record:"
294 f"$.unresolved_items[{index}].status"
295 ),
296 )
297 )
298 if item["status"] != "resolved_by_authorized_professional":
299 _required_text(
300 item["routed_to_local_role"],
301 (
302 "transition_reconciliation_record:"
303 f"$.unresolved_items[{index}].routed_to_local_role"
304 ),
305 issues,
306 )
308 handoff_document = documents["intended_use_handoff_record"]
309 intended = handoff_document["intended_use"]
310 if not intended["intended_users"]:
311 issues.append(
312 Issue(
313 "INTENDED_USER_REQUIRED",
314 "intended_use_handoff_record:$.intended_use.intended_users",
315 )
316 )
317 _required_text(
318 intended["intended_setting"],
319 "intended_use_handoff_record:$.intended_use.intended_setting",
320 issues,
321 )
323 privacy = handoff_document["privacy_process"]
324 for field in (
325 "local_authorization_confirmed",
326 "minimum_necessary_confirmed",
327 "no_external_tools_confirmed",
328 "no_prompt_log_example_copy_confirmed",
329 ):
330 if not privacy[field]:
331 issues.append(
332 Issue(
333 "PRIVACY_PROCESS_ATTESTATION_PENDING",
334 f"intended_use_handoff_record:$.privacy_process.{field}",
335 )
336 )
337 _required_text(
338 privacy["authorized_environment_reference"],
339 (
340 "intended_use_handoff_record:"
341 "$.privacy_process.authorized_environment_reference"
342 ),
343 issues,
344 )
345 _required_text(
346 privacy["retention_policy_reference"],
347 (
348 "intended_use_handoff_record:"
349 "$.privacy_process.retention_policy_reference"
350 ),
351 issues,
352 )
353 classification = handoff_document["data_classification"]
354 expected_review = (
355 {"not_applicable_synthetic", "completed_by_qualified_reviewer"}
356 if classification == "synthetic"
357 else {"completed_by_qualified_reviewer"}
358 )
359 if privacy["privacy_review_status"] not in expected_review:
360 issues.append(
361 Issue(
362 "QUALIFIED_PRIVACY_REVIEW_PENDING",
363 (
364 "intended_use_handoff_record:"
365 "$.privacy_process.privacy_review_status"
366 ),
367 )
368 )
369 if privacy["privacy_review_status"] == "completed_by_qualified_reviewer":
370 _required_text(
371 privacy["privacy_reviewer_role"],
372 (
373 "intended_use_handoff_record:"
374 "$.privacy_process.privacy_reviewer_role"
375 ),
376 issues,
377 )
378 _required_time(
379 privacy["reviewed_at"],
380 "intended_use_handoff_record:$.privacy_process.reviewed_at",
381 issues,
382 )
383 if privacy["data_disposition_status"] == "pending":
384 issues.append(
385 Issue(
386 "DATA_DISPOSITION_PENDING",
387 (
388 "intended_use_handoff_record:"
389 "$.privacy_process.data_disposition_status"
390 ),
391 )
392 )
394 governance = handoff_document["local_governance"]
395 for field in (
396 "institution_or_organization_reference",
397 "clinical_owner_role",
398 "records_owner_role",
399 "change_control_owner_role",
400 ):
401 _required_text(
402 governance[field],
403 f"intended_use_handoff_record:$.local_governance.{field}",
404 issues,
405 )
406 if not governance["policy_references"]:
407 issues.append(
408 Issue(
409 "LOCAL_POLICY_REFERENCE_REQUIRED",
410 (
411 "intended_use_handoff_record:"
412 "$.local_governance.policy_references"
413 ),
414 )
415 )
417 emergency = handoff_document["emergency_routing"]
418 _required_text(
419 emergency["local_process_reference"],
420 (
421 "intended_use_handoff_record:"
422 "$.emergency_routing.local_process_reference"
423 ),
424 issues,
425 )
426 _required_text(
427 emergency["verified_by_role"],
428 (
429 "intended_use_handoff_record:"
430 "$.emergency_routing.verified_by_role"
431 ),
432 issues,
433 )
435 routes = handoff_document["reporting_routes"]
436 for field in (
437 "local_patient_safety_route",
438 "product_event_route",
439 "privacy_incident_route",
440 "responsible_role",
441 ):
442 _required_text(
443 routes[field],
444 f"intended_use_handoff_record:$.reporting_routes.{field}",
445 issues,
446 )
448 handoff = handoff_document["handoff"]
449 for field in ("sender_role", "recipient_role"):
450 _required_text(
451 handoff[field],
452 f"intended_use_handoff_record:$.handoff.{field}",
453 issues,
454 )
455 _required_time(
456 handoff["sent_at"],
457 "intended_use_handoff_record:$.handoff.sent_at",
458 issues,
459 )
460 if handoff["acknowledgment_status"] != "acknowledged":
461 issues.append(
462 Issue(
463 "PACKAGE_HANDOFF_ACKNOWLEDGMENT_PENDING",
464 "intended_use_handoff_record:$.handoff.acknowledgment_status",
465 )
466 )
467 if transition_document["unresolved_items"] and not handoff[
468 "unresolved_items_routed"
469 ]:
470 issues.append(
471 Issue(
472 "UNRESOLVED_ITEM_ROUTING_ATTESTATION_PENDING",
473 (
474 "intended_use_handoff_record:"
475 "$.handoff.unresolved_items_routed"
476 ),
477 )
478 )
480 signoff = handoff_document["clinician_signoff"]
481 if signoff["status"] != "signed":
482 issues.append(
483 Issue(
484 "CLINICIAN_SIGNOFF_PENDING",
485 "intended_use_handoff_record:$.clinician_signoff.status",
486 )
487 )
488 for field in ("signer_role", "credential_authority_reference"):
489 _required_text(
490 signoff[field],
491 f"intended_use_handoff_record:$.clinician_signoff.{field}",
492 issues,
493 )
494 _required_time(
495 signoff["signed_at"],
496 "intended_use_handoff_record:$.clinician_signoff.signed_at",
497 issues,
498 )
500 release = handoff_document["release_gate"]
501 if release["status"] != "released_for_authorized_documentation_handoff":
502 issues.append(
503 Issue(
504 "DOCUMENTATION_HANDOFF_RELEASE_BLOCKED",
505 "intended_use_handoff_record:$.release_gate.status",
506 )
507 )
508 _required_text(
509 release["released_by_role"],
510 "intended_use_handoff_record:$.release_gate.released_by_role",
511 issues,
512 )
513 _required_time(
514 release["released_at"],
515 "intended_use_handoff_record:$.release_gate.released_at",
516 issues,
517 )
518 if release["blocker_codes"]:
519 issues.append(
520 Issue(
521 "RELEASE_BLOCKER_CODES_REMAIN",
522 "intended_use_handoff_record:$.release_gate.blocker_codes",
523 )
524 )
526 issues.sort(key=lambda item: (item.path, item.code))
527 return report_payload(
528 "documentation_completeness",
529 issues,
530 counts={
531 "source_facts": len(facts),
532 "interventions": len(interventions),
533 "goals": len(planning["goals"]),
534 "monitoring_items": len(planning["monitoring_items"]),
535 "checkpoints": len(planning["checkpoints"]),
536 "shared_decision_entries": len(decisions),
537 "handoff_items": len(transition_document["handoff_items"]),
538 "unresolved_items": len(
539 transition_document["unresolved_items"]
540 ),
541 },
542 extra={
543 "documentation_declarations_complete": not issues,
544 "handoff_authorized_by_script": False,
545 "clinical_completeness_determined": False,
546 },
547 )
550def build_parser() -> argparse.ArgumentParser:
551 parser = argparse.ArgumentParser(
552 description=(
553 "Check required documentation, review, routing, sign-off, and "
554 "handoff declarations. No clinical adequacy is assessed."
555 )
556 )
557 parser.add_argument("package", help="Complete local package directory")
558 return parser
561def main() -> int:
562 args = build_parser().parse_args()
563 try:
564 documents, _ = load_package(args.package)
565 report = check_completeness(documents)
566 except ValidationError as exc:
567 report = error_report("documentation_completeness", exc)
568 print_report(report)
569 return 1 if report["status"] == "fail" else 0
572if __name__ == "__main__":
573 sys.exit(main())