o2o-negosium-original/agent/tests/test_input_interpreter.py

"""Phase 3 이해층 검증 — InputInterpreter (자유 발화 NLU → 기대 입력 구조화).

원칙 검증: LLM 은 의도 분류·가격 표현 위치만 찾고, 숫자 계산은 결정론 파서가 한다.
① 한국어 가격 파서 결정론 ② choice 는 선택지 목록 검증 ③ price_text 는 원문 부분문자열 검증
④ 실패/타임아웃 → None(원문 폴백) ⑤ 챗 플로우: 자유 발화로 분기·가격 입력이 진행된다
⑥ LLM 미설정 시 기존(정형 입력) 동작 그대로 — 회귀 없음.
"""

import os

import pytest

from negotiation.chat.service.input_interpreter import (
    InputInterpreter, InterpretedInput, parse_korean_price,
)

_TENANTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tenants")


# ---- 결정론 한국어 가격 파서 -------------------------------------------------
@pytest.mark.parametrize("text,expected", [
    ("10,500원", 10500),
    ("10500", 10500),
    ("1만 500원", 10500),
    ("1만500원", 10500),
    ("만원", 10000),
    ("1.5만", 15000),
    ("3만2천원", 32000),
    ("2억", 200_000_000),
    ("0", None),                 # 0 이하 무효
    ("그건 어렵습니다", None),   # 가격 아님
    ("만원에 3개", None),        # 잡문자 혼입 → 해석 불가(안전 폴백)
])
def test_parse_korean_price(text, expected):
    got = parse_korean_price(text)
    assert (got == expected) if expected is not None else (got is None)


# ---- LLM 출력 검증 (환각 차단) ----------------------------------------------
def _fake(reply: dict):
    def call(messages):
        return reply
    return call


@pytest.mark.asyncio
async def test_choice_mapped_to_option():
    nat = InputInterpreter(llm_call=_fake({"intent": "choice", "choice": "예", "price_text": None}))
    out = await nat.interpret("네 접니다, 말씀하세요", input_mode="yes_no", input_options=["예", "아니오"])
    assert out == InterpretedInput(kind="choice", value="예", source="예")


@pytest.mark.asyncio
async def test_choice_outside_options_rejected():
    nat = InputInterpreter(llm_call=_fake({"intent": "choice", "choice": "글쎄요", "price_text": None}))
    assert await nat.interpret("음...", input_mode="yes_no", input_options=["예", "아니오"]) is None


@pytest.mark.asyncio
async def test_price_span_verified_and_parsed_deterministically():
    nat = InputInterpreter(llm_call=_fake({"intent": "price", "choice": None, "price_text": "1만 500원"}))
    out = await nat.interpret("저희 마진상 1만 500원까지는 맞춰드릴 수 있습니다", input_mode="price")
    assert out is not None and out.kind == "price"
    assert out.value == "10500"          # 숫자는 결정론 파서 산출(LLM 계산 아님)
    assert out.source == "1만 500원"


@pytest.mark.asyncio
async def test_price_span_not_in_text_rejected():
    """LLM 이 원문에 없는 가격 표현을 지어내면 폐기(환각 차단)."""
    nat = InputInterpreter(llm_call=_fake({"intent": "price", "choice": None, "price_text": "9,000원"}))
    assert await nat.interpret("만원이면 가능합니다", input_mode="price") is None


@pytest.mark.asyncio
async def test_unknown_and_failures_fall_back():
    assert await InputInterpreter(llm_call=_fake({"intent": "unknown"})).interpret(
        "글쎄요 검토해 볼게요", input_mode="yes_no", input_options=["예", "아니오"]) is None

    def boom(messages):
        raise RuntimeError("LLM down")
    assert await InputInterpreter(llm_call=boom).interpret("네", input_mode="yes_no", input_options=["예"]) is None

    def slow(messages):
        import time
        time.sleep(0.5)
        return {"intent": "choice", "choice": "예"}
    nat = InputInterpreter(llm_call=slow, timeout_seconds=0.05)
    assert await nat.interpret("네", input_mode="yes_no", input_options=["예"]) is None


# ---- 챗 플로우 E2E (fake LLM) ------------------------------------------------
def _routing_fake(messages):
    """단계별 fake — 기대 입력이 가격이면 price, 아니면 '예' choice 로 응답."""
    user = messages[-1]["content"]
    if "가격(숫자)" in user:
        return {"intent": "price", "choice": None, "price_text": "1만 500원"}
    return {"intent": "choice", "choice": "예", "price_text": None}


@pytest.mark.asyncio
async def test_chat_flow_free_text_negotiation(db_engine):
    """자유 발화만으로 담당자확인 분기 + 가격 입력이 진행된다 (버튼 없는 '진짜 대화')."""
    from router.v1.chat.protocol import Req_Chat
    from services.chat_service import ChatService, reset_sessions
    from tenancy.config_loader import TenantConfigLoader
    from tenancy.registry import TenantEngineRegistry

    reset_sessions()
    reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0))
    eng = await reg.get_engine("imarketkorea")
    eng.config.llm.enabled = True

    svc = ChatService()
    svc._interpreter = InputInterpreter(llm_call=_routing_fake)
    orig = InputInterpreter.available
    InputInterpreter.available = staticmethod(lambda: True)
    try:
        sid = None
        r = await svc.chat(eng, Req_Chat(session_id=sid))                       # 서비스안내
        sid = r.session_id
        r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="확인"))    # 담당자확인 (fast path)
        assert r.step == "담당자확인" and r.interpreted_input is None

        # 자유 발화 → NLU 가 "예" 로 매핑 → 협상품목안내
        r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="네 접니다, 말씀하세요"))
        assert r.step == "협상품목안내"
        assert r.interpreted_input == "예"

        r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="확인"))    # 기존가격제시(price)
        # 자유 발화 가격 → span "1만 500원" → 결정론 파서 10500 → 가격 저장 후 확인 단계
        r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="저희 마진상 1만 500원까지는 맞춰드릴 수 있습니다"))
        assert r.step == "가격협상_확인"
        assert r.interpreted_input == "10500"
        assert "10500" in r.script          # 멘트 치환도 해석된 가격으로
    finally:
        InputInterpreter.available = orig
        eng.config.llm.enabled = False


@pytest.mark.asyncio
async def test_chat_flow_without_llm_keeps_legacy_behavior(db_engine):
    """LLM 미설정(available=False)이면 자유 발화는 원문 그대로 엔진에 전달 — 기존 동작 회귀 없음."""
    from router.v1.chat.protocol import Req_Chat
    from services.chat_service import ChatService, reset_sessions
    from tenancy.config_loader import TenantConfigLoader
    from tenancy.registry import TenantEngineRegistry

    reset_sessions()
    reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0))
    eng = await reg.get_engine("imarketkorea")

    svc = ChatService()
    sid = None
    r = await svc.chat(eng, Req_Chat(session_id=sid))
    sid = r.session_id
    r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="확인"))
    assert r.step == "담당자확인"
    # conftest 가드로 available=False → NLU 미동작, interpreted_input 없음
    r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="네 접니다"))
    assert r.interpreted_input is None