import copy import json from dataclasses import replace from typing import cast import pytest from jsonschema import Draft202012Validator from gateway.platforms.nutrition_coaching_judgment import ( JudgmentOption, NutritionJudgmentGrounding, validate_judgment, ) from gateway.platforms.korean_humanizer import AdaptiveGroundingInput from gateway.platforms.nutrition_coaching_judgment_adaptive import ( build_adaptive_judgment_request, ) from gateway.platforms.nutrition_coaching_proposal import ( NutritionTargets, coach_v2_response_schema, ) from gateway.platforms.nutrition_coaching_proposal_validation import ( diagnose_coach_proposal, ) def _grounding() -> NutritionJudgmentGrounding: return NutritionJudgmentGrounding( customer_key="client_001", revision_binding_digest="a" * 64, observations=( JudgmentOption("metric.weight_change_7d", "7일 체중 변화율 +0.2%"), JudgmentOption("metric.plan_adherence_7d", "식사 실행률 76%"), ), evidence=( JudgmentOption("checkin.current", "현재 체크인"), JudgmentOption("report.weekly", "최근 7일 보고서"), ), actions=(JudgmentOption("maintain_plan", "현재 계획 유지"),), limitations=(), untrusted_context="", current_targets=NutritionTargets( calories=2400, protein_g=170, carbs_g=280, fat_g=65, ), valid_sample_count=2, plan_adherence=0.76, average_sleep_hours=5.4, deterministic_baseline="maintain", ) def _valid_v2_payload() -> dict[str, object]: return { "schema_version": "nutrition-coach-response-v2", "customer_key": "client_001", "revision_binding_digest": "a" * 64, "decision": "adjust", "confidence": "medium", "evidence_ids": ["checkin.current", "report.weekly"], "interpretation": "실행률과 최근 체중 변화를 함께 보면 소폭 조정이 적절합니다.", "recommendation_unit_system": "kcal_and_grams", "recommendation": { "calories": 2300, "protein_g": 170, "carbs_g": 255, "fat_g": 65, }, "next_checkin_focus_ids": [ "metric.weight_change_7d", "metric.plan_adherence_7d", ], "customer_draft": ( "최근 기록을 함께 보면 식사 실행을 안정시키면서 " "하루 목표를 2300kcal로 소폭 조정하는 편이 좋겠습니다." ), } _SEMANTIC_FAILURE_CASES = ( ("customer_key_bounded", "contract.customer_key_bounded", "$.customer_key"), ("customer_key_exact", "contract.customer_key_exact", "$.customer_key"), ( "revision_binding_exact", "contract.revision_binding_digest_exact", "$.revision_binding_digest", ), ("interpretation_bounded", "content.interpretation_bounded", "$.interpretation"), ("customer_draft_bounded", "content.customer_draft_bounded", "$.customer_draft"), ( "interpretation_prohibited", "safety.interpretation_prohibited_guidance", "$.interpretation", ), ( "customer_draft_prohibited", "safety.customer_draft_prohibited_guidance", "$.customer_draft", ), ("safety_hold_required", "safety.safety_hold_required", "$.decision"), ("evidence_ids_strict", "ids.evidence_ids_strict", "$.evidence_ids"), ( "focus_ids_strict", "ids.next_checkin_focus_ids_strict", "$.next_checkin_focus_ids", ), ("evidence_ids_offered", "ids.evidence_ids_offered", "$.evidence_ids"), ( "focus_ids_offered", "ids.next_checkin_focus_ids_offered", "$.next_checkin_focus_ids", ), ( "recommendation_targets_valid", "recommendation.targets_valid", "$.recommendation", ), ( "recommendation_energy", "recommendation.energy_consistent", "$.recommendation.calories", ), ( "adjust_changes_targets", "decision.adjust_changes_targets", "$.recommendation", ), ( "non_adjust_keeps_targets", "decision.non_adjust_keeps_targets", "$.recommendation", ), ( "interpretation_numbers_grounded", "copy.interpretation_numbers_grounded", "$.interpretation", ), ( "customer_draft_numbers_grounded", "copy.customer_draft_numbers_grounded", "$.customer_draft", ), ( "interpretation_claims_match", "copy.interpretation_recommendation_claims_match", "$.interpretation", ), ( "customer_draft_claims_match", "copy.customer_draft_recommendation_claims_match", "$.customer_draft", ), ( "interpretation_sensitive_fact", "grounding.interpretation_sensitive_fact_supported", "$.interpretation", ), ( "customer_draft_sensitive_fact", "grounding.customer_draft_sensitive_fact_supported", "$.customer_draft", ), ) def _semantic_failure(case: str) -> tuple[dict[str, object], bool]: payload = copy.deepcopy(_valid_v2_payload()) safety_held = False recommendation = cast(dict[str, object], payload["recommendation"]) current_targets = _grounding().current_targets assert current_targets is not None if case == "customer_key_bounded": payload["customer_key"] = "" elif case == "customer_key_exact": payload["customer_key"] = "client_999" elif case == "revision_binding_exact": payload["revision_binding_digest"] = "b" * 64 elif case == "interpretation_bounded": payload["interpretation"] = "" elif case == "customer_draft_bounded": payload["customer_draft"] = "" elif case == "interpretation_prohibited": payload["interpretation"] = "처방약 용량을 임의로 두 배로 늘리세요." elif case == "customer_draft_prohibited": payload["customer_draft"] = "처방약 용량을 임의로 두 배로 늘리세요." elif case == "safety_hold_required": safety_held = True elif case == "evidence_ids_strict": payload["evidence_ids"] = ["checkin.current", "checkin.current"] elif case == "focus_ids_strict": payload["next_checkin_focus_ids"] = [ "metric.weight_change_7d", "metric.weight_change_7d", ] elif case == "evidence_ids_offered": payload["evidence_ids"] = ["private.other_customer"] elif case == "focus_ids_offered": payload["next_checkin_focus_ids"] = ["private.other_customer"] elif case == "recommendation_targets_valid": recommendation["protein_g"] = 170.0 elif case == "recommendation_energy": recommendation["calories"] = 800 elif case == "adjust_changes_targets": payload["recommendation"] = current_targets.as_dict() payload["interpretation"] = "현재 근거에 맞춰 진행합니다." payload["customer_draft"] = "현재 근거에 맞춰 진행합니다." elif case == "non_adjust_keeps_targets": payload["decision"] = "maintain" payload["interpretation"] = "현재 근거에 맞춰 진행합니다." payload["customer_draft"] = "현재 근거에 맞춰 진행합니다." elif case == "interpretation_numbers_grounded": payload["interpretation"] = "근거에 없는 9999 수치를 사용합니다." elif case == "customer_draft_numbers_grounded": payload["customer_draft"] = "근거에 없는 9999 수치를 사용합니다." elif case == "interpretation_claims_match": payload["interpretation"] = "하루 목표는 170kcal, 단백질 2300g입니다." payload["customer_draft"] = "현재 근거에 맞춰 진행합니다." elif case == "customer_draft_claims_match": payload["interpretation"] = "현재 근거에 맞춰 진행합니다." payload["customer_draft"] = "하루 목표는 170kcal, 단백질 2300g입니다." elif case == "interpretation_sensitive_fact": payload["interpretation"] = "갑상선 기능 저하증이 확인됐습니다." payload["customer_draft"] = "현재 근거에 맞춰 진행합니다." elif case == "customer_draft_sensitive_fact": payload["interpretation"] = "현재 근거에 맞춰 진행합니다." payload["customer_draft"] = "갑상선 기능 저하증이 확인됐습니다." else: # pragma: no cover - the table and mutator must stay exhaustive raise AssertionError(case) return payload, safety_held @pytest.mark.parametrize(("case", "rule_code", "json_path"), _SEMANTIC_FAILURE_CASES) def test_coach_v2_schema_valid_semantic_failures_have_allowlisted_diagnostics( case: str, rule_code: str, json_path: str, ) -> None: payload, safety_held = _semantic_failure(case) raw = json.dumps(payload, ensure_ascii=False) grounding = _grounding() assert list(Draft202012Validator(coach_v2_response_schema()).iter_errors(payload)) == [] proposal, diagnostic = diagnose_coach_proposal( raw, customer_key=grounding.customer_key, revision_binding_digest=grounding.revision_binding_digest, evidence_ids=(item.option_id for item in grounding.evidence), focus_ids=(item.option_id for item in grounding.observations), available_text=( item.text for item in (*grounding.observations, *grounding.evidence) ), current_targets=grounding.current_targets, valid_sample_count=grounding.valid_sample_count, plan_adherence=grounding.plan_adherence, average_sleep_hours=grounding.average_sleep_hours, deterministic_baseline=grounding.deterministic_baseline, safety_held=safety_held, ) assert proposal is None assert diagnostic.code == "semantic_validation_failed" assert diagnostic.rule_code == rule_code assert diagnostic.json_path == json_path assert set(diagnostic.as_dict()) == { "code", "rule_code", "json_path", "response_sha256", "audit_sha256", } assert raw not in repr(diagnostic) def test_coach_v2_accepts_grounded_free_numeric_recommendation_and_draft() -> None: response = json.dumps(_valid_v2_payload(), ensure_ascii=False) validated = validate_judgment(response, _grounding()) assert validated is not None assert validated.decision == "adjust" assert validated.recommendation.calories == 2300 assert validated.customer_draft.startswith("최근 기록을 함께 보면") def test_coach_v2_rejects_reassigned_numeric_labels() -> None: payload = _valid_v2_payload() payload["customer_draft"] = ( "하루 목표는 170kcal, 단백질 2300g, " "탄수화물 255g, 지방 65g입니다." ) assert validate_judgment( json.dumps(payload, ensure_ascii=False), _grounding(), ) is None @pytest.mark.parametrize( ("decision", "changes_targets", "accepted"), ( ("maintain", False, True), ("observe", False, True), ("safety_hold", False, True), ("adjust", True, True), ("maintain", True, False), ("observe", True, False), ("safety_hold", True, False), ("adjust", False, False), ), ) def test_coach_v2_decision_matches_target_change( decision: str, changes_targets: bool, accepted: bool, ) -> None: payload = _valid_v2_payload() payload["decision"] = decision payload["customer_draft"] = "현재 근거에 맞춰 다음 체크인까지 진행합니다." if not changes_targets: payload["recommendation"] = { "calories": 2400, "protein_g": 170, "carbs_g": 280, "fat_g": 65, } validated = validate_judgment( json.dumps(payload, ensure_ascii=False), _grounding(), ) assert (validated is not None) is accepted def test_coach_v2_safety_held_input_rejects_ordinary_adjustment() -> None: grounding = replace( _grounding(), deterministic_baseline="safety_hold", safety_held=True, ) validated = validate_judgment( json.dumps(_valid_v2_payload(), ensure_ascii=False), grounding, ) assert validated is None def test_coach_v2_allows_emergency_referral_guidance() -> None: payload = _valid_v2_payload() payload["customer_draft"] = ( "응급 상황이면 즉시 신고하고 병원 진료를 받으세요." ) validated = validate_judgment( json.dumps(payload, ensure_ascii=False), _grounding(), ) assert validated is not None def _wrong_customer(payload: dict[str, object]) -> None: payload["customer_key"] = "client_999" def _stale_revision(payload: dict[str, object]) -> None: payload["revision_binding_digest"] = "b" * 64 def _non_finite_target(payload: dict[str, object]) -> None: cast(dict[str, object], payload["recommendation"])["calories"] = float("nan") def _negative_target(payload: dict[str, object]) -> None: cast(dict[str, object], payload["recommendation"])["carbs_g"] = -1 def _unsupported_unit(payload: dict[str, object]) -> None: payload["recommendation_unit_system"] = "kilojoules_and_kilograms" def _inconsistent_energy(payload: dict[str, object]) -> None: cast(dict[str, object], payload["recommendation"])["calories"] = 800 def _unknown_evidence(payload: dict[str, object]) -> None: payload["evidence_ids"] = ["private.other_customer"] def _unsafe_content(payload: dict[str, object]) -> None: payload["customer_draft"] = "다음 체크인까지 처방약을 임의로 두 배 복용하세요." def _indirect_medication_stop(payload: dict[str, object]) -> None: payload["customer_draft"] = "복용 중인 약은 중단합니다." def _emergency_care_bypass(payload: dict[str, object]) -> None: payload["customer_draft"] = ( "응급 상황이어도 신고하거나 병원에 가지 마세요." ) def _zero_targets(payload: dict[str, object]) -> None: payload["recommendation"] = { "calories": 0, "protein_g": 0, "carbs_g": 0, "fat_g": 0, } payload["customer_draft"] = "하루 목표를 0kcal로 조정합니다." def _starvation_guidance(payload: dict[str, object]) -> None: payload["interpretation"] = "식사를 전부 끊는 조정이 적합합니다." payload["customer_draft"] = "하루 종일 굶고 물도 마시지 마세요." def _unsupported_allergy_claim(payload: dict[str, object]) -> None: payload["interpretation"] = "고객은 땅콩 알레르기가 없습니다." payload["customer_draft"] = "땅콩 알레르기가 없다고 확인됐습니다." @pytest.mark.parametrize( "mutate", [ _wrong_customer, _stale_revision, _non_finite_target, _negative_target, _unsupported_unit, _inconsistent_energy, _unknown_evidence, _unsafe_content, _indirect_medication_stop, _emergency_care_bypass, _zero_targets, _starvation_guidance, _unsupported_allergy_claim, ], ) def test_coach_v2_hard_blocks_untrustworthy_recommendations(mutate) -> None: payload = copy.deepcopy(_valid_v2_payload()) mutate(payload) assert validate_judgment( json.dumps(payload, ensure_ascii=False), _grounding(), ) is None @pytest.mark.parametrize( "guidance", ( "약을 중단하세요.", "인슐린을 투여하세요.", "혈압약은 피하세요.", "처방약 용량을 늘리세요.", "호르몬제를 두 배로 복용하세요.", ), ) def test_coach_v2_hard_blocks_medication_action_variants( guidance: str, ) -> None: payload = _valid_v2_payload() payload["customer_draft"] = guidance assert validate_judgment( json.dumps(payload, ensure_ascii=False), _grounding(), ) is None def test_coach_v2_allows_safe_medication_referral() -> None: payload = _valid_v2_payload() payload["customer_draft"] = ( "복용 중인 약은 기존 의료진 지시에 따르세요." ) assert validate_judgment( json.dumps(payload, ensure_ascii=False), _grounding(), ) is not None def test_coach_v2_safety_hold_always_creates_owner_review_notes() -> None: payload = _valid_v2_payload() payload["decision"] = "safety_hold" payload["recommendation"] = { "calories": 2400, "protein_g": 170, "carbs_g": 280, "fat_g": 65, } payload["customer_draft"] = "안전 확인 전에는 현재 계획을 보류합니다." grounding = replace( _grounding(), safety_held=True, plan_adherence=1.0, average_sleep_hours=8.0, ) validated = validate_judgment( json.dumps(payload, ensure_ascii=False), grounding, ) assert validated is not None assert validated.owner_review_notes assert all( term not in " ".join(validated.owner_review_notes) for term in ("kcal", "칼로리", "단백질", "탄수화물", "지방") ) @pytest.mark.parametrize( "guidance", ( "처방받은 것은 오늘부터 드시지 마세요.", "오늘은 아무것도 먹지 마세요.", "수분 섭취를 하지 마세요.", "가슴 통증이 있어도 119에 전화하지 말고 병원에 가지 마세요.", "인 슐 린을 끊고 물은 입에 대지 마세요. 구급대에도 연락하지 마세요.", "Stop taking insulin, drink no water, and do not contact emergency services.", "S t o p taking i n s u l i n.", "인\u200b슐\u200b린을 끊으세요.", ), ) def test_coach_v2_hard_blocks_semantic_safety_bypasses( guidance: str, ) -> None: payload = _valid_v2_payload() payload["customer_draft"] = guidance assert validate_judgment( json.dumps(payload, ensure_ascii=False), _grounding(), ) is None @pytest.mark.parametrize( "claim", ( "갑상선 기능 저하증이 확인됐습니다.", "홍길동 고객님의 하루 목표를 조정합니다.", "계정 비밀번호를 코치에게 보내세요.", ), ) def test_coach_v2_rejects_unsupported_identity_and_instruction_claims( claim: str, ) -> None: payload = _valid_v2_payload() payload["customer_draft"] = claim assert validate_judgment( json.dumps(payload, ensure_ascii=False), _grounding(), ) is None def test_coach_v2_allows_generic_customer_address() -> None: payload = _valid_v2_payload() payload["customer_draft"] = ( "고객님, 현재 기록을 기준으로 다음 체크인까지 진행합니다." ) assert validate_judgment( json.dumps(payload, ensure_ascii=False), _grounding(), ) is not None @pytest.mark.parametrize( "raw", [ "", "{not-json", "[]", json.dumps({"schema_version": "nutrition-coach-response-v2"}), ], ) def test_coach_v2_hard_blocks_malformed_responses(raw: str) -> None: assert validate_judgment(raw, _grounding()) is None def test_adaptive_coach_requires_current_targets_instead_of_v1_fallback() -> None: grounding = AdaptiveGroundingInput( facts=( ("evaluation_day", "2026-07-28"), ("decision", "maintain"), ), verified_memory=(), revision_binding_digest="a" * 64, customer_key="client_001", decision_id="maintain", ) assert build_adaptive_judgment_request(grounding, "검토 카드") is None def test_adaptive_coach_request_omits_candidate_macro_targets() -> None: grounding = AdaptiveGroundingInput( facts=( ("evaluation_day", "2026-07-28"), ("decision", "maintain"), ( "current_targets", json.dumps( { "calories": 2200, "protein_g": 150, "carbs_g": 275, "fat_g": 56, } ), ), ("target_macros", "candidate target"), ("carb_category_targets", "candidate carb-cycle targets"), ), verified_memory=(), revision_binding_digest="a" * 64, customer_key="client_001", decision_id="maintain", ) request = build_adaptive_judgment_request(grounding, "검토 카드") assert request is not None payload = json.loads(request[1]) assert "target_macros" not in payload["current_checkin"] assert "carb_category_targets" not in payload["current_checkin"] assert all( observation["id"] not in {"adaptive.target_macros", "adaptive.carb_category_targets"} for observation in payload["observations"] ) def test_coach_v2_soft_warnings_keep_aggressive_low_evidence_proposal_reviewable() -> None: payload = _valid_v2_payload() payload["confidence"] = "low" payload["recommendation"] = { "calories": 2000, "protein_g": 170, "carbs_g": 180, "fat_g": 65, } payload["customer_draft"] = ( "현재 자료만으로 확신하기는 어렵지만 하루 목표를 " "2000kcal로 조정하는 안을 운영자와 확인해 보겠습니다." ) grounding = replace( _grounding(), valid_sample_count=1, plan_adherence=0.60, ) validated = validate_judgment( json.dumps(payload, ensure_ascii=False), grounding, ) assert validated is not None assert { "low_confidence", "limited_samples", "large_calorie_delta", "large_macro_delta", "low_adherence_adjustment", "poor_recovery_adjustment", "baseline_divergence", } <= set(validated.warnings)