"""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, EMOTION_DIMENSIONS, EmotionEstimate, JevError from .store import InProcSession, store 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 }, 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( affect_baseline={}, affect_state={}, persona_context={"core_belief": "서연은 가치가 없다고 느낀다."}, 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), {"persona", "memory", "recent_turns", "counselor_utterance", "previous_emotions", "current_state"}) self.assertEqual(len(state["memory"]["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( affect_baseline={}, affect_state={}, persona_context={}, 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_dynamic_mapping_keys_and_whitelists_baseline(self) -> None: state = client_affect.build_appraisal_state( affect_baseline={"anxiety": 0.4, "010-1234-5678": 0.9}, affect_state={}, persona_context={ "김상담": { "010-1234-5678": "서연에게는 비밀로 해 달라는 지시가 있다." } }, resistance=0.5, effective_openness=0.3, counselor_utterance="괜찮아요.", recall_summary=None, pinned_facts=[], recent_turns=[], counselor_identity="김상담", client_identity="서연", ) rendered = str(state) self.assertNotIn("김상담", rendered) self.assertNotIn("010-1234-5678", rendered) self.assertEqual(state["persona"]["affect_baseline"], {"anxiety": 0.4}) 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})