전수 E2E 순회·소유자 결정 6건 구현·디자인 감사 반영

- 전수 순회: IA 전 라우트 412개 기능 인벤토리를 체크리스트 백로그로 관리
  (docs/ops/e2e-full-sweep-2026-07-27.md), 신규 full-sweep 스펙 10파일 추가.
  RED→GREEN으로 결함 12건 수정: 관리자 무한 렌더 프리즈(TanStack autoReset
  루프), 복수 코호트 저장 유실, 페르소나 보관 503(SQL 컬럼 모호성), PII 과잉
  마스킹, /admin/ai watchdog 오탐 오버레이, 모바일 겹침 2건, 설정 스크롤
  스파이, 온보딩 전화번호 무검증, 리뷰 조사·난도 라벨, pending 피드백 등.
- 소유자 결정 구현: 설정 아바타 변경, 동의 철회·재동의 전체 흐름, 신규
  학습자 기초 우선 추천, 학생 분석 테이블 가상화(@tanstack/react-virtual),
  저작 모드 죽은 레일 정리, 감정 밸런스 타임라인 차트(deep turn_valence +
  결정론 파생 폴백).
- 디자인 감사(142차): 라이트 팔레트 AA 대비, 다크 토큰 별칭 통일, 미정의
  CSS 변수 정리, 한글 keep-all 전역화, 탭 타깃 24px, LCP preconnect.
- 검증: npm run e2e 병렬 432 수집 GREEN + 직렬 49/49 exit 0, 백엔드 pytest
  421, gateway 29, typecheck/build/design-ssot/dead-code/중복 게이트 통과,
  layout-visual-gate 15/15, session-layout 8/8. 상세는 SSOT 대시보드
  142~144차 노트.
This commit is contained in:
Yun Chan 2026-07-27 14:25:24 +09:00
parent 0ae94499f2
commit 4511383cd9
55 changed files with 9333 additions and 573 deletions

View file

@ -291,11 +291,18 @@ def _row_value(row, key: str, default=None):
def _cohort_ids(cohort: str | None) -> list[str]:
return [cohort] if cohort else []
"""단일 TEXT 컬럼(콤마 직렬화)을 cohort id 리스트로 복원한다."""
if not cohort:
return []
return [part for part in (piece.strip() for piece in cohort.split(",")) if part]
def _cohort_value(cohort_ids: list[str] | None) -> str | None:
return (cohort_ids or [None])[0]
"""cohort id 리스트를 콤마 직렬화해 저장한다. 첫 항목만 남기던 유실 버그 수정."""
cleaned = [part for part in ((item or "").strip() for item in cohort_ids or []) if part]
if not cleaned:
return None
return ",".join(cleaned)
async def _runtime_tables_ready(conn) -> bool:

View file

@ -36,6 +36,12 @@ _REVIEW_COLUMNS = """
source_provenance, is_synthetic, created_at, approved_at
"""
# UPDATE ... FROM target 처럼 다른 테이블이 스코프에 있는 RETURNING에서는
# 무한정 `code`가 target.code와 모호해진다(AmbiguousColumnError → 보관 503).
_REVIEW_COLUMNS_CARD_QUALIFIED = ", ".join(
f"card.{column.strip()}" for column in _REVIEW_COLUMNS.split(",")
)
_PERSONA_STATUSES = {"draft", "review", "approved", "archived"}
_REVIEW_QUEUE_STATUSES = ("draft", "review")
PersonaReviewAction = Literal["approve", "reject"]
@ -818,7 +824,7 @@ async def archive_persona_family(
FROM target
WHERE upper(card.code) = upper(target.code)
AND card.status <> 'archived'
RETURNING {_REVIEW_COLUMNS}
RETURNING {_REVIEW_COLUMNS_CARD_QUALIFIED}
)
SELECT {_REVIEW_COLUMNS}
FROM archived

View file

