#!/usr/bin/env python3 """Fail-closed validator for the C-001 external clinical review contract. The validator proves approval provenance and internal consistency. It does not make a clinical judgment and it never converts technical checks into clinical approval. """ from __future__ import annotations import argparse import hashlib import json import sys from datetime import date from pathlib import Path, PurePosixPath from typing import Any try: from jsonschema import Draft202012Validator, FormatChecker from jsonschema.exceptions import SchemaError except ModuleNotFoundError: # Runtime images need no dev-only jsonschema package. Draft202012Validator = None FormatChecker = None SchemaError = Exception REPO_ROOT = Path(__file__).resolve().parents[1] MANIFEST_REL = Path("data/clinical/crisis-protocol-validation.json") MANIFEST_SCHEMA_REL = Path("data/clinical/crisis-protocol-validation.schema.json") CASE_SET_REL = Path("data/clinical/p1-crisis-review-cases.json") CASE_SCHEMA_REL = Path("data/clinical/p1-crisis-review-cases.schema.json") ALLOWED_STATUSES = { "pending_external_review", "approved", "conditional", "rejected", } STATUS_DECISIONS = { "pending_external_review": None, "approved": "approved", "conditional": "conditional", "rejected": "rejected", } APPROVAL_FIELDS = { "reviewer", "organization", "reviewer_role_or_qualification", "reviewed_at", "decision", "notes", "evidence_ref", "evidence_sha256", "reviewed_protocol_version", "reviewed_case_set_sha256", } MANIFEST_FIELDS = { "schema_version", "protocol_id", "version", "scope", "technical_status", "clinical_status", "external_review_boundary", "review_case_set", "review_case_set_id", "review_case_set_version", "sources", "review_sequence", "technical_gates", "approval", } CASE_SET_FIELDS = { "schema_version", "case_set_id", "version", "protocol_id", "protocol_version", "priority", "scope", "external_review_boundary", "content_safety", "official_source_scope", "case_decision_contract", "youth_review_responses", "cases", } CASE_FIELDS = { "case_id", "title", "synthetic_scenario", "technical_invariants", "reviewer_assessment", } SCENARIO_FIELDS = {"speaker_context", "signal", "method_or_means_detail_present"} INVARIANT_FIELDS = {"invariant_id", "requirement"} ASSESSMENT_FIELDS = { "decision", "rationale", "findings_and_required_changes", "re_review", "reviewed_at", } REQUIRED_COMPLETED_APPROVAL_FIELDS = { "reviewer", "organization", "reviewer_role_or_qualification", "reviewed_at", "decision", "notes", "evidence_ref", "evidence_sha256", "reviewed_protocol_version", "reviewed_case_set_sha256", } ALLOWED_CASE_DECISIONS = {"pass", "conditional", "fail"} EXPECTED_CASE_IDS = {f"P1-CRISIS-{index:03d}" for index in range(1, 7)} NO_REQUIRED_CHANGE_VALUES = { "없음", "수정 요구 없음", "해당 없음", "none", "n/a", "na", } NO_RE_REVIEW_VALUES = { "없음", "재검토 불필요", "해당 없음", "none", "n/a", "na", } YOUTH_REVIEW_RESPONSE_FIELDS = { "guardian_notification", "means_restriction", "risk_disposition", "follow_up_48_72h", "documentation", "safeguarding", } REQUIRED_SOURCE_IDS = { "samhsa_safe_t", "nimh_youth_outpatient_bssa", "mohw_109", "nice_ng225", } REQUIRED_SOURCE_PROVENANCE = { "samhsa_safe_t": ( "SAMHSA", "SAFE-T Suicide Assessment Five-Step Evaluation and Triage", "https://www.samhsa.gov/resource/dbhis/safe-t-pocket-card-suicide-assessment-five-step-evaluation-triage-safe-t-clinicians", ), "nimh_youth_outpatient_bssa": ( "NIMH", "Youth Outpatient Brief Suicide Safety Assessment Guide", "https://www.nimh.nih.gov/research/research-conducted-at-nimh/asq-toolkit-materials/youth-outpatient/youth-outpatient-brief-suicide-safety-assessment-guide", ), "mohw_109": ( "대한민국 보건복지부", "자살예방상담전화 109", "https://www.mohw.go.kr/menu.es?mid=a10716040000", ), "nice_ng225": ( "NICE", "NG225 Self-harm: assessment, management and preventing recurrence", "https://www.nice.org.uk/guidance/ng225", ), } MANIFEST_BOUNDARY = ( "공식 근거와 자동 검사는 기술 안전 범위만 확인하며 외부 임상 검토와 승인을 대신하지 않는다." ) CASE_SET_BOUNDARY = ( "이 사례에는 임상 정답이 없으며 자동 검사는 외부 임상 검토와 승인을 대신하지 않는다." ) SHA256_LENGTH = 64 def _path(root: Path, value: Path | str) -> Path: candidate = Path(value) return candidate if candidate.is_absolute() else root / candidate def _load_json(path: Path, label: str, errors: list[str]) -> Any | None: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: errors.append(f"{label}: 파일이 없다: {path}") except json.JSONDecodeError as exc: errors.append(f"{label}: JSON 파싱 실패: {exc}") except OSError as exc: errors.append(f"{label}: 읽기 실패: {exc}") return None def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _is_nonempty_string(value: Any) -> bool: return isinstance(value, str) and bool(value.strip()) def _is_sha256(value: Any) -> bool: if not isinstance(value, str) or len(value) != SHA256_LENGTH: return False return all(character in "0123456789abcdef" for character in value) def _is_iso_date(value: Any) -> bool: if not isinstance(value, str): return False try: return date.fromisoformat(value).isoformat() == value except ValueError: return False def _schema_errors( instance: Any, schema: Any, label: str, ) -> list[str]: if not isinstance(schema, dict): return [f"{label} schema: 최상위 값은 객체여야 한다"] if Draft202012Validator is None or FormatChecker is None: return [] try: Draft202012Validator.check_schema(schema) except SchemaError as exc: return [f"{label} schema: 스키마 자체가 유효하지 않다: {exc.message}"] validator = Draft202012Validator(schema, format_checker=FormatChecker()) failures: list[str] = [] for error in sorted( validator.iter_errors(instance), key=lambda item: tuple(str(part) for part in item.absolute_path), ): location = ".".join(str(part) for part in error.absolute_path) or "$" failures.append(f"{label} schema: {location}: {error.message}") return failures def _validate_sources(manifest: dict[str, Any], case_set: dict[str, Any]) -> list[str]: errors: list[str] = [] sources = manifest.get("sources") if not isinstance(sources, list): errors.append("manifest.sources: 배열이어야 한다") source_ids: set[Any] = set() else: source_ids = { source.get("source_id") for source in sources if isinstance(source, dict) } if len(source_ids) != len(sources): errors.append("manifest.sources: source_id가 없거나 중복됐다") if source_ids != REQUIRED_SOURCE_IDS: errors.append( "manifest.sources: 공식 범위는 SAMHSA SAFE-T, NIMH Youth Outpatient " "BSSA, 보건복지부 109, NICE NG225 네 항목과 정확히 일치해야 한다" ) if isinstance(sources, list): for source in sources: if not isinstance(source, dict): continue source_id = source.get("source_id") expected = REQUIRED_SOURCE_PROVENANCE.get(source_id) if expected is None: continue actual = (source.get("authority"), source.get("title"), source.get("url")) if actual != expected: errors.append(f"manifest.sources[{source_id}]: 공식 authority/title/url과 다르다") case_sources = case_set.get("official_source_scope") if not isinstance(case_sources, list) or set(case_sources) != REQUIRED_SOURCE_IDS: errors.append("case_set.official_source_scope: manifest의 공식 근거 네 항목과 일치해야 한다") return errors def _validate_case_set_base( manifest: dict[str, Any], case_set: dict[str, Any], ) -> list[str]: errors: list[str] = [] if set(manifest) != MANIFEST_FIELDS: missing = sorted(MANIFEST_FIELDS - set(manifest)) extra = sorted(set(manifest) - MANIFEST_FIELDS) errors.append(f"manifest: 필드 불일치 missing={missing}, extra={extra}") if set(case_set) != CASE_SET_FIELDS: missing = sorted(CASE_SET_FIELDS - set(case_set)) extra = sorted(set(case_set) - CASE_SET_FIELDS) errors.append(f"case_set: 필드 불일치 missing={missing}, extra={extra}") if manifest.get("schema_version") != "vignette.clinical_crisis_review.v2": errors.append("manifest.schema_version: vignette.clinical_crisis_review.v2여야 한다") if case_set.get("schema_version") != "vignette.p1_crisis_review_cases.v2": errors.append("case_set.schema_version: vignette.p1_crisis_review_cases.v2여야 한다") if manifest.get("technical_status") != "verified": errors.append("manifest.technical_status: verified여야 한다") if case_set.get("priority") != "P1": errors.append("case_set.priority: P1이어야 한다") if manifest.get("external_review_boundary") != MANIFEST_BOUNDARY: errors.append("manifest.external_review_boundary: 외부 임상 승인 비대체 경계를 변경할 수 없다") if case_set.get("external_review_boundary") != CASE_SET_BOUNDARY: errors.append("case_set.external_review_boundary: 임상 정답 및 외부 승인 비대체 경계를 변경할 수 없다") safety = case_set.get("content_safety") expected_safety = { "synthetic_only": True, "method_or_means_detail": "forbidden", "clinical_answer_included": False, } if safety != expected_safety: errors.append("case_set.content_safety: 합성 전용·방법 상세 금지·임상 정답 미포함 계약과 다르다") comparisons = ( (case_set.get("protocol_id"), manifest.get("protocol_id"), "protocol_id"), (case_set.get("protocol_version"), manifest.get("version"), "protocol_version"), (case_set.get("case_set_id"), manifest.get("review_case_set_id"), "case_set_id"), (case_set.get("version"), manifest.get("review_case_set_version"), "case_set_version"), ) for actual, expected, label in comparisons: if actual != expected: errors.append(f"case_set.{label}: manifest와 일치하지 않는다") contract = case_set.get("case_decision_contract") if not isinstance(contract, dict): errors.append("case_set.case_decision_contract: 객체여야 한다") else: expected_contract_fields = { "allowed", "pending_value", "approved_rule", "conditional_rule", "rejected_rule", } if set(contract) != expected_contract_fields: errors.append("case_set.case_decision_contract: 필드 계약과 정확히 일치해야 한다") if contract.get("allowed") != ["pass", "conditional", "fail"]: errors.append("case_set.case_decision_contract.allowed: pass/conditional/fail 순서와 일치해야 한다") if contract.get("pending_value", object()) is not None: errors.append("case_set.case_decision_contract.pending_value: null이어야 한다") youth_responses = case_set.get("youth_review_responses") if not isinstance(youth_responses, dict): errors.append("case_set.youth_review_responses: 객체여야 한다") elif set(youth_responses) != YOUTH_REVIEW_RESPONSE_FIELDS: missing = sorted(YOUTH_REVIEW_RESPONSE_FIELDS - set(youth_responses)) extra = sorted(set(youth_responses) - YOUTH_REVIEW_RESPONSE_FIELDS) errors.append( "case_set.youth_review_responses: 필드 불일치 " f"missing={missing}, extra={extra}" ) cases = case_set.get("cases") if not isinstance(cases, list) or not cases: errors.append("case_set.cases: 하나 이상의 합성 사례가 필요하다") return errors case_ids = { case.get("case_id") for case in cases if isinstance(case, dict) } if len(case_ids) != len(cases): errors.append("case_set.cases: case_id가 없거나 중복됐다") if case_ids != EXPECTED_CASE_IDS: errors.append("case_set.cases: P1-CRISIS-001..006을 정확히 포함해야 한다") for index, case in enumerate(cases): label = case.get("case_id", f"cases[{index}]") if isinstance(case, dict) else f"cases[{index}]" if not isinstance(case, dict): errors.append(f"{label}: 객체여야 한다") continue if set(case) != CASE_FIELDS: errors.append(f"{label}: 사례 필드 계약과 정확히 일치해야 한다") scenario = case.get("synthetic_scenario") if not isinstance(scenario, dict): errors.append(f"{label}.synthetic_scenario: 객체여야 한다") else: if set(scenario) != SCENARIO_FIELDS: errors.append(f"{label}.synthetic_scenario: 시나리오 필드 계약과 정확히 일치해야 한다") if scenario.get("method_or_means_detail_present") is not False: errors.append(f"{label}.synthetic_scenario: 수단·방법 상세는 절대 포함할 수 없다") invariants = case.get("technical_invariants") if not isinstance(invariants, list) or not invariants: errors.append(f"{label}.technical_invariants: 하나 이상의 기술 불변조건이 필요하다") else: invariant_ids = { invariant.get("invariant_id") for invariant in invariants if isinstance(invariant, dict) } if len(invariant_ids) != len(invariants): errors.append(f"{label}.technical_invariants: invariant_id가 없거나 중복됐다") for invariant in invariants: if not isinstance(invariant, dict) or set(invariant) != INVARIANT_FIELDS: errors.append(f"{label}.technical_invariants: 불변조건 필드 계약과 정확히 일치해야 한다") break assessment = case.get("reviewer_assessment") if not isinstance(assessment, dict) or set(assessment) != ASSESSMENT_FIELDS: errors.append(f"{label}.reviewer_assessment: 판정 필드 계약과 정확히 일치해야 한다") return errors def _validate_pending( approval: dict[str, Any], cases: list[Any], youth_review_responses: Any, ) -> list[str]: errors: list[str] = [] for field in sorted(APPROVAL_FIELDS): if approval.get(field, object()) is not None: errors.append(f"pending_external_review: approval.{field}는 null이어야 한다") for index, case in enumerate(cases): if not isinstance(case, dict): continue label = case.get("case_id", f"cases[{index}]") assessment = case.get("reviewer_assessment") if not isinstance(assessment, dict): errors.append(f"pending_external_review: {label}.reviewer_assessment가 필요하다") continue for field in sorted(ASSESSMENT_FIELDS): if assessment.get(field, object()) is not None: errors.append( f"pending_external_review: {label}.reviewer_assessment.{field}는 null이어야 한다" ) if not isinstance(youth_review_responses, dict): errors.append("pending_external_review: youth_review_responses 객체가 필요하다") else: for field in sorted(YOUTH_REVIEW_RESPONSE_FIELDS): if youth_review_responses.get(field, object()) is not None: errors.append( "pending_external_review: " f"youth_review_responses.{field}는 null이어야 한다" ) return errors def _validate_evidence( root: Path, approval: dict[str, Any], ) -> list[str]: errors: list[str] = [] evidence_ref = approval.get("evidence_ref") if not _is_nonempty_string(evidence_ref): return ["approval.evidence_ref: 비어 있지 않은 저장소 상대 경로가 필요하다"] if "\\" in evidence_ref: errors.append("approval.evidence_ref: 플랫폼 독립적인 / 구분자를 사용해야 한다") return errors pure = PurePosixPath(evidence_ref) if pure.is_absolute() or ".." in pure.parts: errors.append("approval.evidence_ref: 절대 경로와 상위 경로 이동은 허용하지 않는다") return errors if pure.parts[:3] != ("data", "clinical", "evidence") or len(pure.parts) < 4: errors.append("approval.evidence_ref: data/clinical/evidence/ 아래 파일이어야 한다") return errors root_resolved = root.resolve() evidence_root = (root_resolved / "data/clinical/evidence").resolve() evidence_path = (root_resolved / Path(*pure.parts)).resolve() try: evidence_path.relative_to(evidence_root) except ValueError: errors.append("approval.evidence_ref: evidence 디렉터리 밖을 가리킨다") return errors if not evidence_path.is_file(): errors.append(f"approval.evidence_ref: 증거 파일이 없다: {evidence_ref}") return errors declared_hash = approval.get("evidence_sha256") if not _is_sha256(declared_hash): errors.append("approval.evidence_sha256: 소문자 64자리 SHA-256이어야 한다") elif _sha256(evidence_path) != declared_hash: errors.append("approval.evidence_sha256: 실제 증거 파일 해시와 일치하지 않는다") return errors def _validate_completed( root: Path, status: str, manifest: dict[str, Any], approval: dict[str, Any], case_set_path: Path, cases: list[Any], youth_review_responses: Any, ) -> list[str]: errors: list[str] = [] for field in sorted(REQUIRED_COMPLETED_APPROVAL_FIELDS): if not _is_nonempty_string(approval.get(field)): errors.append(f"{status}: approval.{field} 값이 필요하다") reviewed_at = approval.get("reviewed_at") if not _is_iso_date(reviewed_at): errors.append(f"{status}: approval.reviewed_at은 YYYY-MM-DD 실제 날짜여야 한다") if approval.get("decision") != STATUS_DECISIONS[status]: errors.append(f"{status}: approval.decision은 {STATUS_DECISIONS[status]}이어야 한다") if approval.get("reviewed_protocol_version") != manifest.get("version"): errors.append(f"{status}: approval.reviewed_protocol_version이 현재 protocol version과 다르다") case_hash = approval.get("reviewed_case_set_sha256") if not _is_sha256(case_hash): errors.append(f"{status}: approval.reviewed_case_set_sha256은 소문자 64자리 SHA-256이어야 한다") elif _sha256(case_set_path) != case_hash: errors.append(f"{status}: approval.reviewed_case_set_sha256이 실제 사례 세트 해시와 다르다") errors.extend(_validate_evidence(root, approval)) if not isinstance(youth_review_responses, dict): errors.append(f"{status}: youth_review_responses 객체가 필요하다") else: for field in sorted(YOUTH_REVIEW_RESPONSE_FIELDS): if not _is_nonempty_string(youth_review_responses.get(field)): errors.append( f"{status}: youth_review_responses.{field} 답변이 필요하다" ) decisions: list[str] = [] for index, case in enumerate(cases): if not isinstance(case, dict): continue label = case.get("case_id", f"cases[{index}]") assessment = case.get("reviewer_assessment") if not isinstance(assessment, dict): errors.append(f"{status}: {label}.reviewer_assessment가 필요하다") continue decision = assessment.get("decision") if decision not in ALLOWED_CASE_DECISIONS: errors.append(f"{status}: {label}.reviewer_assessment.decision 판정이 필요하다") continue decisions.append(decision) if not _is_nonempty_string(assessment.get("rationale")): errors.append(f"{status}: {label}.reviewer_assessment.rationale이 필요하다") findings = assessment.get("findings_and_required_changes") re_review = assessment.get("re_review") if not _is_nonempty_string(findings): errors.append( f"{status}: {label}.reviewer_assessment.findings_and_required_changes가 필요하다" ) if not _is_nonempty_string(re_review): errors.append(f"{status}: {label}.reviewer_assessment.re_review가 필요하다") if decision in {"conditional", "fail"}: if isinstance(findings, str) and findings.strip().casefold() in NO_REQUIRED_CHANGE_VALUES: errors.append( f"{status}: {label}.reviewer_assessment.findings_and_required_changes에 " "실제 수정 요구를 기록해야 한다" ) if isinstance(re_review, str) and re_review.strip().casefold() in NO_RE_REVIEW_VALUES: errors.append( f"{status}: {label}.reviewer_assessment.re_review에 " "재검토 제출 조건 또는 일정을 기록해야 한다" ) if assessment.get("reviewed_at") != reviewed_at: errors.append(f"{status}: {label}.reviewer_assessment.reviewed_at이 전체 검토일과 달라서는 안 된다") if len(decisions) == len(cases): if status == "approved" and any(decision != "pass" for decision in decisions): errors.append("approved: 모든 사례 판정이 pass여야 한다") elif status == "conditional": if "fail" in decisions: errors.append("conditional: fail 사례가 있으면 rejected여야 한다") if "conditional" not in decisions: errors.append("conditional: 하나 이상의 conditional 사례가 필요하다") elif status == "rejected" and "fail" not in decisions: errors.append("rejected: 하나 이상의 fail 사례가 필요하다") return errors def validate_review_contract( *, repo_root: Path = REPO_ROOT, manifest_path: Path | str = MANIFEST_REL, manifest_schema_path: Path | str = MANIFEST_SCHEMA_REL, case_set_path: Path | str = CASE_SET_REL, case_schema_path: Path | str = CASE_SCHEMA_REL, ) -> list[str]: """Return every contract violation; an empty list means the gate is green.""" root = repo_root.resolve() manifest_file = _path(root, manifest_path) manifest_schema_file = _path(root, manifest_schema_path) case_set_file = _path(root, case_set_path) case_schema_file = _path(root, case_schema_path) errors: list[str] = [] manifest = _load_json(manifest_file, "manifest", errors) manifest_schema = _load_json(manifest_schema_file, "manifest schema", errors) case_set = _load_json(case_set_file, "case set", errors) case_schema = _load_json(case_schema_file, "case schema", errors) if any(value is None for value in (manifest, manifest_schema, case_set, case_schema)): return errors if not isinstance(manifest, dict) or not isinstance(case_set, dict): errors.append("manifest와 case set의 최상위 값은 객체여야 한다") return errors errors.extend(_schema_errors(manifest, manifest_schema, "manifest")) errors.extend(_schema_errors(case_set, case_schema, "case set")) errors.extend(_validate_sources(manifest, case_set)) errors.extend(_validate_case_set_base(manifest, case_set)) expected_case_ref = CASE_SET_REL.as_posix() if manifest.get("review_case_set") != expected_case_ref: errors.append(f"manifest.review_case_set: {expected_case_ref}여야 한다") status = manifest.get("clinical_status") if status not in ALLOWED_STATUSES: errors.append(f"manifest.clinical_status: 허용되지 않은 상태: {status!r}") return errors approval = manifest.get("approval") if not isinstance(approval, dict): errors.append("manifest.approval: 객체여야 한다") return errors if set(approval) != APPROVAL_FIELDS: missing = sorted(APPROVAL_FIELDS - set(approval)) extra = sorted(set(approval) - APPROVAL_FIELDS) errors.append(f"manifest.approval: 필드 불일치 missing={missing}, extra={extra}") if approval.get("decision") != STATUS_DECISIONS[status]: errors.append( f"manifest: clinical_status={status}와 approval.decision={approval.get('decision')!r}가 일치하지 않는다" ) cases = case_set.get("cases") if not isinstance(cases, list): return errors youth_review_responses = case_set.get("youth_review_responses") if status == "pending_external_review": errors.extend(_validate_pending(approval, cases, youth_review_responses)) else: errors.extend( _validate_completed( root, status, manifest, approval, case_set_file, cases, youth_review_responses, ) ) return errors def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="C-001 외부 임상 검토 상태와 증거를 fail-closed 검증한다.", ) parser.add_argument("--repo-root", type=Path, default=REPO_ROOT) parser.add_argument("--manifest", type=Path, default=MANIFEST_REL) parser.add_argument("--manifest-schema", type=Path, default=MANIFEST_SCHEMA_REL) parser.add_argument("--case-set", type=Path, default=CASE_SET_REL) parser.add_argument("--case-schema", type=Path, default=CASE_SCHEMA_REL) return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) errors = validate_review_contract( repo_root=args.repo_root, manifest_path=args.manifest, manifest_schema_path=args.manifest_schema, case_set_path=args.case_set, case_schema_path=args.case_schema, ) if errors: print("C-001 임상 검토 계약: FAIL", file=sys.stderr) for error in errors: print(f"- {error}", file=sys.stderr) return 1 manifest = json.loads(_path(args.repo_root.resolve(), args.manifest).read_text(encoding="utf-8")) case_set = json.loads(_path(args.repo_root.resolve(), args.case_set).read_text(encoding="utf-8")) print( json.dumps( { "ok": True, "clinical_status": manifest["clinical_status"], "protocol": f"{manifest['protocol_id']}@{manifest['version']}", "case_set": f"{case_set['case_set_id']}@{case_set['version']}", "case_count": len(case_set["cases"]), "review_complete": manifest["clinical_status"] != "pending_external_review", "external_review_replaced": False, }, ensure_ascii=False, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())