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

"""P7 슬라이스 검증 — 대화형 /chat (스크립트 구동 + 카드선택 학습 + 와일드카드).

1. 전체 대화: 서비스안내→담당자확인→협상품목안내→가격협상→와일드카드→협상완료→협상종료(chat_end).
2. 가격협상 턴에서 카드 선택 + 학습(card_id, updated_q).
3. 와일드카드 발동(wild_card_1pct) + 종료 보상(success).
4. 브랜드 치환 테넌트별(데모상사 A/B), 클린룸(스크립트에 KT 흔적 없음).
5. 경험로그 적재 + 종료 후 진행 시 에러.
6. (HTTP) 헤더로 새 세션 시작 + 한 턴 진행.
"""

import os
import re

import pytest

from router.v1.chat.protocol import Req_Chat
from services.chat_service import ChatService, reset_sessions
from negotiation.chat.service.chat_engine import ChatEngine, ChatSession
from negotiation.chat.service.script_repository import ScriptRepository
from negotiation.qtable.infra.repository.learning_repository import LearningRepository
from tenancy.config_loader import TenantConfigLoader
from tenancy.registry import TenantEngineRegistry

_TENANTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tenants")
_KT = re.compile(r"kt\s*commerce|nego-?wiz", re.IGNORECASE)


def _reg():
    return TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0))


async def _run(svc, eng, turns):
    """turns: user_input 리스트(첫 None=시작). 반환: 응답 리스트."""
    sid, out = None, []
    for ui in turns:
        r = await svc.chat(eng, Req_Chat(session_id=sid, user_input=ui))
        sid = r.session_id
        out.append(r)
        if r.chat_end:
            break
    return out


@pytest.mark.asyncio
async def test_full_conversation_reaches_completion(db_engine):
    reset_sessions()
    eng = await _reg().get_engine("ktcommerce")
    svc = ChatService()
    # anchor=9900, target=10000(기본). 11000(>anchor*1.02)→가격협상(카드),
    #   10000(anchor<p≤anchor*1.02)→1% 와일드카드→수락하면 협상완료.
    turns = [None, "확인", "예", "확인", "11000", "예", "10000", "예", "예",
             "협상 내용을 확인했으며, 이의가 없음에 동의합니다."]
    out = await _run(svc, eng, turns)
    steps = [r.step for r in out]

    assert steps[0] == "서비스안내"
    assert "협상완료" in steps
    assert out[-1].step == "협상종료" and out[-1].chat_end is True

    # 가격협상 카드선택 + 학습
    nego = [r for r in out if r.step == "가격협상"]
    assert nego and nego[0].card_id and nego[0].card_id.startswith("NGC-A")
    assert nego[0].updated_q is not None

    # 와일드카드 발동 (1% 인하)
    assert any(r.step == "wild_card_1pct" for r in out)

    # 종료 보상(success)
    done = [r for r in out if r.outcome == "success"]
    assert done and done[0].reward_total is not None

    # 브랜드 치환 + 클린룸
    assert "데모상사 A" in out[0].script
    assert not _KT.search(out[0].script)

    # 경험로그 적재
    repo = LearningRepository("ktcommerce")
    err, cnt = await repo.read(lambda s: repo.count_experience(s))
    assert cnt >= 1


@pytest.mark.asyncio
async def test_wildcard_1pct_accept_settles_at_offer_price(db_engine):
    """와일드카드 1% 인하 제안을 '예'로 수락하면 합의가는 인하 제안가(offer_1pct)여야 한다.

    회귀: 수락해도 input_price 가 직전 제시가로 남아 요약/입찰가가 원래 가격(예: 20000)으로
    잡히던 버그 — settled_price 가 인하가(예: 19800)로 내려와야 한다.
    """
    reset_sessions()
    eng = await _reg().get_engine("ktcommerce")
    svc = ChatService()
    # anchor=9900(기본). 10000 은 anchor*1.02(10098) 이내 → wild_card_1pct 발동, offer_1pct=9900.
    out = await _run(svc, eng, [None, "확인", "예", "확인", "10000", "예", "예",
                                "협상 내용을 확인했으며, 이의가 없음에 동의합니다."])
    steps = [r.step for r in out]
    assert "wild_card_1pct" in steps

    # 수락 후 협상완료(요약 턴)·협상종료(종료 턴) 모두 합의가 = 인하 제안가 9900
    done = [r for r in out if r.step in ("협상완료", "협상종료")]
    assert done and all(r.settled_price == 9900 for r in done)


