29c406d89f
상담자 발화 판정(A)·감정(B)·표현(C) 20문항 질문 세트, 감쇠 없는 이번 턴 반응과 비대칭 기분 전이, 개방도 게이트로 생성 지시를 만들고 ccd.coping_strategy 전달 누락을 고친다. trace v2와 고정 문구 속마음 요약(migration 24, AI 경로 차단 RLS)을 같은 트랜잭션에 저장하고 피드백 정책이 켜진 경우에만 done·TurnResponse·음성 reply·리뷰로 노출한다. 회기 화면 속마음 보기 토글, 리뷰 접힘 블록, 관리자 감정 관측 v2 표시를 추가한다.
1147 lines
46 KiB
Python
1147 lines
46 KiB
Python
"""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)
|