"""Strict client reply quality gate regressions.""" from __future__ import annotations import importlib.util import json import unittest from .paths import repo_path from .services import guardrail def _load_clinical_review_checker(): checker_path = repo_path("scripts", "check-clinical-crisis-review.py") spec = importlib.util.spec_from_file_location("clinical_crisis_review_checker", checker_path) if spec is None or spec.loader is None: raise RuntimeError(f"임상 검토 checker를 불러올 수 없다: {checker_path}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module class ClientReplyQualityGateTest(unittest.TestCase): def test_crisis_protocol_manifest_obeys_status_specific_external_review_contract(self) -> None: manifest = json.loads( repo_path("data", "clinical", "crisis-protocol-validation.json").read_text( encoding="utf-8", ), ) case_set = json.loads( repo_path("data", "clinical", "p1-crisis-review-cases.json").read_text( encoding="utf-8", ), ) self.assertEqual(manifest["technical_status"], "verified") status = manifest["clinical_status"] status_decisions = { "pending_external_review": None, "approved": "approved", "conditional": "conditional", "rejected": "rejected", } self.assertIn(status, status_decisions) required_approval_fields = { "reviewer", "organization", "reviewed_at", "decision", "notes", "evidence_ref", "evidence_sha256", "reviewed_protocol_version", "reviewed_case_set_sha256", } approval = manifest["approval"] self.assertEqual(set(approval), required_approval_fields) self.assertEqual(approval["decision"], status_decisions[status]) assessments = [case["reviewer_assessment"] for case in case_set["cases"]] if status == "pending_external_review": self.assertTrue(all(value is None for value in approval.values())) self.assertTrue( all(value is None for assessment in assessments for value in assessment.values()), ) else: completed_fields = required_approval_fields - {"notes"} self.assertTrue(all(approval[field] for field in completed_fields)) self.assertTrue(all(assessment["decision"] for assessment in assessments)) self.assertTrue(all(assessment["rationale"] for assessment in assessments)) self.assertTrue(all(assessment["reviewed_at"] for assessment in assessments)) source_ids = {source["source_id"] for source in manifest["sources"]} self.assertEqual( source_ids, {"samhsa_safe_t", "nimh_youth_outpatient_bssa", "mohw_109", "nice_ng225"}, ) self.assertGreaterEqual(len(manifest["review_sequence"]), 6) self.assertGreaterEqual(len(manifest["technical_gates"]), 6) self.assertIn("외부 임상 검토와 승인을 대신하지 않는다", manifest["external_review_boundary"]) checker = _load_clinical_review_checker() self.assertEqual(checker.validate_review_contract(repo_root=repo_path()), []) def test_synthetic_crisis_review_cases_contain_only_technical_invariants(self) -> None: case_set = json.loads( repo_path("data", "clinical", "p1-crisis-review-cases.json").read_text( encoding="utf-8", ), ) self.assertEqual( case_set["content_safety"], { "synthetic_only": True, "method_or_means_detail": "forbidden", "clinical_answer_included": False, }, ) self.assertEqual( {case["case_id"] for case in case_set["cases"]}, {f"P1-CRISIS-{index:03d}" for index in range(1, 7)}, ) for case in case_set["cases"]: with self.subTest(case_id=case["case_id"]): self.assertFalse(case["synthetic_scenario"]["method_or_means_detail_present"]) self.assertGreaterEqual(len(case["technical_invariants"]), 1) self.assertNotIn("clinical_answer", case) def test_crisis_protocol_technical_gates(self) -> None: counselor_inquiry = guardrail.classify_crisis( "최근에 자살 생각이 든 적이 있나요?", speaker_is_persona_context=True, ) learner_crisis = guardrail.classify_crisis( "저 지금 자살하고 싶어요. 도와주세요.", speaker_is_persona_context=True, ) client_signal = guardrail.sanitize_client_reply( "가끔 사라지고 싶다는 생각이 들지만, 아직 누구에게도 말하지 못했어요.", ideation_stage=2, turn_seq=4, ) means_detail = guardrail.sanitize_client_reply( "면도날을 항상 가지고 다녀요.", ideation_stage=3, turn_seq=4, ) over_cap = guardrail.sanitize_client_reply( "요즘 버티기 어렵다는 생각이 자주 들어요.", ideation_stage=4, turn_seq=4, ) self.assertEqual(counselor_inquiry.kind, guardrail.CrisisKind.PERSONA_PLAY) self.assertFalse(counselor_inquiry.escalate) self.assertEqual(learner_crisis.kind, guardrail.CrisisKind.LEARNER_REAL) self.assertTrue(learner_crisis.escalate) self.assertEqual(guardrail.crisis_resource()["number"], "109") self.assertFalse(client_signal.needs_regeneration) self.assertTrue(means_detail.needs_regeneration) self.assertTrue(over_cap.blocked) def test_blocks_self_harm_means_detail(self) -> None: """자해 수단·방법 상세는 차단(1.1 확장 — DBT 자해 페르소나 대비).""" samples = [ "커터로 했어요. 그게 제일 쉬워서요.", "면도날을 항상 가지고 다녀요.", "어젯밤에도 손목 긋고 잠들었어요.", ] for sample in samples: with self.subTest(sample=sample): result = guardrail.sanitize_client_reply(sample, ideation_stage=3, turn_seq=4) self.assertTrue(result.needs_regeneration) self.assertTrue(any(r.startswith("means_info:") for r in result.reasons)) def test_allows_boundary_setting_language(self) -> None: """'선을 긋다'(경계 설정) 같은 정상 상담 표현은 차단하지 않는다.""" result = guardrail.sanitize_client_reply( "이제는 엄마랑 선을 긋고 제 생활을 지키고 싶어요.", ideation_stage=1, turn_seq=4, ) self.assertFalse(result.needs_regeneration) def test_blocks_role_meta_speech_variants(self) -> None: samples = [ "내담자 역할로 응답하겠습니다. 엄마가 가보라고 해서요.", "AI로서 답변드리면 저는 우울한 학생입니다.", "상담자 입장에서 보면 제 핵심신념은 무가치감입니다.", "제 핵심신념은 저는 쓸모없다는 것입니다.", ] for sample in samples: with self.subTest(sample=sample): result = guardrail.sanitize_client_reply(sample, ideation_stage=1, turn_seq=2) self.assertTrue(result.needs_regeneration) self.assertIn("role_meta", result.reasons) def test_blocks_repeated_greeting_after_opening_turn_only_when_greeting_starts_reply(self) -> None: blocked = guardrail.sanitize_client_reply( "안녕하세요. 처음 뵙겠습니다. 저는 서연이에요.", ideation_stage=1, turn_seq=3, ) first_turn = guardrail.sanitize_client_reply( "안녕하세요. 엄마가 가보라고 해서 왔어요.", ideation_stage=1, turn_seq=1, ) quoted = guardrail.sanitize_client_reply( "방금 선생님이 안녕하세요라고 말해서 더 어색했어요.", ideation_stage=1, turn_seq=3, ) self.assertTrue(blocked.needs_regeneration) self.assertIn("repeat_greeting_after_opening", blocked.reasons) self.assertFalse(first_turn.needs_regeneration) self.assertFalse(quoted.needs_regeneration) def test_blocks_near_duplicate_previous_client_reply_but_allows_short_overlap(self) -> None: previous = "몰라요. 엄마가 그냥 가보라고 해서 왔어요." duplicate = guardrail.sanitize_client_reply( "몰라요. 엄마가 그냥 가보라고 해서 왔어요.", ideation_stage=1, previous_client_reply=previous, ) near_duplicate = guardrail.sanitize_client_reply( "엄마가 그냥 가보라고 해서 왔어요. 몰라요.", ideation_stage=1, previous_client_reply=previous, ) allowed = guardrail.sanitize_client_reply( "엄마가 가보라고 한 건 맞는데, 지금은 좀 짜증나요.", ideation_stage=1, previous_client_reply=previous, ) self.assertTrue(duplicate.needs_regeneration) self.assertIn("duplicate_client_reply", duplicate.reasons) self.assertTrue(near_duplicate.needs_regeneration) self.assertIn("duplicate_client_reply", near_duplicate.reasons) self.assertFalse(allowed.needs_regeneration) if __name__ == "__main__": unittest.main()