def test_partner_type_enum_mapping():
    """PartnerType 계약값(0=NONE/1=SINGLE/2=MULTIPLE)과 수→유형 매핑."""
    from negotiation.qtable.domain.model.snapshot import PartnerType

    assert PartnerType.NONE == 0 and PartnerType.SINGLE == 1 and PartnerType.MULTIPLE == 2
    assert PartnerType.from_count(0) is PartnerType.NONE
    assert PartnerType.from_count(1) is PartnerType.SINGLE
    assert PartnerType.from_count(2) is PartnerType.MULTIPLE
    assert PartnerType.from_count(7) is PartnerType.MULTIPLE


def test_acceptance_ratio_dynamic_calc():
    """가격 수용률 동적 계산 — 기존 공급가(item_price) 기준 양보율. 첫 라운드부터 실값."""
    calc = ChatService._acceptance_ratio
    assert calc({}) == 0.0                                                   # 가격 입력 전
    # 기존 공급가 12000 기준: 첫 제시 11000 도 첫 라운드부터 양보율 실값 (스크립트 discount_rate 와 동일 기준)
    assert calc({"item_price": 12000, "input_price": 11000}) == pytest.approx(1000 / 12000)
    assert calc({"item_price": 12000, "first_offer_price": 11000, "input_price": 10200}) == pytest.approx(1800 / 12000)
    assert calc({"item_price": 10000, "input_price": 12000}) == 0.0          # 기존가보다 인상(비정상) → 0
    # 기존 공급가 없음(신규 협상) → 첫 제시가 기준 폴백: 첫 제시 0, 이후 양보율
    assert calc({"item_price": 0, "first_offer_price": 11000, "input_price": 11000}) == 0.0
    assert calc({"item_price": 0, "first_offer_price": 11000, "input_price": 10200}) == pytest.approx(800 / 11000)
    assert calc({"input_price": 9800}) == 0.0                                # 기준가 전무(과거 세션 호환)


def test_card_id_fixed_mapping_and_selection_mask():
    """카드 정리 후: action_id↔카드는 테넌트 매핑으로 고정, 견적 선택은 available_mask 로 걸러진다.
    (구 인덱스 방식 폐기 — selected[action_id] 인덱싱은 견적마다 action_id 의미가 달라져 Q-table 오염.)"""
    class _Mapper:
        _m = {i: f"NGC-A{i + 1:03d}" for i in range(11)}
        def get_card_id(self, a):
            return self._m.get(a)
        def get_action_id(self, num):
            return next((a for a, c in self._m.items() if c == num), None)
    class _Engine:
        mapper = _Mapper()
        action_space_size = 11
    eng = _Engine()
    session = ChatSession(
        session_id="00000000-0000-0000-0000-000000000001", tenant_id="ktcommerce", company_id="ktcommerce",
        context={"selected_nego_card_numbers": ["NGC-A003", "NGC-A008"]}, action_space_size=11,
    )

    # ① card_id 는 고정 매핑 (선택 리스트 인덱싱 아님)
    assert ChatService._card_id_for_action(eng, session, 0) == "NGC-A001"
    assert ChatService._card_id_for_action(eng, session, 2) == "NGC-A003"

    # ② 선택은 mask 로 — NGC-A003(action 2), NGC-A008(action 7) 만 pickable
    mask = ChatService._selection_mask(eng, session)
    assert mask is not None and mask[2] and mask[7]
    assert not mask[0] and not mask[5] and mask.sum() == 2

    # ③ 사용한 action 은 mask 에서 제외
    session.used_action_ids = {2}
    mask2 = ChatService._selection_mask(eng, session)
    assert not mask2[2] and mask2[7] and mask2.sum() == 1

    # ④ 선택 없으면 None → 전체 허용(폴백)
    session.context["selected_nego_card_numbers"] = []
    assert ChatService._selection_mask(eng, session) is None


