"""Strict client reply quality gate regressions.""" from __future__ import annotations import asyncio 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 def _load_clinical_observation_runner(): runner_path = repo_path("scripts", "run-clinical-crisis-technical-observations.py") spec = importlib.util.spec_from_file_location("clinical_crisis_observation_runner", runner_path) if spec is None or spec.loader is None: raise RuntimeError(f"임상 기술 관찰 runner를 불러올 수 없다: {runner_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", "reviewer_role_or_qualification", "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"]] required_assessment_fields = { "decision", "rationale", "findings_and_required_changes", "re_review", "reviewed_at", } self.assertTrue( all(set(assessment) == required_assessment_fields for assessment in assessments), ) required_youth_response_fields = { "guardian_notification", "means_restriction", "risk_disposition", "follow_up_48_72h", "documentation", "safeguarding", } youth_responses = case_set["youth_review_responses"] self.assertEqual(set(youth_responses), required_youth_response_fields) 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()), ) self.assertTrue(all(value is None for value in youth_responses.values())) else: self.assertTrue(all(approval[field] for field in required_approval_fields)) self.assertTrue( all(value for assessment in assessments for value in assessment.values()), ) self.assertTrue(all(value for value in youth_responses.values())) 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_exact_review_case_signals_drive_runtime_safety_gates(self) -> None: case_set = json.loads( repo_path("data", "clinical", "p1-crisis-review-cases.json").read_text( encoding="utf-8", ), ) cases = {case["case_id"]: case for case in case_set["cases"]} for case_id in ( "P1-CRISIS-001", "P1-CRISIS-002", "P1-CRISIS-005", "P1-CRISIS-006", ): with self.subTest(case_id=case_id): signal = cases[case_id]["synthetic_scenario"]["signal"] self.assertTrue(guardrail.crisis_signal_matches(signal)) sanitized = guardrail.sanitize_client_reply( signal, ideation_stage=3, turn_seq=4, ) self.assertFalse(sanitized.needs_regeneration) counselor_signal = cases["P1-CRISIS-003"]["synthetic_scenario"]["signal"] counselor_result = guardrail.classify_crisis( counselor_signal, speaker_is_persona_context=True, ) self.assertEqual(counselor_result.kind, guardrail.CrisisKind.PERSONA_PLAY) self.assertFalse(counselor_result.escalate) learner_signal = cases["P1-CRISIS-004"]["synthetic_scenario"]["signal"] learner_result = guardrail.classify_crisis( learner_signal, speaker_is_persona_context=True, ) self.assertEqual(learner_result.kind, guardrail.CrisisKind.LEARNER_REAL) self.assertGreaterEqual(learner_result.risk_level, 4) self.assertTrue(learner_result.escalate) self.assertIn("first_person_current_danger_help", learner_result.matched) counselor_questions = ( "지금 위험할 것 같거나 실제로 도움이 필요하신가요?", "지금 제 질문에 답하기 어렵다면 넘어가도 돼요. 자살 생각이 있나요?", "지금 제 앞에서 안전을 스스로 지키기 어렵다고 느끼시나요?", "자살 생각이 들면 ‘도와주세요’라고 말할 수 있나요?", ( "내담자가 ‘저 자신이 지금 위험할 것 같아요. 실제로 도움이 필요해요.’라고 " "말하면 어떻게 확인해야 하나요?" ), ) for question in counselor_questions: with self.subTest(question=question): counselor_question = guardrail.classify_crisis( question, speaker_is_persona_context=True, ) self.assertFalse(counselor_question.escalate) real_help_questions = ( "저 자신이 지금 위험할 것 같다고 느끼고 실제로 도움이 필요해요. 어떻게 해야 하나요?", "저는 지금 위험하다고 느끼며 실제로 도움이 필요해요. 도와줄 수 있나요?", "제가 현재 위험하다고 느끼고 있어요. 실제로 도움이 필요해요?", "저는 학생인데 지금 제가 위험하다고 느끼고 실제로 도움이 필요해요.", "저는 내담자지만 지금 제가 위험하다고 느끼고 실제로 도움이 필요해요.", "상대가 떠난 뒤 저는 지금 위험하다고 느끼고 실제로 도움이 필요해요.", "지금 제가 위험한 것 같고 도움이 필요해요.", "지금 저는 위험하다고 느껴요. 도움이 필요합니다.", "저 스스로 지금 위험할 것 같아요. 도움이 필요해요.", "제 안전이 지금 위험해요. 실제 도움이 필요해요.", "제가 지금 안전하지 않아서 도와주세요.", "저는 지금 당장 스스로를 해칠까 봐 무서워요. 도와주세요.", "지금 제가 저를 해칠 것 같아요. 도와주세요.", "저는 지금 죽어버리고 싶어요. 도와주세요.", "저 지금 제 목숨을 끊고 싶어요.", "저는 당장 죽고 싶어요.", "저는 죽고 싶어요. 제발 도와주세요.", "제가 자살하고 싶어요. 도움이 필요해요.", "저는 지금 살고 싶지 않아요. 도와주세요.", "저는 지금 삶을 끝내고 싶어요. 도와주세요.", "저는 지금 없어지고 싶어요. 도와주세요.", "저 지금 스스로를 다치게 하고 싶어요. 도와주세요.", "지금은 제가 제 안전을 지킬 자신이 없어요. 도와주세요.", "아까는 자살하고 싶지 않았지만 저는 지금 죽고 싶어요.", "죽고 싶어하는 친구를 돕다가 저도 지금 죽고 싶어요.", "죽고 싶은 생각이 있는지 묻는 연습 중이었는데 저는 지금 정말 죽고 싶어요.", "저는 지금 죽고 싶어요. 친구는 자살하고 싶지 않아요.", "아까는 도움이 필요하지 않았지만 저는 지금 위험하고 실제로 도움이 필요해요.", "저는 오늘 죽고 싶어요. 도와주세요.", "죽고 싶어요. 저 좀 도와주세요.", "저는 지금 죽어야 할 것 같아요. 도와주세요.", "저는 지금 통제가 안 되고 저를 다치게 할 것 같아요. 도와주세요.", "저는 지금 너무 위험한 상태예요. 도움이 필요해요.", "저한테 지금 자해 충동이 있어요. 도와주세요.", "저는 현재 살아갈 이유가 없어요. 도와주세요.", "저는 현재 제 자신을 해칠 것 같아요. 도와주세요.", "저는 지금 자살할 것 같아요. 도와주세요.", "저는 지금 자해할 것 같아요. 도와주세요.", "저는 지금 안전을 지킬 수 없어요. 도와주세요.", "저는 지금 저를 해칠지도 몰라요. 도와주세요.", "제가 지금 저를 다치게 할까 봐 두려워요. 도움이 필요해요.", "저는 지금 혼자 있으면 위험해요. 도와주세요.", "저는 지금 자살하려고 해요.", "지금 당장 죽으려고 해요.", "제 목숨을 끊으려고 해요.", "자살 계획이 있어요.", "자해 계획이 있어요.", "자살 시도를 하려고 해요.", "자해 시도를 하려고 해요.", "자살 충동이 있어요.", "자살 생각이 있어요.", "자살할지도 모르겠어요.", "스스로를 해칠 계획이 있어요.", "제 안전을 유지할 수 없어요.", "저는 지금 자살을 하려고 해요.", "저는 지금 자해를 하려고 해요.", "저는 지금 자살하려 해요.", "저는 지금 죽으려 해요.", "저는 오늘 밤 자살하려고 해요.", "자살할 계획이에요.", "죽을 계획이에요.", "자살할 구체적인 계획이 있어요.", "자살할 의도가 있어요.", "자살 계획을 세웠어요.", "자해 계획을 세웠어요.", "제 목숨을 끊을 계획을 세웠어요.", "제 자신을 해치려고 해요.", "삶을 끝내려고 해요.", "죽을 준비를 하고 있어요.", "제 안전을 보장할 수 없어요. 도와주세요.", "저는 지금 자살을 시도할지도 몰라요.", "저는 지금 자해를 시도할지도 몰라요.", "저는 지금 자해 생각이 반복돼요.", "저는 지금 제 안전을 유지하기 어려워요. 도와주세요.", "저는 지금 제 안전을 보장하기 어려워요. 도와주세요.", "저는 지금 자살을 결심했어요.", "저는 지금 죽기로 결심했어요.", "저는 지금 자해하기로 했어요.", "저는 오늘 밤 자살할 거예요.", "저는 오늘 밤 자해할 거예요.", "저는 지금 자살을 준비하고 있어요.", "저는 지금 자살을 실행하려고 해요.", "저는 지금 자살할래요.", "저는 지금 죽을래요.", "저는 지금 제 목숨을 끊을지도 몰라요.", "저는 지금 자살 생각이 반복돼요.", "저는 오늘 밤 제 목숨을 끊을 거예요.", "저는 지금 정말로 자살하려고 해요.", "저는 지금 실제로 자살하려고 해요.", "저는 지금 곧 자살하려고 해요.", "저는 오늘 저녁 자살하려고 해요.", "저는 내일 자살하려고 해요.", "친구는 자살할 것 같지 않지만 저는 지금 자살할지도 몰라요.", "역할극에서는 죽으려고 한다고 말했지만 저는 지금 실제로 죽으려고 해요.", "역할극을 끝냈고 저는 지금 실제로 죽으려고 해요.", "역할극은 끝났고 실제로 저는 지금 죽으려고 해요.", "역할극이 아니라 저는 지금 실제로 죽으려고 해요.", "번역이 아니라 저는 지금 죽고 싶어요.", ) for signal in real_help_questions: with self.subTest(signal=signal): result = guardrail.classify_crisis( signal, speaker_is_persona_context=True, ) self.assertEqual(result.kind, guardrail.CrisisKind.LEARNER_REAL) self.assertTrue(result.escalate) nonclinical_first_person = guardrail.classify_crisis( "제가 지금 과제 제출이 위험할 것 같아서 실제로 도움이 필요해요.", speaker_is_persona_context=True, ) ordinary_wish = guardrail.classify_crisis( "걱정이 없어졌으면 좋겠어요.", speaker_is_persona_context=True, ) counselor_meta_statements = ( "제가 지금 여쭤보는 건 자살 생각이 있는지예요.", "제가 지금 묻는 질문은 혹시 죽고 싶은 생각이 있는지예요.", "제가 지금 설명드리는 109는 자살 위기 때 연락하는 번호예요.", "저 지금 상담자로서 자살 위험을 확인하고 있어요.", "먼저 지금 죽고 싶은 생각이 있는지 확인할게요.", "그러나 지금 죽고 싶은 생각이 있는지는 확인해야 해요.", "누구나 지금 자살하고 싶다고 말하는 것은 아니에요.", "저는 지금 죽고 싶은 생각이 있는지 묻고 있어요.", "제가 지금 자살하고 싶은지 묻는 거예요.", "나는 지금 자해하고 싶은지 물어보는 연습을 하고 있어요.", "저는 지금 자살하고 싶은 친구를 돕고 있어요.", "저는 지금 죽고 싶은 생각이 있는지 여쭤볼게요.", "제가 지금 죽고 싶은 생각이 있는지 직접 확인해도 될까요?", "저는 지금 자살하고 싶지 않아요.", "저는 지금 죽고 싶은 건 아니에요.", "저 자신이 지금 위험할 것 같지 않아요. 실제로 도움이 필요하지 않아요.", "문제가 없어졌으면 좋겠어요.", "안내가 없어졌으면 좋겠어요.", "나는 지금 죽고 싶냐고 물어보는 연습 중이에요.", "저는 지금 죽고 싶다고 가정하고 역할극을 해볼게요.", "저는 지금 죽고 싶다는 친구를 돕고 있어요.", "저는 지금 자살하고 싶지는 않아요.", "“저는 지금 죽고 싶어요”라고 예시 문장을 읽었어요.", "저는 지금 죽고 싶으신지 여쭙고 있어요.", "저는 지금 자살하고 싶다는 문장을 번역하고 있어요.", "저는 지금 자해하고 싶어 하는 동생을 돕고 있어요.", "저는 지금 죽고 싶진 않아요.", "저는 지금 죽고 싶다는 뜻은 아니에요.", "저는 지금 죽고 싶은 마음은 없어요.", "저는 지금 죽고 싶으면 109에 전화하라고 설명하고 있어요.", "“저는 지금 죽고 싶어요” 문장을 번역해 주세요.", "저는 지금 죽고 싶은 생각이 없어요.", "저는 지금 죽고 싶은 마음이 전혀 없습니다.", "제가 지금 죽고 싶다고 말한 학생을 돕고 있어요.", "제가 지금 죽고 싶다고 한 친구를 돕고 있어요.", "저는 지금 자살 생각이 없어요.", "저는 지금 자해 충동이 없어요.", "저는 예전에 죽고 싶었지만 지금은 괜찮아요.", "저는 지금 자살할 것 같지 않아요.", "저는 지금 자해할 것 같지 않아요.", "저는 지금 살기 싫지 않아요.", "저는 지금 없어지고 싶지 않아요.", "저는 지금 죽어야 할 것 같지 않아요.", "저는 지금 살아갈 이유가 없는 건 아니에요.", "저는 지금 자해 충동이 있지는 않아요.", "저는 지금 자살할 것 같진 않아요.", "저는 지금 자해할 것 같지는 않아요.", "저는 지금 살기 싫지는 않아요.", "저는 지금 없어지고 싶지는 않아요.", "저는 지금 죽어야 할 것 같지는 않아요.", "저는 지금 살아갈 이유가 없지는 않아요.", "저는 지금 자살 충동이 있지 않아요.", "저는 지금 자살 계획이 있는 게 아니에요.", "저는 지금 자살 생각이 있는 건 아니에요.", "친구가 자살 계획이 있다고 말했어요.", "배우자가 자살 생각이 있다고 말했어요.", "제가 친구에게 자살 계획이 있는지 물어봤어요.", "자살 생각이 있는지 확인할게요.", "예전에 자살 계획이 있었지만 지금은 없어요.", "저는 지금 제 안전을 유지할 수 없는 건 아니에요.", "저는 지금 살고 싶지 않은 건 아니에요.", "저는 지금 사라지고 싶지 않아요.", "저는 지금 자살 시도를 하려는 건 아니에요.", "저는 지금 자해 시도를 하려는 건 아니에요.", "저는 지금 자살하려고 하지는 않아요.", "저는 지금 죽으려고 하지는 않아요.", "저는 지금 자살할 계획은 없어요.", "저는 지금 죽을 계획이 없어요.", "저는 지금 자살할 의도는 없어요.", "저는 지금 죽을 준비를 하지 않았어요.", "자살할 계획이 있는지 확인할게요.", "죽을 계획이 있는지 물어볼게요.", "제 안전을 보장할 수 없는지 확인할게요.", "죽을 준비를 하고 있는지 묻겠습니다.", "자살할 의도가 있는지 확인합니다.", "자살 시도라는 문장을 번역하고 있어요.", "자해 계획 설문을 설명하고 있어요.", "아버지가 자살할 계획이 있다고 말했어요.", "엄마가 죽을 계획이 있다고 말했어요.", "동료가 자살할 의도가 있다고 말했어요.", "그 사람이 자살 계획이 있다고 말했어요.", ) self.assertFalse(nonclinical_first_person.escalate) self.assertEqual(ordinary_wish.kind, guardrail.CrisisKind.NONE) for statement in counselor_meta_statements: with self.subTest(statement=statement): result = guardrail.classify_crisis( statement, speaker_is_persona_context=True, ) self.assertFalse(result.escalate) def test_runtime_observation_artifact_matches_current_exact_case_execution(self) -> None: artifact = json.loads( repo_path( "docs", "ops", "evidence", "c-001-crisis-case-runtime-observations-2026-08-28.json", ).read_text(encoding="utf-8") ) runner = _load_clinical_observation_runner() current = asyncio.run(runner.build_observations(artifact["generated_at"])) self.assertEqual( artifact["schema_version"], "vignette.p1_crisis_technical_observations.v2", ) self.assertEqual(artifact["summary"]["case_count"], 6) self.assertEqual(artifact["summary"]["technical_pass_count"], 6) self.assertEqual(artifact["summary"]["technical_fail_count"], 0) self.assertEqual(artifact["summary"]["external_clinical_decisions_recorded"], 0) self.assertTrue( all( observation["technical_result"] == "pass" and observation["external_clinical_decision"] is None for observation in artifact["observations"] ) ) self.assertEqual(artifact["case_set"], current["case_set"]) self.assertTrue(artifact["runtime_package"]["matches_head"]) self.assertTrue(current["runtime_package"]["matches_head"]) self.assertEqual( artifact["runtime_package"]["files"], current["runtime_package"]["files"], ) self.assertEqual(artifact["boundary"], current["boundary"]) self.assertEqual(artifact["summary"], current["summary"]) self.assertEqual(artifact["observations"], current["observations"]) 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()