447 lines
22 KiB
Python
447 lines
22 KiB
Python
"""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",
|
|
"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_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()
|