@ -7,8 +7,10 @@ import time
from pathlib import Path
from typing import Literal
import re
from fastapi import APIRouter, File, HTTPException, Query, UploadFile, status
from pydantic import BaseModel, Field, model_validator
from pydantic import BaseModel, Field, field_validator, model_validator
from ..auth_types import RoleName
from ..auth_sessions import (
@ -128,6 +130,20 @@ class UserProfileResponse(BaseModel):
onboarding_required: bool = True
# 연락처: 선행 + 허용, 숫자·하이픈·공백·괄호·점만, 숫자 9~15자리.
_PHONE_ALLOWED_RE = re.compile(r"^\+?[0-9()\-\s.]+$")
def _validate_phone_format(value: str) -> str:
trimmed = value.strip()
digits = re.sub(r"\D", "", trimmed)
if not _PHONE_ALLOWED_RE.fullmatch(trimmed) or not 9 <= len(digits) <= 15:
raise ValueError(
"연락처는 숫자 9~15자리의 전화번호 형식이어야 합니다 (예: 010-1234-5678)."
)
return trimmed
class UserProfilePatch(BaseModel):
display_name: str | None = Field(default=None, min_length=1, max_length=80)
affiliation: str | None = Field(default=None, max_length=120)
@ -140,6 +156,13 @@ class UserProfilePatch(BaseModel):
self_introduction: str | None = Field(default=None, max_length=600)
avatar_url: str | None = Field(default=None, max_length=500)
@field_validator("phone")
@classmethod
def _check_phone(cls, value: str | None) -> str | None:
if value is None:
return value
return _validate_phone_format(value)
class OnboardingRequest(BaseModel):
legal_name: str = Field(..., min_length=1, max_length=80)
@ -154,6 +177,11 @@ class OnboardingRequest(BaseModel):
terms_accepted: bool
privacy_accepted: bool
@field_validator("phone")
@classmethod
def _check_phone(cls, value: str) -> str:
return _validate_phone_format(value)
class AvatarUploadResponse(BaseModel):
avatar_url: str

View file

@ -264,6 +264,13 @@ class TechniqueDistribution(BaseModel):
underused: list[str] = Field(default_factory=list) # 과소/미사용 군집
class TurnValencePoint(BaseModel):
"""deep-loop 턴별 내담자 정서가 — 축어록 seq(1-based) 지목 + v(1~+1)."""
seq: int # 마스킹 축어록의 1-based 턴 번호
v: float # 정서가(1 매우 부정 ~ +1 매우 긍정)
class SessionEvaluation(BaseModel):
"""deep-loop 회기말/단계전환 정밀 평가 결과.
@ -282,6 +289,9 @@ class SessionEvaluation(BaseModel):
supervisor_rationale: Optional[str] = None # CommentKind.RATIONALE 종합
supervisor_critique: Optional[str] = None # CommentKind.CRITIQUE 종합
alternative_utterances: list[str] = Field(default_factory=list) # 대안 발화 제시
turn_valence: list[TurnValencePoint] = Field(
default_factory=list
) # 내담자 발화별 정서가(리뷰 감정 밸런스 차트 원천)
theory_mode: Optional[str] = None
error: Optional[str] = None
@ -378,6 +388,19 @@ def _deep_schema() -> dict[str, Any]:
"supervisor_rationale": {"type": "string"},
"supervisor_critique": {"type": "string"},
"alternative_utterances": {"type": "array", "items": {"type": "string"}},
# 선택 필드 — 내담자 발화별 정서가(리뷰 감정 밸런스 차트 원천).
"turn_valence": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"properties": {
"seq": {"type": "integer", "minimum": 1},
"v": {"type": "number", "minimum": -1, "maximum": 1},
},
"required": ["seq", "v"],
},
},
},
"required": ["strengths", "improvements"],
}
@ -583,7 +606,10 @@ def build_deep_messages(
"- intent_deviations: '의도와 다른 부분' 전부 {dimension, expected, actual, severity}로.\n"
"- supervisor_rationale: 회기 전반에서 적절했던 개입의 근거(rationale) 종합.\n"
"- supervisor_critique: 과도/부족/평가적 시각 등 주의점(critique) 종합.\n"
"- alternative_utterances: 핵심 장면에 더 나은 대안 상담자 발화 1~3개."
"- alternative_utterances: 핵심 장면에 더 나은 대안 상담자 발화 1~3개.\n"
"- turn_valence: *내담자* 발화 각각의 정서가를 {seq, v}로 산출. seq 는 축어록의 "
"1-based 턴 번호, v 는 1(매우 부정)~+1(매우 긍정). 근거 없는 극단값을 피하고 "
"불확실하면 0 근처로."
),
]
)
@ -854,6 +880,18 @@ async def evaluate_session(
dev = _parse_intent_deviation(d)
if dev is not None:
base.intent_deviations.append(dev)
for item in payload.get("turn_valence") or []:
if not isinstance(item, dict):
continue
seq = item.get("seq")
v = item.get("v")
if not isinstance(seq, int) or isinstance(seq, bool) or seq < 1:
continue
if not isinstance(v, (int, float)) or isinstance(v, bool):
continue
base.turn_valence.append(
TurnValencePoint(seq=seq, v=max(-1.0, min(1.0, float(v))))
)
_evaluator_cache_put(cache_key, base.model_dump())
return base
@ -888,6 +926,7 @@ __all__ = [
"ClientStateRead",
"TurnEvaluation",
"TechniqueDistribution",
"TurnValencePoint",
"SessionEvaluation",
"aggregate_distribution",
"evaluate_turn",

View file

@ -46,6 +46,10 @@ _KOREAN_SURNAME_CHARS = (
_KOREAN_FULL_NAME = rf"[{_KOREAN_SURNAME_CHARS}][가-힣]{{1,3}}"
_KOREAN_FULL_NAME_BEFORE_SUFFIX = rf"[{_KOREAN_SURNAME_CHARS}][가-힣]{{1,3}}?"
_KOREAN_CONTEXTLESS_NAME = rf"[{_KOREAN_SURNAME_CHARS}][가-힣]{{2,3}}"
# 인명 마지막 글자로 사실상 쓰이지 않는 용언 활용/명사화 꼬리 글자. 문맥 단서가
# 없는 인명 패턴이 "유의점은·방치되는·마무리했을·연결감과·표현함을·어지러움은"
# 같은 일반 단어를 이름으로 오탐해 평가 문장을 훼손하는 것을 막는다(정밀도 가드).
_KOREAN_NAME_TAIL_GUARD = "(?<![했됐되된될것듯점음움됨함감])"
_KOREAN_NAME_STOPWORDS = {
"연락",
"연락처",
@ -109,7 +113,7 @@ _PII_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
"NAME",
re.compile(
r"(?P<prefix>(?:(?:저는|나는|제가|내가)\s*)?)"
rf"(?P<value>{_KOREAN_FULL_NAME_BEFORE_SUFFIX})"
rf"(?P<value>{_KOREAN_FULL_NAME_BEFORE_SUFFIX}){_KOREAN_NAME_TAIL_GUARD}"
r"(?P<suffix>\s*(?:입니다|이에요|예요|이고|이고요))"
r"(?=$|[\s,.;!?。])"
),
@ -128,7 +132,7 @@ _PII_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
(
"NAME",
re.compile(
rf"(?<![가-힣])(?P<value>{_KOREAN_CONTEXTLESS_NAME})"
rf"(?<![가-힣])(?P<value>{_KOREAN_CONTEXTLESS_NAME}){_KOREAN_NAME_TAIL_GUARD}"
r"(?P<suffix>(?:은|는|이|가|을|를|와|과|에게|한테|라고|이라는))"
),
),

View file

@ -431,7 +431,11 @@ async def _session_review_payload_and_recipients(
OR lower(u.email) = ANY($1::text[])
OR (
u.role = 'instructor'
AND ($2 = '' OR u.cohort = $2)
AND (
$2 = ''
OR string_to_array(COALESCE(u.cohort, ''), ',')
&& string_to_array($2, ',')
)
)
)
AND COALESCE((p.notifications->>'learner_progress')::boolean, true)

View file

@ -750,6 +750,24 @@ def _client_name(raw: str) -> str:
return name or raw.strip() or "내담자"
_DIFFICULTY_KO = {"easy": "기초", "moderate": "중간", "hard": "고난도"}
def _persona_label(code: str, difficulty: str) -> str:
return f"{code} · {_DIFFICULTY_KO.get(difficulty, difficulty)}"
def _josa_wa_gwa(name: str) -> str:
"""이름 마지막 글자의 받침 유무로 와/과를 고른다. 한글이 아니면 병기한다."""
tail = name[-1] if name else ""
if "" <= tail <= "":
has_final = (ord(tail) - 0xAC00) % 28 != 0
if has_final:
return ""
return ""
return "와(과)"
def _review_summary(*, client_name: str, reached_phase: StageLabel, turns: list[ReviewTurn]) -> str:
if not turns:
return (
@ -760,7 +778,7 @@ def _review_summary(*, client_name: str, reached_phase: StageLabel, turns: list[
client_count = sum(1 for turn in turns if turn.speaker == "client")
return (
f"이 리뷰는 현재 세션에 저장된 실제 축어록 {len(turns)}개를 기반으로 합니다. "
f"{client_name}의 회기는 {reached_phase} 단계까지 진행되었고, "
f"{client_name}{_josa_wa_gwa(client_name)}의 회기는 {reached_phase} 단계까지 진행되었고, "
f"학습자 발화 {learner_count}개와 내담자 응답 {client_count}개가 기록되었습니다. "
"평가 AI 또는 교수자 코멘트가 아직 생성되지 않은 항목은 빈 상태로 남겨 둡니다."
)
@ -1405,6 +1423,185 @@ def _review_nonverbal_events(turn: TurnRecord) -> list[ReviewNonverbalEvent]:
return events
# ─ 감정 밸런스 타임라인(valence) 파생 — 순수 함수 ──────────────────────────────
_VALENCE_MAX_POINTS = 10
# client_state_read 코드 → 정서가 극성(taxonomy.ClientState 코드 기준 휴리스틱 맵).
_CLIENT_STATE_VALENCE: dict[str, float] = {
# 음의 극성 — 방어·위축·위기 신호
"involuntary": -0.5,
"defensive": -0.7,
"suicidal_ideation_admit": -0.9,
"negative_self_perception": -0.7,
"conflicted": -0.4,
"lack_of_confidence": -0.4,
"compliant_surface": -0.2,
"externalizing": -0.3,
"apparent_competence": -0.2,
"active_passivity": -0.3,
"self_harm_disclosure": -0.9,
"somatic_complaint": -0.4,
"affect_masking": -0.3,
"focus_drift_fusion": -0.3,
# 양의 극성 — 개방·접촉·진전 신호
"affect_contact": 0.6,
"thought_organizing": 0.5,
"responds_to_exploration": 0.5,
"expresses_plan": 0.7,
"defense_loosening": 0.6,
"seeks_guidance": 0.2,
}
def _clamp_valence(value: float) -> float:
return max(-1.0, min(1.0, value))
def _valence_t(
created_at: float, first_turn_ts: float, duration_seconds: float
) -> float:
"""턴 시각 → 회기 진행률(0~1 클램프)."""
if duration_seconds <= 0:
return 0.0
return max(0.0, min(1.0, (created_at - first_turn_ts) / duration_seconds))
def _finalize_valence_points(
points: list[ReviewValencePoint],
) -> list[ReviewValencePoint]:
"""2개 미만이면 빈 배열(차트 빈 상태), 10개 초과면 균등 리샘플(양 끝점 유지)."""
if len(points) < 2:
return []
if len(points) <= _VALENCE_MAX_POINTS:
return points
last = len(points) - 1
indices: list[int] = []
for i in range(_VALENCE_MAX_POINTS):
idx = round(i * last / (_VALENCE_MAX_POINTS - 1))
if not indices or idx != indices[-1]:
indices.append(idx)
return [points[i] for i in indices]
def counselor_baseline_points(
turns: list[TurnRecord],
*,
first_turn_ts: float,
duration_seconds: float,
) -> list[ReviewValencePoint]:
"""학습자 턴 rapport_signal 누적 이동평균 → 상담자 기준선 궤적."""
points: list[ReviewValencePoint] = []
total = 0.0
count = 0
for turn in turns:
if turn.speaker != "counselor":
continue
ev = session_metrics.turn_eval(turn)
if ev is None or str(ev.get("error") or "").strip():
continue
rapport = session_metrics.turn_rapport(ev)
if rapport is None:
continue
count += 1
total += rapport
points.append(
ReviewValencePoint(
t=_valence_t(turn.created_at, first_turn_ts, duration_seconds),
v=_clamp_valence(total / count),
)
)
return _finalize_valence_points(points)
def _fallback_client_valence_points(
turns: list[TurnRecord],
*,
first_turn_ts: float,
duration_seconds: float,
) -> list[ReviewValencePoint]:
"""폴백 — 상담자 턴 평가의 client_state_read 극성과 appropriateness(0~1) 결합."""
points: list[ReviewValencePoint] = []
for turn in turns:
if turn.speaker != "counselor":
continue
ev = session_metrics.turn_eval(turn)
if ev is None or str(ev.get("error") or "").strip():
continue
polarities: list[float] = []
for state in ev.get("client_state_read") or []:
code = state.get("code") if isinstance(state, dict) else state
mapped = _CLIENT_STATE_VALENCE.get(str(code or "").strip())
if mapped is not None:
polarities.append(mapped)
score01 = session_metrics.turn_score(ev)
parts: list[float] = []
if polarities:
parts.append(0.7 * (sum(polarities) / len(polarities)))
if score01 is not None:
parts.append(0.3 * (score01 * 2.0 - 1.0))
if not parts:
continue
points.append(
ReviewValencePoint(
t=_valence_t(turn.created_at, first_turn_ts, duration_seconds),
v=_clamp_valence(sum(parts)),
)
)
return points
def client_valence_points(
turns: list[TurnRecord],
evaluation_payload: dict[str, object],
*,
first_turn_ts: float,
duration_seconds: float,
) -> list[ReviewValencePoint]:
"""내담자 정서가 궤적 — deep 평가 turn_valence 우선, 없으면 턴 평가 기반 폴백."""
raw = (
evaluation_payload.get("turn_valence")
if isinstance(evaluation_payload, dict)
else None
)
points: list[ReviewValencePoint] = []
if isinstance(raw, list):
for item in raw:
if not isinstance(item, dict):
continue
seq = item.get("seq")
v = item.get("v")
if isinstance(seq, bool) or not isinstance(seq, int):
continue
if isinstance(v, bool) or not isinstance(v, (int, float)):
continue
index = seq - 1
if index < 0 or index >= len(turns):
continue
points.append(
ReviewValencePoint(
t=_valence_t(
turns[index].created_at, first_turn_ts, duration_seconds
),
v=_clamp_valence(float(v)),
)
)
points.sort(key=lambda point: point.t)
if not points:
points = _fallback_client_valence_points(
turns, first_turn_ts=first_turn_ts, duration_seconds=duration_seconds
)
return _finalize_valence_points(points)
def _valence_axis(
duration_seconds: int, *, has_points: bool, fallback: list[str]
) -> list[str]:
"""포인트가 있으면 시간 라벨 4개(0~회기말 균등), 없으면 기존 axis 유지."""
if not has_points or duration_seconds <= 0:
return fallback
return [_offset_label(duration_seconds * i / 3) for i in range(4)]
def build_session_review(read_input: SessionReviewReadInput) -> SessionReviewResponse:
sess = read_input.session
visible_turns = learner_visible_turns(sess)
@ -1455,6 +1652,27 @@ def build_session_review(read_input: SessionReviewReadInput) -> SessionReviewRes
)
)
# 감정 밸런스 타임라인 — 학습자 기준선 + 내담자 정서가(비공개 턴 존재 시 비산출)
counselor_baseline: list[ReviewValencePoint] = []
client_valence: list[ReviewValencePoint] = []
if not hidden_turns:
counselor_baseline = counselor_baseline_points(
visible_turns,
first_turn_ts=first_turn_ts,
duration_seconds=duration_seconds,
)
client_valence = client_valence_points(
visible_turns,
evaluation_payload,
first_turn_ts=first_turn_ts,
duration_seconds=duration_seconds,
)
valence_axis = _valence_axis(
duration_seconds,
has_points=bool(counselor_baseline or client_valence),
fallback=axis,
)
if not turns:
session_signal = "기록 없음"
elif sess.ended:
@ -1533,7 +1751,7 @@ def build_session_review(read_input: SessionReviewReadInput) -> SessionReviewRes
client=ReviewClient(
name=client_name,
initial=client_initial,
persona=f"{sess.persona_code} · {sess.persona.difficulty}",
persona=_persona_label(sess.persona_code, str(sess.persona.difficulty)),
),
date=datetime.fromtimestamp(sess.created_at).strftime("%Y-%m-%d"),
durationLabel=_duration_label(duration_seconds),
@ -1545,9 +1763,9 @@ def build_session_review(read_input: SessionReviewReadInput) -> SessionReviewRes
summary=summary,
phases=_phase_segments(stage_labels),
phaseAxis=axis,
valenceAxis=axis,
clientValence=[],
counselorBaseline=[],
valenceAxis=valence_axis,
clientValence=client_valence,
counselorBaseline=counselor_baseline,
turns=turns,
rubric=rubric,
goodMoments=good_moments,

View file

@ -21,6 +21,26 @@ class _Acquire:
return None
class CohortSerializationTest(unittest.TestCase):
"""2026-07-27 전수 순회 결함 #20: 서버가 cohort_ids 첫 항목만 저장하던 유실 회귀 가드."""
def test_multi_cohort_round_trip_preserves_every_id(self) -> None:
from .auth_sessions import _cohort_ids, _cohort_value
self.assertEqual(_cohort_value(["co-a", "co-b"]), "co-a,co-b")
self.assertEqual(_cohort_ids("co-a,co-b"), ["co-a", "co-b"])
self.assertEqual(_cohort_ids(_cohort_value(["x-1", "x-2", "x-3"])), ["x-1", "x-2", "x-3"])
def test_cohort_value_trims_and_drops_empty_entries(self) -> None:
from .auth_sessions import _cohort_ids, _cohort_value
self.assertEqual(_cohort_value([" co-a ", "", " ", "co-b"]), "co-a,co-b")
self.assertIsNone(_cohort_value([]))
self.assertIsNone(_cohort_value(None))
self.assertEqual(_cohort_ids(None), [])
self.assertEqual(_cohort_ids(" co-a , ,co-b "), ["co-a", "co-b"])
class AdminOpsTest(unittest.IsolatedAsyncioTestCase):
async def test_usage_from_database_orders_by_aggregated_token_sum(self) -> None:
case = self

View file

@ -214,6 +214,21 @@ class OrchestratorMaskingGateTest(unittest.IsolatedAsyncioTestCase):
)
self.assertNotIn("NAME", masked.entities)
def test_mask_pii_does_not_mask_evaluator_abstract_words_as_names(self) -> None:
# 실제 교수자 리뷰 화면에서 관측된 오탐 문형: 성씨 문자로 시작하는 일반
# 명사/용언(유의점·방치되는·마무리했을·연결감·표현함·어지러움)이
# 조사와 결합하면 문맥-없는 인명 패턴에 걸려 평가 문장이 훼손된다.
text = (
"유의점은 세 가지다. 정서 신호가 방치되는 것을 피하고, 회기를 "
"마무리했을 때 연결감과 작업동맹을 지키며, 감정을 표현함을 존중한다. "
"어지러움은 신체 신호로 다룬다."
)
masked = guardrail.mask_pii(text)
self.assertEqual(masked.text_masked, text)
self.assertNotIn("NAME", masked.entities)
def test_prepare_turn_masks_korean_pii_from_engine_messages(self) -> None:
ctx = orchestrator.prepare_turn(
session_id="masking-session",

View file

@ -0,0 +1,50 @@
"""리뷰 요약 문장의 한국어 조사(와/과) 선택 회귀 테스트.
2026-07-27 전수 순회에서 발견: 받침 있는 내담자 이름(하늘) 뒤에도 "와의"
붙어 "하늘와의 회기는"으로 표기됐다. 마지막 글자 받침 유무에 따라 /과를
골라야 한다.
"""
from __future__ import annotations
import unittest
from .session_read_model import ReviewTurn, _review_summary
def _turns() -> list[ReviewTurn]:
return [
ReviewTurn(id="t1", ts="0:00", speaker="learner", who="", text="안녕하세요."),
ReviewTurn(id="t2", ts="0:01", speaker="client", who="하늘", text="...네."),
]
class ReviewPersonaLabelTest(unittest.TestCase):
def test_difficulty_is_localized_to_korean(self) -> None:
from .session_read_model import _persona_label
self.assertEqual(_persona_label("P4", "easy"), "P4 · 기초")
self.assertEqual(_persona_label("P2", "moderate"), "P2 · 중간")
self.assertEqual(_persona_label("P1", "hard"), "P1 · 고난도")
# 알 수 없는 값은 원문 유지(빈 라벨로 숨기지 않는다)
self.assertEqual(_persona_label("P9", "custom"), "P9 · custom")
class ReviewSummaryJosaTest(unittest.TestCase):
def test_name_with_final_consonant_uses_gwa(self) -> None:
summary = _review_summary(client_name="하늘", reached_phase="라포", turns=_turns())
self.assertIn("하늘과의 회기는", summary)
self.assertNotIn("하늘와의", summary)
def test_name_without_final_consonant_uses_wa(self) -> None:
summary = _review_summary(client_name="지우", reached_phase="라포", turns=_turns())
self.assertIn("지우와의 회기는", summary)
self.assertNotIn("지우과의", summary)
def test_non_hangul_tail_falls_back_to_gwa_slash_wa(self) -> None:
summary = _review_summary(client_name="P4", reached_phase="라포", turns=_turns())
self.assertIn("P4와(과)의 회기는", summary)
if __name__ == "__main__":
unittest.main()

View file

@ -0,0 +1,287 @@
"""감정 밸런스 타임라인(리뷰 valence 차트) 파생 로직 회귀 테스트."""
from __future__ import annotations
import unittest
from typing import Any
from . import session_read_model
from .engine_client import GenerateResponse
from .services import evaluator, persona as persona_service, state_machine
from .store import InProcSession, TurnRecord
def _session(*, created_at: float = 1_000.0) -> InProcSession:
card = persona_service.P1
return InProcSession(
session_id="review-valence-session",
case_id="review-valence-case",
learner_id="00000000-0000-0000-0000-000000000201",
persona_code=card.code,
theory_mode="humanistic",
persona=card,
state=state_machine.SessionState(
resistance=card.base_resistance(),
ideation_stage=card.ideation_baseline(),
),
created_at=created_at,
)
def _turn(
seq: int,
speaker: str,
created_at: float,
evaluation: dict[str, Any] | None = None,
) -> TurnRecord:
return TurnRecord(
turn_seq=seq,
speaker=speaker,
stage="rapport",
text=f"발화 {seq}",
text_masked=f"발화 {seq}",
created_at=created_at,
evaluation=evaluation,
)
class CounselorBaselineTest(unittest.TestCase):
def test_cumulative_average_and_t_normalization(self) -> None:
base = 1_000.0
turns = [
_turn(1, "counselor", base, {"rapport_signal": 0.5}),
_turn(2, "client", base + 10.0),
_turn(3, "counselor", base + 30.0, {"rapport_signal": -0.5}),
_turn(4, "client", base + 40.0),
_turn(5, "counselor", base + 60.0, {"rapport_signal": 0.5}),
]
points = session_read_model.counselor_baseline_points(
turns, first_turn_ts=base, duration_seconds=120
)
self.assertEqual(len(points), 3)
self.assertAlmostEqual(points[0].v, 0.5)
self.assertAlmostEqual(points[1].v, 0.0)
self.assertAlmostEqual(points[2].v, 0.5 / 3.0)
self.assertAlmostEqual(points[0].t, 0.0)
self.assertAlmostEqual(points[1].t, 0.25)
self.assertAlmostEqual(points[2].t, 0.5)
def test_resamples_to_at_most_ten_points(self) -> None:
base = 1_000.0
turns = [
_turn(i + 1, "counselor", base + i * 10.0, {"rapport_signal": 0.1})
for i in range(12)
]
points = session_read_model.counselor_baseline_points(
turns, first_turn_ts=base, duration_seconds=110
)
self.assertEqual(len(points), 10)
# 균등 리샘플이어도 양 끝점은 유지된다.
self.assertAlmostEqual(points[0].t, 0.0)
self.assertAlmostEqual(points[-1].t, 1.0)
# t 는 단조 증가·클램프 범위 내.
for prev, cur in zip(points, points[1:]):
self.assertLessEqual(prev.t, cur.t)
for point in points:
self.assertGreaterEqual(point.v, -1.0)
self.assertLessEqual(point.v, 1.0)
def test_fewer_than_two_points_returns_empty(self) -> None:
base = 1_000.0
turns = [
_turn(1, "counselor", base, {"rapport_signal": 0.4}),
_turn(2, "client", base + 10.0),
]
self.assertEqual(
session_read_model.counselor_baseline_points(
turns, first_turn_ts=base, duration_seconds=60
),
[],
)
class ClientValenceTest(unittest.TestCase):
def test_prefers_payload_turn_valence(self) -> None:
base = 1_000.0
turns = [
_turn(1, "counselor", base),
_turn(2, "client", base + 30.0),
_turn(3, "counselor", base + 60.0),
_turn(4, "client", base + 90.0),
]
payload = {
"turn_valence": [
{"seq": 2, "v": 0.4},
{"seq": 4, "v": -0.6},
]
}
points = session_read_model.client_valence_points(
turns, payload, first_turn_ts=base, duration_seconds=120
)
self.assertEqual(len(points), 2)
self.assertAlmostEqual(points[0].t, 0.25)
self.assertAlmostEqual(points[0].v, 0.4)
self.assertAlmostEqual(points[1].t, 0.75)
self.assertAlmostEqual(points[1].v, -0.6)
def test_fallback_derives_from_turn_evaluations(self) -> None:
base = 1_000.0
negative_eval = {
"client_state_read": [{"code": "defensive", "label_ko": "방어"}],
"appropriateness": "warn",
}
positive_eval = {
"client_state_read": [{"code": "defense_loosening", "label_ko": "방어 완화"}],
"appropriateness": "pos",
}
turns = [
_turn(1, "counselor", base, negative_eval),
_turn(2, "client", base + 30.0),
_turn(3, "counselor", base + 60.0, positive_eval),
_turn(4, "client", base + 90.0),
]
points = session_read_model.client_valence_points(
turns, {}, first_turn_ts=base, duration_seconds=120
)
self.assertEqual(len(points), 2)
self.assertLess(points[0].v, 0.0)
self.assertGreater(points[1].v, 0.0)
for point in points:
self.assertGreaterEqual(point.v, -1.0)
self.assertLessEqual(point.v, 1.0)
class BuildSessionReviewValenceTest(unittest.TestCase):
def test_review_fills_valence_arrays_and_axis(self) -> None:
base = 1_000.0
sess = _session(created_at=base)
sess.ended = True
sess.ended_at = base + 120.0
for i in range(4):
counselor_eval = {
"rapport_signal": 0.2 + i * 0.1,
"client_state_read": [{"code": "affect_contact", "label_ko": "정서 접촉/표현"}],
"appropriateness": "pos",
}
sess.turns.append(
_turn(i * 2 + 1, "counselor", base + i * 30.0, counselor_eval)
)
sess.turns.append(_turn(i * 2 + 2, "client", base + i * 30.0 + 15.0))
evaluation_record = {
"status": "ready",
"payload": {
"strengths": ["감정 반영이 좋았습니다."],
"improvements": [],
"turn_valence": [
{"seq": 2, "v": -0.3},
{"seq": 4, "v": -0.1},
{"seq": 6, "v": 0.2},
{"seq": 8, "v": 0.5},
],
},
}
review = session_read_model.build_session_review(
session_read_model.SessionReviewReadInput(
session=sess,
evaluation_record=evaluation_record,
evaluation_durable=True,
now_ts=base + 200.0,
)
)
self.assertEqual(len(review.clientValence), 4)
self.assertEqual(len(review.counselorBaseline), 4)
# payload turn_valence 를 그대로 쓴다.
self.assertAlmostEqual(review.clientValence[0].v, -0.3)
self.assertAlmostEqual(review.clientValence[-1].v, 0.5)
# x축 라벨은 3~5개.
self.assertGreaterEqual(len(review.valenceAxis), 3)
self.assertLessEqual(len(review.valenceAxis), 5)
self.assertEqual(review.valenceAxis[0], "0:00")
def test_review_without_signals_keeps_empty_arrays(self) -> None:
base = 1_000.0
sess = _session(created_at=base)
sess.ended = True
sess.ended_at = base + 60.0
sess.turns.append(_turn(1, "counselor", base))
sess.turns.append(_turn(2, "client", base + 10.0))
review = session_read_model.build_session_review(
session_read_model.SessionReviewReadInput(
session=sess,
evaluation_record={"status": "ready", "payload": {}},
evaluation_durable=True,
now_ts=base + 100.0,
)
)
self.assertEqual(review.clientValence, [])
self.assertEqual(review.counselorBaseline, [])
# 포인트가 없으면 기존 2원소 axis 를 유지한다.
self.assertEqual(len(review.valenceAxis), 2)
class _DeepValenceEngine:
"""deep-loop 구조화 응답에 turn_valence 를 포함하는 가짜 엔진."""
def __init__(self) -> None:
self.requests: list[Any] = []
async def generate(self, req: Any) -> GenerateResponse:
self.requests.append(req)
return GenerateResponse(
text="",
model="fake-deep",
provider="fake-provider",
structured={
"strengths": ["공감 표현"],
"improvements": [],
"turn_valence": [
{"seq": 2, "v": -0.4},
{"seq": 4, "v": 1.7}, # 범위 초과 → 클램프
{"seq": 0, "v": 0.2}, # 잘못된 seq → 버림
],
},
)
class DeepSchemaTurnValenceTest(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self) -> None:
evaluator.clear_evaluator_semantic_cache()
async def asyncTearDown(self) -> None:
evaluator.clear_evaluator_semantic_cache()
def test_deep_schema_declares_turn_valence(self) -> None:
schema = evaluator._deep_schema()
turn_valence = schema["properties"].get("turn_valence")
self.assertIsNotNone(turn_valence)
self.assertEqual(turn_valence["type"], "array")
item_props = turn_valence["items"]["properties"]
self.assertIn("seq", item_props)
self.assertIn("v", item_props)
# 선택 필드라 required 에는 없어야 한다.
self.assertNotIn("turn_valence", schema["required"])
async def test_evaluate_session_parses_turn_valence(self) -> None:
engine = _DeepValenceEngine()
result = await evaluator.evaluate_session(
session_id="review-valence-session",
stage="라포",
masked_turns=[
{"seq": 1, "speaker": "counselor", "text": "요즘 어떠세요?"},
{"seq": 2, "speaker": "client", "text": "잘 모르겠어요."},
],
engine=engine, # type: ignore[arg-type]
)
self.assertIsNone(result.error)
self.assertEqual(len(result.turn_valence), 2)
self.assertEqual(result.turn_valence[0].seq, 2)
self.assertAlmostEqual(result.turn_valence[0].v, -0.4)
self.assertAlmostEqual(result.turn_valence[1].v, 1.0) # 클램프
# 영속 payload 경로(to_dict)에 turn_valence 가 실린다.
payload = result.to_dict()
self.assertIn("turn_valence", payload)
self.assertEqual(payload["turn_valence"][0]["seq"], 2)
if __name__ == "__main__":
unittest.main()

View file

@ -0,0 +1,68 @@
"""온보딩/프로필 연락처(phone) 형식 검증 회귀 테스트.
2026-07-27 전수 순회에서 발견: 온보딩 연락처에 "전화번호아님텍스트" 같은
임의 문자열을 넣어도 클라이언트·서버 모두 통과되어 DB에 저장됐다.
OnboardingRequest / UserProfilePatch 숫자 9~15자리(하이픈·공백·괄호··
선행 + 허용) 받아야 한다.
"""
from __future__ import annotations
import unittest
from pydantic import ValidationError
from .routes.users import OnboardingRequest, UserProfilePatch
def _onboarding_kwargs(**overrides: object) -> dict:
base: dict = {
"legal_name": "검증 사용자",
"affiliation": "한신대학교",
"department": "상담심리학과",
"grade_level": "3학년",
"phone": "010-1234-5678",
"contact_address": "경기도 오산시 한신대길 137",
"nickname": "검증",
"self_introduction": "연락처 검증 테스트 사용자입니다.",
"avatar_url": "",
"terms_accepted": True,
"privacy_accepted": True,
}
base.update(overrides)
return base
class OnboardingPhoneValidationTest(unittest.TestCase):
def test_onboarding_rejects_non_numeric_phone_text(self) -> None:
with self.assertRaises(ValidationError):
OnboardingRequest(**_onboarding_kwargs(phone="전화번호아님텍스트"))
def test_onboarding_rejects_phone_with_too_few_digits(self) -> None:
with self.assertRaises(ValidationError):
OnboardingRequest(**_onboarding_kwargs(phone="010-12"))
def test_onboarding_accepts_common_korean_phone_formats(self) -> None:
for phone in (
"010-1234-5678",
"01012345678",
"010 1234 5678",
"02-123-4567",
"031-8000-9000",
"+82-10-1234-5678",
):
with self.subTest(phone=phone):
req = OnboardingRequest(**_onboarding_kwargs(phone=phone))
self.assertEqual(req.phone, phone.strip())
def test_profile_patch_rejects_non_numeric_phone_text(self) -> None:
with self.assertRaises(ValidationError):
UserProfilePatch(phone="전화번호아님텍스트")
def test_profile_patch_allows_omitted_phone(self) -> None:
patch = UserProfilePatch()
self.assertIsNone(patch.phone)
if __name__ == "__main__":
unittest.main()