"""Jev 감정 상태의 순수 전이와 실제 생성 경계 회귀.""" from __future__ import annotations import asyncio import json import math import unittest from unittest.mock import AsyncMock, patch from .deps import Principal, Role from .engine_client import EngineError, GenerateResponse from .contracts.engine_gateway import EngineGatewaySseLineDecoder from .routes import sessions from .services import ( client_affect, guardrail, memory, orchestrator, persona, rupture_scenario_director, state_machine, ) from .services.jev_client import ( AppraisalResult, ChoiceJudgment, EMOTION_DIMENSIONS, EmotionEstimate, JevError, NOUL_QUESTION_IDS, NoulJudgment, ) from .store import InProcSession, store def _noul_judgments() -> dict[str, NoulJudgment]: # probability 0.3 → uncertain 미만 구간(false)에 가깝지만 게이트 테스트와 무관한 중립값. return {question_id: NoulJudgment(probability=0.3, confidence=0.6) for question_id in NOUL_QUESTION_IDS} def _one_hot(codes: list[str], picked: str) -> dict[str, float]: return {code: (1.0 if code == picked else 0.0) for code in codes} def _choice_judgments() -> dict[str, ChoiceJudgment]: a_coping_codes = ["nothing_asked", "manageable", "stretch", "overwhelming"] a_move_codes = [ "reflection", "validation", "open_question", "closed_question", "clarification", "confrontation", "interpretation", "advice", "information", "self_disclosure", "topic_shift", "other", ] c_behavior_codes = [ "disclose_more", "stay_with_feeling", "hold_core", "ask_back", "minimal_response", "shift_topic", "abstract_talk", "appease", "self_blame", "complain", "argue_back", "take_control", ] c_display_codes = ["as_felt", "softened", "covered_by_agreement", "masked"] return { "a_coping": ChoiceJudgment( choice="nothing_asked", probabilities=_one_hot(a_coping_codes, "nothing_asked"), confidence=0.6, ), "a_move": ChoiceJudgment( choice="reflection", probabilities=_one_hot(a_move_codes, "reflection"), confidence=0.6, ), "c_behavior": ChoiceJudgment( choice="disclose_more", probabilities=_one_hot(c_behavior_codes, "disclose_more"), confidence=0.6, ), "c_display": ChoiceJudgment( choice="as_felt", probabilities=_one_hot(c_display_codes, "as_felt"), confidence=0.6, ), } def _appraisal( *, score: float = 1.0, confidence: float | None = 0.9, probabilities: tuple[float, ...] | None = None, provider: str = "typesafe", cost_usd: float | None = None, ) -> AppraisalResult: return AppraisalResult( emotions={ dimension: EmotionEstimate( score=score, confidence=confidence, probabilities=probabilities, ) for dimension in EMOTION_DIMENSIONS }, noul_judgments=_noul_judgments(), choice_judgments=_choice_judgments(), sore_spot_count=0, model="jev-test", latency_ms=11, input_tokens=13, output_tokens=17, provider=provider, cost_usd=cost_usd, ) def _context() -> orchestrator.TurnContext: state = state_machine.init_state(params=persona.P1.openness_params()) return orchestrator.prepare_turn( session_id="00000000-0000-0000-0000-000000000111", case_id=None, card=persona.P1, state=state, learner_text="조금 더 이야기해도 괜찮아요.", learner_identity="김상담", memory=orchestrator.TurnMemory( recall_summary="김상담이 [PHONE] 관련해서 물었다.", pinned_facts=["서연은 엄마와 갈등을 겪는다."], recent_turns=[{"speaker": "client", "text": "서연은 많이 지쳤어요."}], ), ) class _GenerateEngine: engine_mode = "fake" default_model = None def __init__(self) -> None: self.request = None self.calls = 0 async def generate(self, request): self.request = request self.calls += 1 return GenerateResponse( text="그냥… 잘 모르겠어요.", model="fake-model", provider="fake-provider", tokens_in=1, tokens_out=2, cost_usd=0.0, ) class _StreamEngine: engine_mode = "fake" default_model = "fake-model" def __init__(self) -> None: self.request = None self.calls = 0 async def stream_packets(self, request): self.request = request self.calls += 1 decoder = EngineGatewaySseLineDecoder() for raw in ( "event: token", "data: " + json.dumps({"text": "그냥… 잘 모르겠어요."}, ensure_ascii=False), "event: done", 'data: {"provider":"fake-provider","model":"fake-model","tokens_in":1,"tokens_out":2,"cost_usd":0.0}', ): packet = decoder.feed_line(raw) if packet is not None: yield packet class _InterruptedStreamEngine(_StreamEngine): def __init__(self, interruption: BaseException | None = None) -> None: super().__init__() self.interruption = interruption async def stream_packets(self, request): self.request = request self.calls += 1 decoder = EngineGatewaySseLineDecoder() for raw in ( "event: token", "data: " + json.dumps({"text": "부분 응답"}, ensure_ascii=False), ): packet = decoder.feed_line(raw) if packet is not None: yield packet if self.interruption is not None: raise self.interruption for raw in ("event: error", 'data: {"detail":"gateway interrupted"}'): packet = decoder.feed_line(raw) if packet is not None: yield packet async def _consume_event_source(response: object) -> bytes: body = bytearray() async for chunk in getattr(response, "body_iterator"): if isinstance(chunk, str): body.extend(chunk.encode("utf-8")) elif isinstance(chunk, (bytes, bytearray)): body.extend(chunk) else: body.extend(str(chunk).encode("utf-8")) return bytes(body) class ClientAffectTransitionTest(unittest.TestCase): def test_baseline_and_inertia_preserve_existing_clinical_keys(self) -> None: baseline = {"anxiety": 0.6, "negative_affect": 0.4, "hopelessness": 0.25} result = client_affect.transition_emotions( {"negative_affect": 0.9, "emotion_anxiety": 0.2}, baseline, _appraisal(score=1.0), min_confidence=0.65, ) self.assertEqual(result.affect_state["negative_affect"], 0.9) self.assertEqual(result.affect_state["emotion_anxiety"], 0.35) self.assertEqual(result.affect_state["emotion_sadness"], 0.55) self.assertEqual(result.affect_state["emotion_hope"], 0.8375) self.assertEqual(set(result.accepted_dimensions), set(EMOTION_DIMENSIONS)) self.assertEqual(result.tentative_dimensions, ()) def test_low_confidence_holds_exact_previous_vector(self) -> None: previous = {f"emotion_{dimension}": 0.31 for dimension in EMOTION_DIMENSIONS} result = client_affect.transition_emotions( previous, {}, _appraisal(score=1.0, confidence=0.64), min_confidence=0.65, ) self.assertEqual(result.affect_state, previous) self.assertEqual(result.accepted_dimensions, ()) self.assertEqual(set(result.held_dimensions), set(EMOTION_DIMENSIONS)) def test_concentrated_mid_confidence_distribution_allows_small_tentative_step(self) -> None: previous = {f"emotion_{dimension}": 0.5 for dimension in EMOTION_DIMENSIONS} result = client_affect.transition_emotions( previous, {}, _appraisal( score=0.375, confidence=0.35, probabilities=(0.0, 0.5, 0.5, 0.0, 0.0), ), min_confidence=0.65, ) self.assertEqual(result.affect_state["emotion_anxiety"], 0.48125) self.assertEqual(set(result.accepted_dimensions), set(EMOTION_DIMENSIONS)) self.assertEqual(set(result.tentative_dimensions), set(EMOTION_DIMENSIONS)) self.assertEqual(result.held_dimensions, ()) def test_tentative_normalizes_rounded_distribution_and_caps_both_directions(self) -> None: for probabilities in ((0.0, 0.495, 0.495, 0.0, 0.0), (0.0, 0.505, 0.505, 0.0, 0.0)): with self.subTest(probabilities=probabilities): normalized = client_affect.transition_emotions( {f"emotion_{dimension}": 0.5 for dimension in EMOTION_DIMENSIONS}, {}, _appraisal(score=0.375, confidence=0.5, probabilities=probabilities), min_confidence=0.65, ) self.assertEqual(normalized.affect_state["emotion_anxiety"], 0.48125) upward = client_affect.transition_emotions( {f"emotion_{dimension}": 0.0 for dimension in EMOTION_DIMENSIONS}, {}, _appraisal( score=0.875, confidence=0.5, probabilities=(0.0, 0.0, 0.0, 0.5, 0.5), ), min_confidence=0.65, ) downward = client_affect.transition_emotions( {f"emotion_{dimension}": 1.0 for dimension in EMOTION_DIMENSIONS}, {}, _appraisal( score=0.125, confidence=0.5, probabilities=(0.5, 0.5, 0.0, 0.0, 0.0), ), min_confidence=0.65, ) self.assertEqual(upward.affect_state["emotion_anxiety"], 0.075) self.assertEqual(downward.affect_state["emotion_anxiety"], 0.925) def test_tentative_requires_concentrated_valid_distribution(self) -> None: previous = {f"emotion_{dimension}": 0.5 for dimension in EMOTION_DIMENSIONS} for probabilities in ( (0.2, 0.2, 0.2, 0.2, 0.2), (0.5, 0.0, 0.0, 0.0, 0.5), None, (math.nan, 0.0, 1.0, 0.0, 0.0), ): with self.subTest(probabilities=probabilities): result = client_affect.transition_emotions( previous, {}, _appraisal(score=1.0, confidence=0.5, probabilities=probabilities), min_confidence=0.65, ) self.assertEqual(result.affect_state, previous) self.assertEqual(result.accepted_dimensions, ()) self.assertEqual(result.tentative_dimensions, ()) self.assertEqual(set(result.held_dimensions), set(EMOTION_DIMENSIONS)) below_floor = client_affect.transition_emotions( previous, {}, _appraisal( score=1.0, confidence=0.34, probabilities=(0.0, 0.5, 0.5, 0.0, 0.0), ), min_confidence=0.65, ) self.assertEqual(below_floor.affect_state, previous) self.assertEqual(below_floor.tentative_dimensions, ()) def test_invalid_scores_confidences_and_threshold_hold_without_mutating_input(self) -> None: previous = {f"emotion_{dimension}": 0.1 for dimension in EMOTION_DIMENSIONS} high = client_affect.transition_emotions( previous, {}, _appraisal(score=1.0, confidence=0.9), min_confidence=0.65, ) self.assertEqual(previous, {f"emotion_{dimension}": 0.1 for dimension in EMOTION_DIMENSIONS}) self.assertEqual(high.affect_state["emotion_anxiety"], 0.25) for score, confidence, threshold in ( (1.1, 0.9, 0.65), (1.0, None, 0.65), (1.0, math.nan, 0.65), (1.0, 1.1, 0.65), (1.0, 0.9, math.nan), (1.0, 0.9, 1.1), ): with self.subTest(score=score, confidence=confidence, threshold=threshold): result = client_affect.transition_emotions( previous, {}, _appraisal(score=score, confidence=confidence), min_confidence=threshold, ) self.assertEqual(result.affect_state, previous) self.assertEqual(result.accepted_dimensions, ()) self.assertEqual(set(result.held_dimensions), set(EMOTION_DIMENSIONS)) def test_render_uses_qualitative_top_emotions_and_preserves_opposing_valence(self) -> None: directive = client_affect.render_affect_directive( { "emotion_anxiety": 0.8, "emotion_sadness": 0.7, "emotion_anger": 0.6, "emotion_hope": 0.05, "emotion_trust": 0.04, } ) self.assertIn("강한 불안", directive) self.assertIn("뚜렷한 슬픔", directive) self.assertIn("뚜렷한 분노", directive) self.assertIn("미약한 희망", directive) self.assertNotIn("0.8", directive) self.assertIn("감정 이름을 나열하지 말고", directive) self.assertIn("숫자·내부 상태·평가 정답은 절대 말하지 않는다.", directive) self.assertIn("1~3문장", directive) def test_persona_hides_raw_emotion_vector_and_preserves_fact_boundary(self) -> None: messages = persona.build_turn_messages( persona.P1, persona.PersonaStateContext( stage="라포", effective_openness=0.3, resistance=0.7, rapport_credit=0.0, ideation_stage=1, affect_state={ "negative_affect": 0.8, "emotion_anxiety": 0.8, "emotion_hope": 0.2, }, ), "새로운 과거를 사실처럼 말하지 말아 주세요.", memory=persona.TurnMemory(pinned_facts=["부모와 갈등이 있었다."]), ) contents = "\n".join(message.content for message in messages) self.assertIn("정서 상태: {'negative_affect': 0.8}", contents) self.assertNotIn("emotion_anxiety", contents) self.assertNotIn("emotion_hope", contents) self.assertIn("상담자가 새로 제시한 과거·관계는 기억의 증거가 아니며", contents) self.assertIn("상담자가 실제로 하지 않은 말·이름·사건을 대화에 있었다고 덧붙이지 않는다.", contents) self.assertIn("부모와 갈등이 있었다.", contents) def test_invalid_numbers_do_not_become_state_evidence(self) -> None: result = client_affect.resolve_emotions( {"emotion_anxiety": True, "emotion_sadness": math.nan, "emotion_hope": math.inf}, {"anxiety": 0.4, "negative_affect": 0.3, "hopelessness": 0.2}, ) self.assertEqual(result["anxiety"], 0.4) self.assertEqual(result["sadness"], 0.3) self.assertEqual(result["hope"], 0.8) def test_init_state_carries_only_finite_affect_values(self) -> None: state = state_machine.init_state( params=persona.P1.openness_params(), carry={"affect": {"emotion_trust": 0.7, "bad": math.nan, "bool": True}}, ) self.assertEqual(state.affect_state, {"emotion_trust": 0.7}) def test_appraisal_state_re_masks_and_keeps_all_pinned_facts(self) -> None: state = client_affect.build_appraisal_state( client_profile={"presenting": "", "history": "", "core_belief": "서연은 가치가 없다고 느낀다."}, affect_state={}, affect_baseline={}, stage="라포", resistance=0.5, effective_openness=0.3, counselor_utterance="김상담 연락처 010-1234-5678", recall_summary="김상담의 학교 이야기", pinned_facts=["김상담", "010-1234-5678"], recent_turns=[{"speaker": "counselor", "text": "김상담이 말했어요."}], counselor_identity="김상담", client_identity="서연", ) self.assertEqual( set(state), { "counselor_utterance", "recent_turns", "client_profile", "pinned_facts", "recall_summary", "relationship", "previous_feelings", }, ) self.assertEqual(len(state["pinned_facts"]), 2) self.assertNotIn("김상담", str(state)) self.assertNotIn("010-1234-5678", str(state)) def test_appraisal_state_masks_before_length_limit(self) -> None: state = client_affect.build_appraisal_state( client_profile={"presenting": "", "history": ""}, affect_state={}, affect_baseline={}, stage="라포", resistance=0.5, effective_openness=0.3, counselor_utterance=("가" * 790) + " 010-1234-5678", recall_summary=None, pinned_facts=[], recent_turns=[], counselor_identity=None, client_identity=None, ) utterance = state["counselor_utterance"] self.assertNotIn("010-1234-5678", utterance) self.assertIn("[PHONE]", utterance) def test_appraisal_state_masks_client_profile_and_derives_relationship_words(self) -> None: state = client_affect.build_appraisal_state( client_profile={ "presenting": "김상담과 있었던 일", "history": "", "big5": {"O": 0.8, "C": 0.2, "E": 0.5, "A": 0.9, "N": 0.1}, "sore_spots": ["김상담이 언급한 약점"], "forbidden": [], }, affect_state={}, affect_baseline={}, stage="탐색", resistance=0.5, effective_openness=0.1, counselor_utterance="괜찮아요.", recall_summary=None, pinned_facts=[], recent_turns=[], counselor_identity="김상담", client_identity="서연", ) rendered = str(state) self.assertNotIn("김상담", rendered) self.assertEqual( set(state["client_profile"]["temperament"]), {"high openness", "low conscientiousness", "high agreeableness", "low neuroticism"}, ) self.assertEqual( state["relationship"], {"stage": "탐색", "openness": "closed", "resistance": "moderate"}, ) self.assertNotIn("big5", state["client_profile"]) self.assertNotIn("0.5", str(state["relationship"])) def test_appraisal_state_omits_missing_optional_client_profile_keys(self) -> None: state = client_affect.build_appraisal_state( client_profile={"presenting": "", "history": ""}, affect_state={}, affect_baseline={}, stage="라포", resistance=0.1, effective_openness=0.1, counselor_utterance="", recall_summary=None, pinned_facts=[], recent_turns=[], counselor_identity=None, client_identity=None, ) self.assertNotIn("core_belief", state["client_profile"]) self.assertNotIn("automatic_thought", state["client_profile"]) self.assertNotIn("coping_strategy", state["client_profile"]) self.assertNotIn("temperament", state["client_profile"]) self.assertNotIn("speech_style", state["client_profile"]) self.assertEqual(state["client_profile"]["sore_spots"], []) self.assertEqual(state["client_profile"]["forbidden"], []) def test_appraisal_state_reads_coping_strategy_key_not_legacy_coping_key(self) -> None: state = client_affect.build_appraisal_state( client_profile={ "presenting": "", "history": "", "coping_strategy": "거리를 두고 관찰한다.", }, affect_state={}, affect_baseline={}, stage="라포", resistance=0.1, effective_openness=0.1, counselor_utterance="", recall_summary=None, pinned_facts=[], recent_turns=[], counselor_identity=None, client_identity=None, ) self.assertEqual(state["client_profile"]["coping_strategy"], "거리를 두고 관찰한다.") def test_appraisal_state_previous_feelings_use_word_buckets_not_numbers(self) -> None: state = client_affect.build_appraisal_state( client_profile={"presenting": "", "history": ""}, affect_state={"emotion_anxiety": 0.05, "emotion_trust": 0.95}, affect_baseline={}, stage="라포", resistance=0.1, effective_openness=0.1, counselor_utterance="", recall_summary=None, pinned_facts=[], recent_turns=[], counselor_identity=None, client_identity=None, ) self.assertEqual(state["previous_feelings"]["anxiety"], "absent") self.assertEqual(state["previous_feelings"]["trust"], "overwhelming") self.assertNotIn("0.05", str(state)) self.assertNotIn("0.95", str(state)) class ClientAffectRuntimeTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: store._sessions.clear() sessions._RECALL_CACHE.clear() async def asyncTearDown(self) -> None: store._sessions.clear() sessions._RECALL_CACHE.clear() def _route_session(self) -> tuple[InProcSession, Principal]: principal = Principal( user_id="00000000-0000-0000-0000-000000000333", role=Role.LEARNER, cohort_ids=[], email="learner@example.test", display_name="학습자", consent_at=1.0, profile_completed_at=1.0, ) sess = InProcSession( session_id="00000000-0000-0000-0000-000000000334", case_id="00000000-0000-0000-0000-000000000335", learner_id=principal.user_id, persona_code="P1", theory_mode="humanistic", persona=persona.P1, state=state_machine.init_state(params=persona.P1.openness_params()), ) store.put(sess) sessions._RECALL_CACHE[sess.session_id] = memory.RecallContext() return sess, principal async def _run_route_stream(self, sess: InProcSession, principal: Principal, engine: object) -> bytes: with ( patch.object(orchestrator.settings, "client_affect_provider", "jev"), patch.object(orchestrator.jev_client, "appraise", AsyncMock(return_value=_appraisal())), patch.object(sessions, "engine_client", engine), patch.object( sessions.rupture_scenario_director, "load_stored_scenario_context", AsyncMock(return_value=None), ), patch.object( sessions.turn_runtime.session_persistence, "append_client_turn_with_affect_trace", AsyncMock(return_value=True), ), patch.object(sessions, "_schedule_stream_turn_evaluation"), ): response = await sessions.stream_turn( sess.session_id, sessions.TurnRequest(text="조금 더 말해도 괜찮아요."), principal, ) return await _consume_event_source(response) async def test_generate_applies_once_before_request_with_internal_provenance(self) -> None: ctx = _context() engine = _GenerateEngine() audits: list[dict] = [] async def audit(payload: dict) -> None: audits.append(payload) with ( patch.object(orchestrator.settings, "client_affect_provider", "jev"), patch.object( orchestrator.jev_client, "appraise", AsyncMock(return_value=_appraisal(provider="OpenRouter", cost_usd=0.000019992)), ) as appraise, ): result = await orchestrator.run_turn_generate(ctx, engine, audit_hook=audit) # type: ignore[arg-type] self.assertEqual(appraise.await_count, 1) self.assertEqual(engine.calls, 1) self.assertGreater(result.state_after.affect_state["emotion_anxiety"], 0.0) self.assertEqual(engine.request.metadata["client_affect"]["provider"], "OpenRouter") self.assertEqual(engine.request.metadata["client_affect"]["tentative_dimensions"], []) self.assertEqual([payload["provider"] for payload in audits], ["OpenRouter", "fake-provider"]) self.assertEqual(audits[0]["cost_usd"], 0.000019992) self.assertNotIn("previous_emotions", str(engine.request.metadata)) async def test_appraisal_rebuild_preserves_theory_and_scenario_directives(self) -> None: ctx = _context() ctx.theory_mode = "cbt" cue = "고개를 숙이고 잠시 대답을 미룬다." ctx.scenario_directive = rupture_scenario_director.ScenarioDirective( scenario_id="g3-scenario-0123456789abcdef0123456789abcdef", rupture_type="withdrawal", behavior_cue=cue, turn_seq=ctx.state_after.turn_seq, opportunity_index=0, context_fingerprint="test-context", ) orchestrator._rebuild_persona_messages(ctx) engine = _GenerateEngine() with ( patch.object(orchestrator.settings, "client_affect_provider", "jev"), patch.object(orchestrator.jev_client, "appraise", AsyncMock(return_value=_appraisal())), ): await orchestrator.run_turn_generate(ctx, engine) # type: ignore[arg-type] contents = "\n".join(message.content for message in engine.request.messages) self.assertIn("[L3-T 이론모드: CBT]", contents) self.assertIn("자동적 사고, 감정, 행동의 연결", contents) self.assertIn(cue, contents) async def test_legacy_does_not_appraise_or_add_baseline_vector(self) -> None: ctx = _context() engine = _GenerateEngine() with ( patch.object(orchestrator.settings, "client_affect_provider", "legacy"), patch.object(orchestrator.jev_client, "appraise", AsyncMock()) as appraise, ): await orchestrator.run_turn_generate(ctx, engine) # type: ignore[arg-type] self.assertEqual(appraise.await_count, 0) self.assertFalse(any(key.startswith("emotion_") for key in ctx.state_after.affect_state)) async def test_crisis_stops_before_jev_and_generation(self) -> None: ctx = _context() ctx.crisis = guardrail.CrisisResult( kind=guardrail.CrisisKind.LEARNER_REAL, risk_level=3, escalate=True, ) engine = _GenerateEngine() with ( patch.object(orchestrator.settings, "client_affect_provider", "jev"), patch.object(orchestrator.jev_client, "appraise", AsyncMock()) as appraise, ): result = await orchestrator.run_turn_generate(ctx, engine) # type: ignore[arg-type] self.assertEqual(appraise.await_count, 0) self.assertEqual(engine.calls, 0) self.assertTrue(result.conversation_stopped) async def test_appraisal_failure_prevents_generation_without_mutating_original_state(self) -> None: ctx = _context() before = dict(ctx.state_before.affect_state) engine = _GenerateEngine() with ( patch.object(orchestrator.settings, "client_affect_provider", "jev"), patch.object( orchestrator.jev_client, "appraise", AsyncMock(side_effect=JevError("timeout")), ), ): with self.assertRaisesRegex(EngineError, "client_affect_timeout"): await orchestrator.run_turn_generate(ctx, engine) # type: ignore[arg-type] self.assertEqual(engine.calls, 0) self.assertEqual(ctx.state_before.affect_state, before) async def test_cancellation_propagates_from_appraisal(self) -> None: ctx = _context() engine = _GenerateEngine() with ( patch.object(orchestrator.settings, "client_affect_provider", "jev"), patch.object( orchestrator.jev_client, "appraise", AsyncMock(side_effect=asyncio.CancelledError()), ), ): with self.assertRaises(asyncio.CancelledError): await orchestrator.run_turn_generate(ctx, engine) # type: ignore[arg-type] async def test_stream_applies_once_before_engine_and_hides_affect_metadata_from_sse(self) -> None: ctx = _context() engine = _StreamEngine() with ( patch.object(orchestrator.settings, "client_affect_provider", "jev"), patch.object(orchestrator.jev_client, "appraise", AsyncMock(return_value=_appraisal())) as appraise, ): events = [ event async for event in orchestrator.run_turn_stream(ctx, engine) # type: ignore[arg-type] ] self.assertEqual(appraise.await_count, 1) self.assertEqual(engine.calls, 1) self.assertEqual(events[-1].event, "done") self.assertIn("client_affect", engine.request.metadata) self.assertNotIn("client_affect", events[-1].data) self.assertNotIn("accepted_dimensions", events[-1].data) async def test_stream_appraisal_failure_emits_error_without_generation(self) -> None: ctx = _context() engine = _StreamEngine() with ( patch.object(orchestrator.settings, "client_affect_provider", "jev"), patch.object( orchestrator.jev_client, "appraise", AsyncMock(side_effect=JevError("timeout")), ), ): events = [ event async for event in orchestrator.run_turn_stream(ctx, engine) # type: ignore[arg-type] ] self.assertEqual([(event.event, event.data) for event in events], [("error", {"detail": "client_affect_timeout"})]) self.assertEqual(engine.calls, 0) async def test_route_stream_error_after_appraisal_does_not_finalize_affect(self) -> None: sess, principal = self._route_session() before = dict(sess.state.affect_state) body = await self._run_route_stream(sess, principal, _InterruptedStreamEngine()) self.assertIn(b"gateway interrupted", body) self.assertEqual(sess.state.affect_state, before) self.assertEqual(sess.turns, []) async def test_route_stream_cancellation_after_appraisal_does_not_finalize_affect(self) -> None: sess, principal = self._route_session() before = dict(sess.state.affect_state) engine = _InterruptedStreamEngine(asyncio.CancelledError()) with self.assertRaises(asyncio.CancelledError): await self._run_route_stream(sess, principal, engine) self.assertEqual(sess.state.affect_state, before) self.assertEqual(sess.turns, []) async def test_route_stream_done_finalizes_jev_affect(self) -> None: sess, principal = self._route_session() body = await self._run_route_stream(sess, principal, _StreamEngine()) self.assertIn(b"done", body) self.assertIn("emotion_anxiety", sess.state.affect_state) self.assertEqual(len(sess.turns), 2) def test_public_end_state_retains_clinical_affect_but_hides_jev_vector(self) -> None: internal = { "stage": "라포", "affect": { "negative_affect": 0.7, "emotion_anxiety": 0.6, "emotion_trust": 0.2, }, } public = client_affect.public_end_state(internal) self.assertEqual(internal["affect"]["emotion_anxiety"], 0.6) self.assertEqual(public["affect"], {"negative_affect": 0.7}) async def test_end_route_preserves_internal_snapshot_and_hides_jev_vector(self) -> None: sess = InProcSession( session_id="00000000-0000-0000-0000-000000000222", case_id="00000000-0000-0000-0000-000000000223", learner_id="00000000-0000-0000-0000-000000000224", persona_code="P1", theory_mode="humanistic", persona=persona.P1, state=state_machine.SessionState( affect_state={"negative_affect": 0.7, "emotion_anxiety": 0.6}, ), ) carry = memory.CarryOver(end_state=sess.state.snapshot()) principal = Principal( user_id=sess.learner_id, role=Role.LEARNER, cohort_ids=[], email="learner@example.test", display_name="학습자", consent_at=1.0, profile_completed_at=1.0, ) with ( patch.object(sessions, "_load_session_or_404", AsyncMock(return_value=sess)), patch.object(sessions.memory, "make_carry_over", return_value=carry), patch.object(sessions, "_end_persisted_session", AsyncMock()), patch.object(sessions, "invalidate_session_context_cache"), patch.object(sessions.rupture_runtime, "schedule_session_scan"), ): response = await sessions.end_session(sess.session_id, principal) self.assertEqual(carry.end_state["affect"]["emotion_anxiety"], 0.6) self.assertEqual(response.end_state["affect"], {"negative_affect": 0.7}) def _judged_noul(probability: float, confidence: float | None = 0.7) -> NoulJudgment: return NoulJudgment(probability=probability, confidence=confidence) def _judged_choice( codes: list[str], choice: str, *, top_probability: float = 0.7, confidence: float | None = 0.6 ) -> ChoiceJudgment: rest = (1.0 - top_probability) / (len(codes) - 1) if len(codes) > 1 else 0.0 return ChoiceJudgment( choice=choice, probabilities={code: (top_probability if code == choice else rest) for code in codes}, confidence=confidence, ) _A_CODING_CODES = [ "reflection", "validation", "open_question", "closed_question", "clarification", "confrontation", "interpretation", "advice", "information", "self_disclosure", "topic_shift", "other", ] _COPING_CODES = ["nothing_asked", "manageable", "stretch", "overwhelming"] _BEHAVIOR_CODES = [ "disclose_more", "stay_with_feeling", "hold_core", "ask_back", "minimal_response", "shift_topic", "abstract_talk", "appease", "self_blame", "complain", "argue_back", "take_control", ] _DISPLAY_CODES = ["as_felt", "softened", "covered_by_agreement", "masked"] def _v2_appraisal( *, noul_overrides: dict[str, NoulJudgment] | None = None, choice_overrides: dict[str, ChoiceJudgment] | None = None, emotion_score: float = 0.5, emotion_confidence: float | None = 0.9, sore_spot_count: int = 0, ) -> AppraisalResult: nouls = {question_id: _judged_noul(0.5) for question_id in NOUL_QUESTION_IDS} nouls.update(noul_overrides or {}) choices = { "a_coping": _judged_choice(_COPING_CODES, "nothing_asked"), "a_move": _judged_choice(_A_CODING_CODES, "reflection"), "c_behavior": _judged_choice(_BEHAVIOR_CODES, "disclose_more"), "c_display": _judged_choice(_DISPLAY_CODES, "as_felt"), } choices.update(choice_overrides or {}) return AppraisalResult( emotions={ dimension: EmotionEstimate(score=emotion_score, confidence=emotion_confidence) for dimension in EMOTION_DIMENSIONS }, noul_judgments=nouls, choice_judgments=choices, sore_spot_count=sore_spot_count, model="jev-test", latency_ms=5, input_tokens=1, output_tokens=1, provider="typesafe", cost_usd=None, ) class ClientAffectV2CompositionTest(unittest.TestCase): def test_interpret_noul_thresholds(self) -> None: self.assertIs(client_affect.interpret_noul(NoulJudgment(probability=0.6)), True) self.assertIs(client_affect.interpret_noul(NoulJudgment(probability=0.4)), False) self.assertEqual(client_affect.interpret_noul(NoulJudgment(probability=0.5)), "uncertain") self.assertIsNone(client_affect.interpret_noul(None)) def test_interpret_choice_threshold(self) -> None: decided = ChoiceJudgment(choice="a", probabilities={"a": 0.45, "b": 0.55}) self.assertEqual(client_affect.interpret_choice(decided), "b") undecided = ChoiceJudgment(choice="a", probabilities={"a": 0.34, "b": 0.33, "c": 0.33}) self.assertEqual(client_affect.interpret_choice(undecided), "uncertain") self.assertIsNone(client_affect.interpret_choice(None)) def test_build_reaction_requires_confidence_at_least_035_and_uses_raw_score(self) -> None: appraisal = _v2_appraisal(emotion_score=0.6, emotion_confidence=0.35) included = client_affect.build_reaction(appraisal) self.assertEqual(set(included), set(EMOTION_DIMENSIONS)) self.assertEqual(included["anxiety"], 0.6) excluded = _v2_appraisal(emotion_score=0.6, emotion_confidence=0.34) self.assertEqual(client_affect.build_reaction(excluded), {}) def test_openness_gate_three_zones(self) -> None: closed = client_affect.build_expression_plan( _v2_appraisal( choice_overrides={"c_behavior": _judged_choice(_BEHAVIOR_CODES, "disclose_more")} ), effective_openness=0.1, ) self.assertEqual(closed.gated_behavior, "minimal_response") self.assertEqual(closed.gate_reason, "openness_closed") guarded = client_affect.build_expression_plan( _v2_appraisal( choice_overrides={"c_behavior": _judged_choice(_BEHAVIOR_CODES, "disclose_more")} ), effective_openness=0.3, ) self.assertEqual(guarded.gated_behavior, "hold_core") self.assertEqual(guarded.gate_reason, "openness_guarded") open_zone = client_affect.build_expression_plan( _v2_appraisal( choice_overrides={"c_behavior": _judged_choice(_BEHAVIOR_CODES, "disclose_more")} ), effective_openness=0.5, ) self.assertEqual(open_zone.gated_behavior, "disclose_more") self.assertIsNone(open_zone.gate_reason) def test_stance_derives_from_gated_behavior(self) -> None: plan = client_affect.build_expression_plan( _v2_appraisal( choice_overrides={"c_behavior": _judged_choice(_BEHAVIOR_CODES, "complain")} ), effective_openness=0.9, ) self.assertEqual(plan.stance, "push_back") def test_hidden_gap_requires_masking_display_and_strong_negative_reaction(self) -> None: masked_and_strong = client_affect.build_expression_plan( _v2_appraisal( choice_overrides={"c_display": _judged_choice(_DISPLAY_CODES, "masked")}, emotion_score=0.6, emotion_confidence=0.9, ), effective_openness=0.9, ) self.assertTrue(masked_and_strong.hidden_gap) as_felt_and_strong = client_affect.build_expression_plan( _v2_appraisal(emotion_score=0.6, emotion_confidence=0.9), effective_openness=0.9, ) self.assertFalse(as_felt_and_strong.hidden_gap) masked_but_weak = client_affect.build_expression_plan( _v2_appraisal( choice_overrides={"c_display": _judged_choice(_DISPLAY_CODES, "masked")}, emotion_score=0.2, emotion_confidence=0.9, ), effective_openness=0.9, ) self.assertFalse(masked_but_weak.hidden_gap) def test_experienced_phrases_priority_order_and_limit(self) -> None: appraisal = _v2_appraisal( noul_overrides={ "a_fact_conflict": _judged_noul(0.9), "a_judged": _judged_noul(0.9), "a_autonomy": _judged_noul(0.9), }, choice_overrides={ "a_coping": _judged_choice(_COPING_CODES, "overwhelming"), }, ) top2 = client_affect.experienced_phrases(appraisal, limit=2) self.assertEqual( top2, [ "자신의 사정과 다른 전제를 들었다고 느꼈다", "평가받거나 탓을 듣는 것처럼 느꼈다", ], ) top3 = client_affect.experienced_phrases(appraisal, limit=3) self.assertEqual(len(top3), 3) self.assertEqual(top3[2], "무엇을 할지 정해 주는 것 같아 압박을 느꼈다") def test_experienced_phrases_empty_when_all_uncertain_or_false(self) -> None: appraisal = _v2_appraisal() # 모든 noul probability=0.3 → false, choice는 experience에 안 걸림 self.assertEqual(client_affect.experienced_phrases(appraisal, limit=2), []) def _assert_no_numbers_english_codes_or_leak_markers(self, directive: str) -> None: self.assertIn("정서 연기 지시:", directive) self.assertNotIn("내부 상태", directive) for forbidden_code in ( "disclose_more", "hold_core", "as_felt", "withdrawal", "confrontation", "RUPTURE_TYPES", ): self.assertNotIn(forbidden_code, directive) # 고정 문구("1~3문장")를 제외하면 확률·점수 같은 소수 숫자가 없어야 한다. self.assertNotRegex(directive.replace("1~3문장", ""), r"\d") def test_render_affect_directive_v2_has_no_numbers_english_codes_or_leak_markers(self) -> None: appraisal = _v2_appraisal( noul_overrides={"a_judged": _judged_noul(0.9)}, emotion_score=0.8, emotion_confidence=0.9, ) expression = client_affect.build_expression_plan(appraisal, effective_openness=0.9) directive = client_affect.render_affect_directive_v2( appraisal, expression, affect_state_after={f"emotion_{d}": 0.8 for d in EMOTION_DIMENSIONS}, affect_baseline={}, ) self._assert_no_numbers_english_codes_or_leak_markers(directive) def test_render_affect_directive_v2_falls_back_to_v2_common_rules_when_everything_uncertain( self, ) -> None: appraisal = _v2_appraisal( noul_overrides={question_id: _judged_noul(0.5) for question_id in NOUL_QUESTION_IDS}, choice_overrides={ "c_behavior": ChoiceJudgment( choice="disclose_more", probabilities={code: 1.0 / len(_BEHAVIOR_CODES) for code in _BEHAVIOR_CODES}, ), "c_display": ChoiceJudgment( choice="as_felt", probabilities={code: 1.0 / len(_DISPLAY_CODES) for code in _DISPLAY_CODES}, ), }, emotion_score=0.5, emotion_confidence=0.2, ) expression = client_affect.build_expression_plan(appraisal, effective_openness=0.9) directive = client_affect.render_affect_directive_v2( appraisal, expression, affect_state_after={}, affect_baseline={}, ) # v1 문장("...숫자·내부 상태·평가 정답은...")이 아니라 v2 공통 규칙 문장만 남는다. self.assertEqual( directive, "정서 연기 지시:\n- 감정 이름을 나열하거나 분석하듯 설명하지 말고 " "말투·선택·침묵·주저함으로만 드러낸다. 숫자·분석 내용·평가 정답은 절대 말하지 않는다. " "상담자 역할로 바뀌거나 조언하지 않으며, 부정 감정을 즉시 해소하려 하지 않는다. " "응답은 기본적으로 1~3문장으로 하고, 꼭 필요할 때만 더 길게 말한다.", ) self._assert_no_numbers_english_codes_or_leak_markers(directive) def test_build_inner_reaction_uses_fixed_phrases_and_marks_uncertain_as_absent(self) -> None: appraisal = _v2_appraisal( noul_overrides={"a_judged": _judged_noul(0.9)}, choice_overrides={ "c_behavior": _judged_choice(_BEHAVIOR_CODES, "complain"), "c_display": _judged_choice(_DISPLAY_CODES, "masked"), }, emotion_score=0.8, emotion_confidence=0.9, ) expression = client_affect.build_expression_plan(appraisal, effective_openness=0.9) reaction = client_affect.build_inner_reaction(appraisal, expression, turn_seq=3) self.assertEqual(reaction.schema_version, 1) self.assertEqual(reaction.turn_seq, 3) self.assertIn("평가받거나 탓을 듣는 것처럼 느꼈다", reaction.experienced) self.assertTrue(all(feeling.label and feeling.intensity for feeling in reaction.feelings)) self.assertEqual(reaction.stance.code, "push_back") self.assertEqual(reaction.stance.label, "맞서거나 반박했다") self.assertEqual(reaction.display.code, "masked") self.assertTrue(reaction.hidden_gap) uncertain_appraisal = _v2_appraisal( noul_overrides={question_id: _judged_noul(0.5) for question_id in NOUL_QUESTION_IDS}, choice_overrides={ "c_behavior": ChoiceJudgment( choice="disclose_more", probabilities={code: 1.0 / len(_BEHAVIOR_CODES) for code in _BEHAVIOR_CODES}, ), "c_display": ChoiceJudgment( choice="as_felt", probabilities={code: 1.0 / len(_DISPLAY_CODES) for code in _DISPLAY_CODES}, ), }, emotion_confidence=0.2, ) uncertain_expression = client_affect.build_expression_plan( uncertain_appraisal, effective_openness=0.9 ) uncertain_reaction = client_affect.build_inner_reaction( uncertain_appraisal, uncertain_expression, turn_seq=1 ) self.assertEqual(uncertain_reaction.experienced, ()) self.assertEqual(uncertain_reaction.feelings, ()) self.assertIsNone(uncertain_reaction.stance) self.assertIsNone(uncertain_reaction.display) self.assertFalse(uncertain_reaction.hidden_gap)