Skill Tracer
Skills › Scientific Agent Skills › treatment-plans › _common.py

_common.py

1191 lines · skills/treatment-plans/scripts/_common.py at 5f2bc03

Open on GitHub
1#!/usr/bin/env python3
2"""Bounded, dependency-free helpers for local treatment-plan JSON records."""
3
4from __future__ import annotations
5
6import json
7import os
8import re
9import tempfile
10from dataclasses import dataclass
11from datetime import date, datetime
12from pathlib import Path
13from typing import Any, Iterable
15SCHEMA_VERSION = "2.0"
16STATUS = "DRAFT_NOT_FOR_CLINICAL_USE"
17NOTICE = (
18 "DRAFT — NOT MEDICAL ADVICE — DOCUMENTATION-ONLY — "
19 "AUTHORIZED CLINICIAN SIGN-OFF REQUIRED"
21EMERGENCY_STATEMENT = (
22 "If a concern may be urgent or emergent, stop this documentation workflow "
23 "and use the institution's current clinical escalation or emergency process; "
24 "this package does not determine urgency or provide emergency instructions."
26SIGNOFF_ATTESTATION = (
27 "The signer attests only that this package accurately transcribes "
28 "clinician-supplied and verified decisions for authorized documentation "
29 "handoff; it is not stand-alone medical advice."
32MAX_INPUT_BYTES = 2 * 1024 * 1024
33MAX_TEXT_CHARS = 8_000
34MAX_LIST_ITEMS = 500
35MAX_OBJECT_FIELDS = 100
36MAX_DEPTH = 20
37MAX_TOTAL_NODES = 20_000
39IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9._:-]{1,95}$")
40DATETIME_RE = re.compile(
41 r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}"
42 r"(?:\.[0-9]{1,6})?(?:Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])"
44BLOCKER_RE = re.compile(r"^[A-Z][A-Z0-9_]{2,95}$")
45DIRECT_IDENTIFIER_KEYS = {
46 "address",
47 "date_of_birth",
48 "dob",
49 "email",
50 "fax",
51 "full_name",
52 "medical_record_number",
53 "mrn",
54 "name",
55 "patient_id",
56 "phone",
57 "social_security_number",
58 "ssn",
60PROHIBITED_USES = {
61 "diagnosis_or_assessment",
62 "therapy_selection_or_recommendation",
63 "medication_or_dose_decision",
64 "start_stop_hold_resume_titrate_or_taper_decision",
65 "interaction_allergy_or_contraindication_check",
66 "triage_or_urgency_determination",
67 "emergency_advice_or_safety_plan",
68 "prognosis_or_outcome_prediction",
69 "patient_specific_recommendation",
70 "replacement_of_authorized_professional_review",
73TEMPLATE_FILES = {
74 "source_fact_manifest": (
75 "source_fact_manifest_template.json",
76 "source_fact_manifest.json",
77 ),
78 "clinician_authored_intervention_record": (
79 "clinician_authored_intervention_template.json",
80 "clinician_authored_intervention.json",
81 ),
82 "goals_monitoring_checkpoint_record": (
83 "goals_monitoring_checkpoint_template.json",
84 "goals_monitoring_checkpoint.json",
85 ),
86 "informed_preference_shared_decision_record": (
87 "informed_preference_shared_decision_template.json",
88 "informed_preference_shared_decision.json",
89 ),
90 "transition_reconciliation_record": (
91 "transition_reconciliation_template.json",
92 "transition_reconciliation.json",
93 ),
94 "intended_use_handoff_record": (
95 "intended_use_handoff_template.json",
96 "intended_use_handoff.json",
97 ),
101class ValidationError(ValueError):
102 """A deterministic failure that never includes a clinical value."""
104 def __init__(self, code: str, path: str = "$") -> None:
105 super().__init__(code)
106 self.code = code
107 self.path = path
110@dataclass(frozen=True)
111class Issue:
112 """A minimized validation issue."""
114 code: str
115 path: str
116 level: str = "error"
118 def as_dict(self) -> dict[str, str]:
119 return {"code": self.code, "path": self.path, "level": self.level}
122def _text(
123 *,
124 minimum: int = 0,
125 maximum: int = MAX_TEXT_CHARS,
126 enum: Iterable[str] | None = None,
127 const: str | None = None,
128 pattern: str | None = None,
129 format_name: str | None = None,
130) -> dict[str, Any]:
131 return {
132 "kind": "string",
133 "minimum": minimum,
134 "maximum": maximum,
135 "enum": set(enum) if enum is not None else None,
136 "const": const,
137 "pattern": re.compile(pattern) if pattern else None,
138 "format": format_name,
139 }
142def _boolean() -> dict[str, str]:
143 return {"kind": "boolean"}
146def _nullable(schema: dict[str, Any]) -> dict[str, Any]:
147 return {"kind": "nullable", "schema": schema}
150def _array(
151 item: dict[str, Any], *, minimum: int = 0, maximum: int = MAX_LIST_ITEMS
152) -> dict[str, Any]:
153 return {
154 "kind": "array",
155 "item": item,
156 "minimum": minimum,
157 "maximum": maximum,
158 }
161def _object(properties: dict[str, dict[str, Any]]) -> dict[str, Any]:
162 return {"kind": "object", "properties": properties}
165IDENTIFIER = _text(minimum=2, maximum=96, pattern=IDENTIFIER_RE.pattern)
166ROLE = _text(maximum=200)
167NONEMPTY_ROLE = _text(minimum=1, maximum=200)
168DATE = _text(minimum=10, maximum=10, format_name="date")
169DATETIME = _text(minimum=20, maximum=40, format_name="datetime")
170FACT_IDS = _array(IDENTIFIER, minimum=1, maximum=100)
172HEADER = {
173 "schema_version": _text(const=SCHEMA_VERSION),
174 "document_type": _text(minimum=1, maximum=96),
175 "document_id": IDENTIFIER,
176 "subject_ref": IDENTIFIER,
177 "data_classification": _text(
178 enum={
179 "synthetic",
180 "deidentified_qualified_review",
181 "real_patient_minimum_necessary",
182 }
183 ),
184 "status": _text(const=STATUS),
185 "notice": _text(const=NOTICE),
186 "created_at": _nullable(DATETIME),
189VERIFICATION = _object(
190 {
191 "status": _text(enum={"pending", "verified"}),
192 "verified_by_role": ROLE,
193 "verified_at": _nullable(DATETIME),
194 }
197SOURCE_FACT_SCHEMA = _object(
198 {
199 "fact_id": IDENTIFIER,
200 "fact_kind": _text(
201 enum={
202 "clinician_authored_decision",
203 "verified_local_record",
204 "current_official_source",
205 "local_policy",
206 }
207 ),
208 "statement_as_supplied": _text(minimum=1, maximum=4_000),
209 "source": _object(
210 {
211 "source_type": _text(
212 enum={
213 "signed_clinician_record",
214 "authorized_ehr_record",
215 "current_fda_labeling",
216 "current_rems_material",
217 "current_official_guidance",
218 "local_policy",
219 }
220 ),
221 "title": _text(minimum=1, maximum=500),
222 "locator": _text(minimum=1, maximum=1_000),
223 "version_or_date": _text(minimum=1, maximum=200),
224 }
225 ),
226 "verification": VERIFICATION,
227 "applicability": _object(
228 {
229 "status": _text(
230 enum={
231 "not_assessed",
232 "confirmed_by_authorized_clinician",
233 "not_applicable_local_record",
234 }
235 ),
236 "confirmed_by_role": ROLE,
237 "confirmed_at": _nullable(DATETIME),
238 }
239 ),
240 }
243PARAMETER_SCHEMA = _object(
244 {
245 "parameter_id": IDENTIFIER,
246 "label": _text(minimum=1, maximum=200),
247 "value_as_supplied": _text(minimum=1, maximum=1_000),
248 "source_fact_ids": FACT_IDS,
249 }
252INTERVENTION_SCHEMA = _object(
253 {
254 "intervention_id": IDENTIFIER,
255 "intervention_kind": _text(
256 enum={
257 "medication",
258 "procedure",
259 "therapy",
260 "diagnostic",
261 "education",
262 "care_coordination",
263 "equipment_or_service",
264 "other_clinician_authored",
265 }
266 ),
267 "clinical_decision_status": _text(
268 enum={
269 "pending_authorized_verification",
270 "supplied_and_verified_by_authorized_clinician",
271 }
272 ),
273 "clinician_authored_action": _text(minimum=1, maximum=4_000),
274 "parameters_as_supplied": _array(PARAMETER_SCHEMA, maximum=100),
275 "source_fact_ids": FACT_IDS,
276 "owner_role": NONEMPTY_ROLE,
277 "start_date": _nullable(DATE),
278 "end_date": _nullable(DATE),
279 "verification": VERIFICATION,
280 }
283GOAL_SCHEMA = _object(
284 {
285 "goal_id": IDENTIFIER,
286 "statement_as_supplied": _text(minimum=1, maximum=2_000),
287 "measurement_as_supplied": _text(maximum=1_000),
288 "target_as_supplied": _text(maximum=1_000),
289 "target_date": _nullable(DATE),
290 "source_fact_ids": FACT_IDS,
291 "verified_by_role": NONEMPTY_ROLE,
292 }
295MONITORING_SCHEMA = _object(
296 {
297 "monitoring_id": IDENTIFIER,
298 "item_as_supplied": _text(minimum=1, maximum=2_000),
299 "method_as_supplied": _text(maximum=1_000),
300 "frequency_as_supplied": _text(maximum=1_000),
301 "next_due_date": _nullable(DATE),
302 "owner_role": NONEMPTY_ROLE,
303 "source_fact_ids": FACT_IDS,
304 }
307CHECKPOINT_SCHEMA = _object(
308 {
309 "checkpoint_id": IDENTIFIER,
310 "checkpoint_date": DATE,
311 "purpose_as_supplied": _text(minimum=1, maximum=2_000),
312 "owner_role": NONEMPTY_ROLE,
313 "linked_goal_ids": _array(IDENTIFIER, maximum=100),
314 "linked_intervention_ids": _array(IDENTIFIER, maximum=100),
315 "source_fact_ids": FACT_IDS,
316 }
319OPTION_SCHEMA = _object(
320 {
321 "option_id": IDENTIFIER,
322 "option_as_documented": _text(minimum=1, maximum=2_000),
323 "source_fact_ids": FACT_IDS,
324 }
327SHARED_DECISION_SCHEMA = _object(
328 {
329 "decision_id": IDENTIFIER,
330 "decision_topic_as_documented": _text(minimum=1, maximum=2_000),
331 "options_as_documented": _array(OPTION_SCHEMA, minimum=1, maximum=50),
332 "benefits_harms_uncertainty_documented": _boolean(),
333 "preference_as_documented": _text(maximum=2_000),
334 "outcome_as_documented": _text(maximum=2_000),
335 "outcome_source_fact_ids": FACT_IDS,
336 "decision_status": _text(
337 enum={
338 "pending_clinician_documentation",
339 "documented_by_authorized_clinician",
340 }
341 ),
342 "participant_roles": _array(
343 _text(minimum=1, maximum=200), minimum=1, maximum=20
344 ),
345 "documented_by_role": ROLE,
346 "documented_at": _nullable(DATETIME),
347 "acknowledgment_status": _text(
348 enum={
349 "pending",
350 "acknowledged",
351 "not_applicable_by_authorized_clinician",
352 }
353 ),
354 }
357HANDOFF_ITEM_SCHEMA = _object(
358 {
359 "item_id": IDENTIFIER,
360 "category": _text(
361 enum={
362 "medications",
363 "interventions",
364 "goals",
365 "monitoring",
366 "pending_results",
367 "appointments",
368 "equipment_or_services",
369 "preferences_or_access_needs",
370 "local_process_reference",
371 "other_clinician_supplied",
372 }
373 ),
374 "summary_as_supplied": _text(minimum=1, maximum=2_000),
375 "source_fact_ids": FACT_IDS,
376 "owner_role": NONEMPTY_ROLE,
377 "recipient_role": NONEMPTY_ROLE,
378 "acknowledgment_status": _text(
379 enum={
380 "pending",
381 "acknowledged",
382 "not_applicable_by_authorized_role",
383 }
384 ),
385 }
388UNRESOLVED_ITEM_SCHEMA = _object(
389 {
390 "item_id": IDENTIFIER,
391 "issue_as_documented": _text(minimum=1, maximum=2_000),
392 "source_fact_ids": FACT_IDS,
393 "routed_to_local_role": ROLE,
394 "status": _text(
395 enum={
396 "open_unrouted",
397 "open_routed",
398 "acknowledged_by_owner",
399 "resolved_by_authorized_professional",
400 }
401 ),
402 }
405DOCUMENT_SCHEMAS = {
406 "source_fact_manifest": _object(
407 {
408 **HEADER,
409 "document_type": _text(const="source_fact_manifest"),
410 "facts": _array(SOURCE_FACT_SCHEMA, maximum=500),
411 }
412 ),
413 "clinician_authored_intervention_record": _object(
414 {
415 **HEADER,
416 "document_type": _text(
417 const="clinician_authored_intervention_record"
418 ),
419 "interventions": _array(INTERVENTION_SCHEMA, maximum=200),
420 }
421 ),
422 "goals_monitoring_checkpoint_record": _object(
423 {
424 **HEADER,
425 "document_type": _text(
426 const="goals_monitoring_checkpoint_record"
427 ),
428 "goals": _array(GOAL_SCHEMA, maximum=200),
429 "monitoring_items": _array(MONITORING_SCHEMA, maximum=300),
430 "checkpoints": _array(CHECKPOINT_SCHEMA, maximum=500),
431 }
432 ),
433 "informed_preference_shared_decision_record": _object(
434 {
435 **HEADER,
436 "document_type": _text(
437 const="informed_preference_shared_decision_record"
438 ),
439 "entries": _array(SHARED_DECISION_SCHEMA, maximum=100),
440 }
441 ),
442 "transition_reconciliation_record": _object(
443 {
444 **HEADER,
445 "document_type": _text(const="transition_reconciliation_record"),
446 "transition": _object(
447 {
448 "from_setting_as_documented": _text(maximum=500),
449 "to_setting_as_documented": _text(maximum=500),
450 "handoff_date": _nullable(DATE),
451 "responsible_sender_role": ROLE,
452 "responsible_receiver_role": ROLE,
453 }
454 ),
455 "medication_reconciliation": _object(
456 {
457 "status": _text(
458 enum={
459 "pending_authorized_review",
460 "completed_by_authorized_clinician",
461 "not_applicable_by_authorized_clinician",
462 }
463 ),
464 "source_list_fact_ids": _array(
465 IDENTIFIER, maximum=200
466 ),
467 "destination_list_fact_ids": _array(
468 IDENTIFIER, maximum=200
469 ),
470 "discrepancy_status": _text(
471 enum={
472 "not_assessed",
473 "none_documented",
474 "routed_to_authorized_clinician",
475 "resolved_by_authorized_clinician",
476 }
477 ),
478 "completed_by_role": ROLE,
479 "completed_at": _nullable(DATETIME),
480 }
481 ),
482 "handoff_items": _array(HANDOFF_ITEM_SCHEMA, maximum=300),
483 "unresolved_items": _array(
484 UNRESOLVED_ITEM_SCHEMA, maximum=200
485 ),
486 }
487 ),
488 "intended_use_handoff_record": _object(
489 {
490 **HEADER,
491 "document_type": _text(const="intended_use_handoff_record"),
492 "intended_use": _object(
493 {
494 "purpose": _text(
495 const=(
496 "format_and_validate_clinician_supplied_plan_"
497 "documentation"
498 )
499 ),
500 "intended_users": _array(
501 _text(minimum=1, maximum=200), maximum=50
502 ),
503 "intended_setting": _text(maximum=500),
504 "output_role": _text(const="documentation_only"),
505 "prohibited_uses": _array(
506 _text(enum=PROHIBITED_USES),
507 minimum=len(PROHIBITED_USES),
508 maximum=len(PROHIBITED_USES),
509 ),
510 }
511 ),
512 "privacy_process": _object(
513 {
514 "data_classification": _text(
515 enum={
516 "synthetic",
517 "deidentified_qualified_review",
518 "real_patient_minimum_necessary",
519 }
520 ),
521 "local_authorization_confirmed": _boolean(),
522 "authorized_environment_reference": _text(
523 maximum=1_000
524 ),
525 "minimum_necessary_confirmed": _boolean(),
526 "no_external_tools_confirmed": _boolean(),
527 "no_prompt_log_example_copy_confirmed": _boolean(),
528 "privacy_review_status": _text(
529 enum={
530 "pending",
531 "not_applicable_synthetic",
532 "completed_by_qualified_reviewer",
533 }
534 ),
535 "privacy_reviewer_role": ROLE,
536 "reviewed_at": _nullable(DATETIME),
537 "retention_policy_reference": _text(maximum=1_000),
538 "data_disposition_status": _text(
539 enum={
540 "pending",
541 "retain_under_local_policy",
542 "delete_under_local_policy",
543 "completed_under_local_policy",
544 }
545 ),
546 }
547 ),
548 "local_governance": _object(
549 {
550 "institution_or_organization_reference": _text(
551 maximum=1_000
552 ),
553 "clinical_owner_role": ROLE,
554 "policy_references": _array(
555 _text(minimum=1, maximum=1_000), maximum=100
556 ),
557 "records_owner_role": ROLE,
558 "change_control_owner_role": ROLE,
559 }
560 ),
561 "emergency_routing": _object(
562 {
563 "statement": _text(const=EMERGENCY_STATEMENT),
564 "local_process_reference": _text(maximum=1_000),
565 "verified_by_role": ROLE,
566 }
567 ),
568 "reporting_routes": _object(
569 {
570 "local_patient_safety_route": _text(maximum=1_000),
571 "product_event_route": _text(maximum=1_000),
572 "privacy_incident_route": _text(maximum=1_000),
573 "responsible_role": ROLE,
574 }
575 ),
576 "handoff": _object(
577 {
578 "sender_role": ROLE,
579 "recipient_role": ROLE,
580 "sent_at": _nullable(DATETIME),
581 "acknowledgment_status": _text(
582 enum={"pending", "acknowledged"}
583 ),
584 "unresolved_items_routed": _boolean(),
585 }
586 ),
587 "clinician_signoff": _object(
588 {
589 "status": _text(enum={"pending", "signed"}),
590 "signer_role": ROLE,
591 "credential_authority_reference": _text(
592 maximum=1_000
593 ),
594 "signed_at": _nullable(DATETIME),
595 "attestation": _text(const=SIGNOFF_ATTESTATION),
596 }
597 ),
598 "release_gate": _object(
599 {
600 "status": _text(
601 enum={
602 "blocked",
603 "released_for_authorized_documentation_handoff",
604 }
605 ),
606 "released_by_role": ROLE,
607 "released_at": _nullable(DATETIME),
608 "blocker_codes": _array(
609 _text(
610 minimum=3,
611 maximum=96,
612 pattern=BLOCKER_RE.pattern,
613 ),
614 maximum=100,
615 ),
616 }
617 ),
618 }
619 ),
623def _reject_nonlocal_path(raw_path: str | Path) -> Path:
624 text = str(raw_path)
625 lowered = text.strip().lower()
626 if not lowered or "\x00" in text or "://" in lowered:
627 raise ValidationError("NONLOCAL_OR_EMPTY_PATH")
628 if lowered.startswith(("\\\\", "//")):
629 raise ValidationError("NETWORK_SHARE_PATH_REJECTED")
630 return Path(text).expanduser()
633def safe_input_file(raw_path: str | Path) -> Path:
634 """Resolve one bounded, regular, non-symlink local JSON file."""
636 path = _reject_nonlocal_path(raw_path)
637 if path.is_symlink():
638 raise ValidationError("SYMLINK_INPUT_REJECTED")
639 try:
640 resolved = path.resolve(strict=True)
641 except FileNotFoundError as exc:
642 raise ValidationError("INPUT_FILE_MISSING", path.name) from exc
643 if not resolved.is_file() or resolved.suffix.lower() != ".json":
644 raise ValidationError("INPUT_MUST_BE_JSON_FILE", path.name)
645 if resolved.stat().st_size > MAX_INPUT_BYTES:
646 raise ValidationError("INPUT_FILE_TOO_LARGE", path.name)
647 return resolved
650def safe_input_directory(raw_path: str | Path) -> Path:
651 """Resolve one local, non-symlink directory."""
653 path = _reject_nonlocal_path(raw_path)
654 if path.is_symlink():
655 raise ValidationError("SYMLINK_DIRECTORY_REJECTED")
656 try:
657 resolved = path.resolve(strict=True)
658 except FileNotFoundError as exc:
659 raise ValidationError("INPUT_DIRECTORY_MISSING") from exc
660 if not resolved.is_dir():
661 raise ValidationError("INPUT_MUST_BE_DIRECTORY")
662 return resolved
665def safe_new_directory(raw_path: str | Path) -> Path:
666 """Create a private directory without reusing an existing path."""
668 path = _reject_nonlocal_path(raw_path)
669 if path.exists() or path.is_symlink():
670 raise ValidationError("OUTPUT_DIRECTORY_EXISTS")
671 try:
672 parent = path.parent.resolve(strict=True)
673 except FileNotFoundError as exc:
674 raise ValidationError("OUTPUT_PARENT_MISSING") from exc
675 if not parent.is_dir():
676 raise ValidationError("OUTPUT_PARENT_NOT_DIRECTORY")
677 resolved = parent / path.name
678 resolved.mkdir(mode=0o700)
679 return resolved
682def safe_output_file(raw_path: str | Path) -> Path:
683 """Resolve a new local JSON output without implicit overwrite."""
685 path = _reject_nonlocal_path(raw_path)
686 if path.suffix.lower() != ".json":
687 raise ValidationError("OUTPUT_MUST_BE_JSON")
688 try:
689 parent = path.parent.resolve(strict=True)
690 except FileNotFoundError as exc:
691 raise ValidationError("OUTPUT_PARENT_MISSING") from exc
692 if not parent.is_dir():
693 raise ValidationError("OUTPUT_PARENT_NOT_DIRECTORY")
694 resolved = parent / path.name
695 if resolved.exists() or resolved.is_symlink():
696 raise ValidationError("OUTPUT_ALREADY_EXISTS", resolved.name)
697 return resolved
700def _duplicate_safe_object(
701 pairs: list[tuple[str, Any]],
702) -> dict[str, Any]:
703 result: dict[str, Any] = {}
704 for key, value in pairs:
705 if key in result:
706 raise ValidationError("JSON_DUPLICATE_KEY")
707 result[key] = value
708 return result
711def _reject_constant(_: str) -> None:
712 raise ValidationError("JSON_NONFINITE_NUMBER")
715def _check_bounds(value: Any, *, depth: int = 0, counter: list[int]) -> None:
716 if depth > MAX_DEPTH:
717 raise ValidationError("JSON_MAX_DEPTH_EXCEEDED")
718 counter[0] += 1
719 if counter[0] > MAX_TOTAL_NODES:
720 raise ValidationError("JSON_NODE_LIMIT_EXCEEDED")
721 if isinstance(value, dict):
722 if len(value) > MAX_OBJECT_FIELDS:
723 raise ValidationError("JSON_OBJECT_FIELD_LIMIT_EXCEEDED")
724 for key, nested in value.items():
725 if not isinstance(key, str) or len(key) > 128:
726 raise ValidationError("JSON_KEY_INVALID")
727 _check_bounds(nested, depth=depth + 1, counter=counter)
728 elif isinstance(value, list):
729 if len(value) > MAX_LIST_ITEMS:
730 raise ValidationError("JSON_LIST_LIMIT_EXCEEDED")
731 for nested in value:
732 _check_bounds(nested, depth=depth + 1, counter=counter)
733 elif isinstance(value, str):
734 if len(value) > MAX_TEXT_CHARS:
735 raise ValidationError("JSON_TEXT_LIMIT_EXCEEDED")
738def read_json(raw_path: str | Path) -> dict[str, Any]:
739 """Read strict UTF-8 JSON without exposing values in failures."""
741 path = safe_input_file(raw_path)
742 try:
743 with path.open("rb") as handle:
744 raw = handle.read(MAX_INPUT_BYTES + 1)
745 if len(raw) > MAX_INPUT_BYTES:
746 raise ValidationError("INPUT_FILE_TOO_LARGE", path.name)
747 text = raw.decode("utf-8")
748 except UnicodeDecodeError as exc:
749 raise ValidationError("JSON_NOT_UTF8", path.name) from exc
750 except OSError as exc:
751 raise ValidationError("INPUT_READ_FAILED", path.name) from exc
752 if "\x00" in text:
753 raise ValidationError("JSON_CONTAINS_NUL", path.name)
754 try:
755 value = json.loads(
756 text,
757 object_pairs_hook=_duplicate_safe_object,
758 parse_constant=_reject_constant,
759 )
760 except json.JSONDecodeError as exc:
761 raise ValidationError(
762 "JSON_PARSE_ERROR", f"{path.name}:{exc.lineno}:{exc.colno}"
763 ) from exc
764 except RecursionError as exc:
765 raise ValidationError("JSON_MAX_DEPTH_EXCEEDED") from exc
766 except ValidationError:
767 raise
768 except ValueError as exc:
769 raise ValidationError("JSON_NUMBER_INVALID") from exc
770 if not isinstance(value, dict):
771 raise ValidationError("JSON_ROOT_NOT_OBJECT", path.name)
772 _check_bounds(value, counter=[0])
773 return value
776def load_package(
777 raw_directory: str | Path,
778) -> tuple[dict[str, dict[str, Any]], dict[str, Path]]:
779 """Load the six required package records by fixed local filename."""
781 root = safe_input_directory(raw_directory)
782 documents: dict[str, dict[str, Any]] = {}
783 paths: dict[str, Path] = {}
784 for expected_type, (_, filename) in TEMPLATE_FILES.items():
785 path = root / filename
786 if not path.exists():
787 raise ValidationError("PACKAGE_FILE_MISSING", filename)
788 document = read_json(path)
789 documents[expected_type] = document
790 paths[expected_type] = path
791 return documents, paths
794def load_target(
795 raw_path: str | Path, *, require_package: bool = False
796) -> tuple[dict[str, dict[str, Any]], dict[str, Path]]:
797 """Load one record or a complete package."""
799 path = _reject_nonlocal_path(raw_path)
800 if path.is_dir() and not path.is_symlink():
801 return load_package(path)
802 if require_package:
803 raise ValidationError("PACKAGE_DIRECTORY_REQUIRED")
804 document = read_json(path)
805 document_type = document.get("document_type")
806 if not isinstance(document_type, str):
807 raise ValidationError("DOCUMENT_TYPE_MISSING")
808 return {document_type: document}, {document_type: safe_input_file(path)}
811def _valid_iso_date(value: str) -> bool:
812 try:
813 return date.fromisoformat(value).isoformat() == value
814 except ValueError:
815 return False
818def parse_iso_datetime(value: str) -> datetime | None:
819 """Parse an extended calendar datetime with seconds and an explicit offset."""
821 if not isinstance(value, str) or DATETIME_RE.fullmatch(value) is None:
822 return None
823 try:
824 parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
825 except ValueError:
826 return None
827 return parsed if parsed.tzinfo is not None else None
830def _validate_schema(
831 value: Any,
832 schema: dict[str, Any],
833 path: str,
834 issues: list[Issue],
835) -> None:
836 kind = schema["kind"]
837 if kind == "nullable":
838 if value is None:
839 return
840 _validate_schema(value, schema["schema"], path, issues)
841 return
842 if kind == "object":
843 if not isinstance(value, dict):
844 issues.append(Issue("SCHEMA_TYPE_OBJECT", path))
845 return
846 properties = schema["properties"]
847 for key in sorted(set(properties) - set(value)):
848 issues.append(Issue("SCHEMA_REQUIRED_FIELD", f"{path}.{key}"))
849 for _ in set(value) - set(properties):
850 # Unknown keys may themselves contain identifying or clinical text.
851 issues.append(Issue("SCHEMA_UNKNOWN_FIELD", f"{path}.[unknown_field]"))
852 for key in properties.keys() & value.keys():
853 _validate_schema(
854 value[key], properties[key], f"{path}.{key}", issues
855 )
856 return
857 if kind == "array":
858 if not isinstance(value, list):
859 issues.append(Issue("SCHEMA_TYPE_ARRAY", path))
860 return
861 if not schema["minimum"] <= len(value) <= schema["maximum"]:
862 issues.append(Issue("SCHEMA_ARRAY_BOUNDS", path))
863 for index, nested in enumerate(value):
864 _validate_schema(
865 nested, schema["item"], f"{path}[{index}]", issues
866 )
867 return
868 if kind == "boolean":
869 if type(value) is not bool:
870 issues.append(Issue("SCHEMA_TYPE_BOOLEAN", path))
871 return
872 if kind == "string":
873 if not isinstance(value, str):
874 issues.append(Issue("SCHEMA_TYPE_STRING", path))
875 return
876 if not schema["minimum"] <= len(value) <= schema["maximum"]:
877 issues.append(Issue("SCHEMA_STRING_BOUNDS", path))
878 if schema["minimum"] and not value.strip():
879 issues.append(Issue("SCHEMA_STRING_BLANK", path))
880 if schema["const"] is not None and value != schema["const"]:
881 issues.append(Issue("SCHEMA_CONST", path))
882 if schema["enum"] is not None and value not in schema["enum"]:
883 issues.append(Issue("SCHEMA_ENUM", path))
884 if schema["pattern"] is not None and not schema["pattern"].fullmatch(
885 value
886 ):
887 issues.append(Issue("SCHEMA_PATTERN", path))
888 if schema["format"] == "date" and not _valid_iso_date(value):
889 issues.append(Issue("SCHEMA_ISO_DATE", path))
890 if schema["format"] == "datetime" and parse_iso_datetime(value) is None:
891 issues.append(Issue("SCHEMA_ISO_DATETIME", path))
892 return
893 issues.append(Issue("SCHEMA_INTERNAL_KIND", path))
896def _duplicate_field_issues(
897 records: Any, field: str, path: str
898) -> list[Issue]:
899 if not isinstance(records, list):
900 return []
901 seen: set[str] = set()
902 issues: list[Issue] = []
903 for index, record in enumerate(records):
904 if not isinstance(record, dict):
905 continue
906 value = record.get(field)
907 if isinstance(value, str):
908 if value in seen:
909 issues.append(
910 Issue("DUPLICATE_RECORD_ID", f"{path}[{index}].{field}")
911 )
912 seen.add(value)
913 return issues
916def validate_document(
917 document: dict[str, Any], expected_type: str | None = None
918) -> list[Issue]:
919 """Validate strict structure without evaluating clinical content."""
921 document_type = document.get("document_type")
922 if not isinstance(document_type, str):
923 return [Issue("DOCUMENT_TYPE_MISSING", "$.document_type")]
924 if expected_type is not None and document_type != expected_type:
925 return [Issue("DOCUMENT_TYPE_FILENAME_MISMATCH", "$.document_type")]
926 schema = DOCUMENT_SCHEMAS.get(document_type)
927 if schema is None:
928 return [Issue("DOCUMENT_TYPE_UNKNOWN", "$.document_type")]
929 issues: list[Issue] = []
930 _validate_schema(document, schema, "$", issues)
932 if document_type == "source_fact_manifest":
933 issues.extend(
934 _duplicate_field_issues(document.get("facts"), "fact_id", "$.facts")
935 )
936 elif document_type == "clinician_authored_intervention_record":
937 interventions = document.get("interventions")
938 issues.extend(
939 _duplicate_field_issues(
940 interventions, "intervention_id", "$.interventions"
941 )
942 )
943 if isinstance(interventions, list):
944 for index, record in enumerate(interventions):
945 if isinstance(record, dict):
946 issues.extend(
947 _duplicate_field_issues(
948 record.get("parameters_as_supplied"),
949 "parameter_id",
950 f"$.interventions[{index}].parameters_as_supplied",
951 )
952 )
953 elif document_type == "goals_monitoring_checkpoint_record":
954 for field, identifier_field in (
955 ("goals", "goal_id"),
956 ("monitoring_items", "monitoring_id"),
957 ("checkpoints", "checkpoint_id"),
958 ):
959 issues.extend(
960 _duplicate_field_issues(
961 document.get(field), identifier_field, f"$.{field}"
962 )
963 )
964 elif document_type == "informed_preference_shared_decision_record":
965 entries = document.get("entries")
966 issues.extend(
967 _duplicate_field_issues(entries, "decision_id", "$.entries")
968 )
969 if isinstance(entries, list):
970 for index, record in enumerate(entries):
971 if isinstance(record, dict):
972 issues.extend(
973 _duplicate_field_issues(
974 record.get("options_as_documented"),
975 "option_id",
976 f"$.entries[{index}].options_as_documented",
977 )
978 )
979 elif document_type == "transition_reconciliation_record":
980 for field in ("handoff_items", "unresolved_items"):
981 issues.extend(
982 _duplicate_field_issues(
983 document.get(field), "item_id", f"$.{field}"
984 )
985 )
986 elif document_type == "intended_use_handoff_record":
987 intended = document.get("intended_use")
988 if isinstance(intended, dict):
989 uses = intended.get("prohibited_uses")
990 if isinstance(uses, list) and (
991 any(not isinstance(item, str) for item in uses)
992 or set(uses) != PROHIBITED_USES
993 ):
994 issues.append(
995 Issue(
996 "PROHIBITED_USE_SET_CHANGED",
997 "$.intended_use.prohibited_uses",
998 )
999 )
1000 return sorted(issues, key=lambda item: (item.path, item.code))
1003def validate_package_structure(
1004 documents: dict[str, dict[str, Any]]
1005) -> list[Issue]:
1006 """Validate all package documents and prefix paths with document types."""
1008 issues: list[Issue] = []
1009 for expected_type in TEMPLATE_FILES:
1010 document = documents.get(expected_type)
1011 if document is None:
1012 issues.append(Issue("PACKAGE_DOCUMENT_MISSING", expected_type))
1013 continue
1014 for issue in validate_document(document, expected_type):
1015 issues.append(
1016 Issue(
1017 issue.code,
1018 f"{expected_type}:{issue.path}",
1019 issue.level,
1022 return sorted(issues, key=lambda item: (item.path, item.code))
1025def has_errors(issues: Iterable[Issue]) -> bool:
1026 return any(issue.level == "error" for issue in issues)
1029def report_payload(
1030 check: str,
1031 issues: Iterable[Issue],
1032 *,
1033 counts: dict[str, int] | None = None,
1034 extra: dict[str, Any] | None = None,
1035) -> dict[str, Any]:
1036 issue_list = list(issues)
1037 payload: dict[str, Any] = {
1038 "check": check,
1039 "status": "fail" if has_errors(issue_list) else "pass",
1040 "issues": [item.as_dict() for item in issue_list],
1041 "counts": counts or {},
1042 "notice": NOTICE,
1043 "determinations_not_made": [
1044 "clinical_correctness",
1045 "clinical_safety",
1046 "diagnosis",
1047 "therapy_or_medication_selection",
1048 "dose_or_schedule",
1049 "interaction_or_contraindication",
1050 "triage_or_emergency_status",
1051 "prognosis",
1052 "privacy_or_legal_compliance",
1053 ],
1055 if extra:
1056 payload.update(extra)
1057 return payload
1060def error_report(check: str, error: ValidationError) -> dict[str, Any]:
1061 return report_payload(check, [Issue(error.code, error.path)])
1064def print_report(report: dict[str, Any]) -> None:
1065 print(json.dumps(report, indent=2, sort_keys=True, ensure_ascii=False))
1068def atomic_write_json(destination: Path, payload: dict[str, Any]) -> Path:
1069 """Write private JSON atomically without implicit overwrite."""
1071 if destination.exists() or destination.is_symlink():
1072 raise ValidationError("OUTPUT_ALREADY_EXISTS", destination.name)
1073 temporary_name: str | None = None
1074 try:
1075 with tempfile.NamedTemporaryFile(
1076 "w",
1077 encoding="utf-8",
1078 dir=destination.parent,
1079 prefix=f".{destination.name}.",
1080 suffix=".tmp",
1081 delete=False,
1082 ) as handle:
1083 temporary_name = handle.name
1084 json.dump(
1085 payload,
1086 handle,
1087 indent=2,
1088 sort_keys=True,
1089 ensure_ascii=False,
1091 handle.write("\n")
1092 handle.flush()
1093 os.fsync(handle.fileno())
1094 os.chmod(temporary_name, 0o600)
1095 # Publish a completed file without replacing a path created by another
1096 # writer after the initial existence check. Fail on filesystems without
1097 # hard-link support rather than fall back to a clobbering rename.
1098 os.link(temporary_name, destination)
1099 Path(temporary_name).unlink()
1100 return destination
1101 except FileExistsError as exc:
1102 raise ValidationError("OUTPUT_ALREADY_EXISTS", destination.name) from exc
1103 except OSError as exc:
1104 raise ValidationError("OUTPUT_WRITE_FAILED", destination.name) from exc
1105 finally:
1106 if temporary_name:
1107 try:
1108 Path(temporary_name).unlink(missing_ok=True)
1109 except OSError:
1110 pass
1113def copy_template_document(
1114 template_path: Path,
1115 *,
1116 subject_ref: str,
1117 classification: str,
1118) -> dict[str, Any]:
1119 document = read_json(template_path)
1120 document["subject_ref"] = subject_ref
1121 document["data_classification"] = classification
1122 if document.get("document_type") == "intended_use_handoff_record":
1123 privacy = document.get("privacy_process")
1124 if isinstance(privacy, dict):
1125 privacy["data_classification"] = classification
1126 return document
1129def direct_identifier_key_paths(
1130 value: Any, path: str = "$"
1131) -> list[str]:
1132 """Return paths for obvious direct-identifier field names, not values."""
1134 matches: list[str] = []
1135 if isinstance(value, dict):
1136 for key, nested in value.items():
1137 normalized = str(key).strip().lower()
1138 nested_path = f"{path}.{key}"
1139 if normalized in DIRECT_IDENTIFIER_KEYS:
1140 matches.append(nested_path)
1141 matches.extend(direct_identifier_key_paths(nested, nested_path))
1142 elif isinstance(value, list):
1143 for index, nested in enumerate(value):
1144 matches.extend(
1145 direct_identifier_key_paths(nested, f"{path}[{index}]")
1147 return matches
1150def all_fact_references(
1151 documents: dict[str, dict[str, Any]]
1152) -> list[tuple[str, str]]:
1153 """Collect every source-fact reference with a nonclinical field path."""
1155 references: list[tuple[str, str]] = []
1157 def collect(value: Any, path: str) -> None:
1158 if isinstance(value, dict):
1159 for key, nested in value.items():
1160 nested_path = f"{path}.{key}"
1161 if key.endswith("fact_ids") and isinstance(nested, list):
1162 for index, identifier in enumerate(nested):
1163 if isinstance(identifier, str):
1164 references.append(
1165 (f"{nested_path}[{index}]", identifier)
1167 else:
1168 collect(nested, nested_path)
1169 elif isinstance(value, list):
1170 for index, nested in enumerate(value):
1171 collect(nested, f"{path}[{index}]")
1173 for document_type, document in documents.items():
1174 if document_type != "source_fact_manifest":
1175 collect(document, document_type)
1176 return references
1179def record_ids(
1180 document: dict[str, Any],
1181 list_field: str,
1182 id_field: str,
1183) -> set[str]:
1184 records = document.get(list_field)
1185 if not isinstance(records, list):
1186 return set()
1187 return {
1188 record[id_field]
1189 for record in records
1190 if isinstance(record, dict) and isinstance(record.get(id_field), str)