def test_default_1pct_wildcard_still_runs_without_selected_wildcard():
    """1% 인하는 기본 제공 카드라 DB 견적에서 와일드카드를 선택하지 않아도 발동한다."""
    cfg = TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0).load("ktcommerce")
    repo = ScriptRepository(cfg, _TENANTS_DIR)
    engine = ChatEngine(repo, rq_type="재협상")
    session = ChatSession(
        session_id="00000000-0000-0000-0000-000000000002",
        tenant_id="ktcommerce",
        company_id="ktcommerce",
        step="가격협상_확인",
        action_space_size=0,
        context={
            "input_price": 10000,
            "anchor_price": 9900,
            "target_price": 10000,
            "round": 1,
            "allow_selected_wildcards": False,
        },
    )

    view = engine.advance(session, "예")
    assert view.step == "wild_card_1pct"


def test_budget_wildcard_requires_selected_wildcard_for_db_context():
    """재원부족 구간은 DB 견적에서 와일드카드를 선택했을 때만 발동한다."""
    cfg = TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0).load("ktcommerce")
    repo = ScriptRepository(cfg, _TENANTS_DIR)
    engine = ChatEngine(repo, rq_type="재협상")
    session = ChatSession(
        session_id="00000000-0000-0000-0000-000000000003",
        tenant_id="ktcommerce",
        company_id="ktcommerce",
        step="가격협상_확인",
        action_space_size=0,
        context={
            "input_price": 10200,
            "anchor_price": 9900,
            "target_price": 10000,
            "round": 1,
            "allow_selected_wildcards": False,
        },
    )

    view = engine.advance(session, "예")
    assert view.step == "가격협상"
    assert not view.step.startswith("wild_card_")


@pytest.mark.asyncio
async def test_priority_completes_without_wildcard(db_engine):
    reset_sessions()
    eng = await _reg().get_engine("ktcommerce")
    svc = ChatService()
    # 첫 제시가가 앵커가(9900) 이하 → 우선협상 → 바로 협상완료(카드/와일드카드 없이)
    out = await _run(svc, eng, [None, "확인", "예", "확인", "9800", "예",
                                "협상 내용을 확인했으며, 이의가 없음에 동의합니다."])
    steps = [r.step for r in out]
    assert "협상완료" in steps
    assert "wild_card_budget" not in steps
    assert "가격협상" not in steps   # 우선협상이라 카드 협상 없이 타결


@pytest.mark.asyncio
async def test_tenant_brand_isolation(db_engine):
    reset_sessions()
    svc = ChatService()
    reg = _reg()
    ik = await svc.chat(await reg.get_engine("imarketkorea"), Req_Chat(session_id=None))
    assert "데모상사 B" in ik.script


@pytest.mark.asyncio
async def test_advance_after_end_errors(db_engine):
    reset_sessions()
    eng = await _reg().get_engine("ktcommerce")
    svc = ChatService()
    out = await _run(svc, eng, [None, "확인", "예", "확인", "1000", "예",
                                "협상 내용을 확인했으며, 이의가 없음에 동의합니다."])
    sid = out[-1].session_id
    r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="확인"))
    assert r.result.desc == "NEGO_INVALID_STEP"


@pytest.mark.asyncio
async def test_http_chat_start(client):
    r = await client.post("/v1/chat", headers={"X-Tenant-ID": "ktcommerce"}, json={"rq_type": "재협상"})
    assert r.status_code == 200
    d = r.json()
    assert d["step"] == "서비스안내"
    assert d["input_options"] == ["확인"]
    assert "데모상사 A" in d["script"]
    # 헤더 없으면 400
    r2 = await client.post("/v1/chat", json={"rq_type": "재협상"})
    assert r2.status_code == 400