"""협상 불변식 시나리오 하네스 — 실서비스 스택(ChatService + 실 DB 카드)으로 13개 협상을 완주시키고, IMK 가 잡은 두 부류의 사고(같은 카드 반복 · 이상한 금액)가 어떤 흐름에서도 안 나는지 검사한다. 시나리오별 기대 이벤트(카드가 나간 턴의 step·카드·금액)를 정확히 못박고, 공통 불변식을 전 턴에 건다: · 카드 중복 없음 — 한 협상에서 같은 card_id 2회 발동 금지 · 카드 자리 규칙 — 종결 전용(WC-03·05)은 가격협상_카운터에서만, 비종결 와일드는 wild_card_dynamic 에서만 · 타결가 ≤ 목표가 — 어떤 성공 경로도 목표가 초과로 안 끝남 · 카운터 멘트의 금액 = 수락 시 타결가 (표시가=타결가) """ import uuid as _uuid from dataclasses import dataclass, field from typing import Optional import pytest 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 from tests.test_card_tactics import _TENANTS_DIR, _cleanup, _seed_quote_session # 종결 전용 와일드카드(DB tactic 시드와 동일) — 자리 규칙 검사용. _CLOSING_WILDS = {"WC-03", "WC-05"} _NONCLOSING_WILDS = {"WC-01", "WC-02", "WC-04"} # 카드가 나갈 수 있는 스텝(이벤트로 수집). _CARD_STEPS = {"가격협상", "wild_card_dynamic", "wild_card_1pct", "가격협상_카운터"} _BOILERPLATE = [None, "확인", "예", "확인"] @dataclass class Scenario: name: str inputs: list # 서두(안내~기존가격제시) 이후의 협력사 입력 시퀀스 # 기대 이벤트: (step, card, offer_substring). card="NGC-*" 는 임의 협상카드(중복만 검사). events: list settled: Optional[int] # 기대 타결가(원). None=결렬 nego: list = field(default_factory=lambda: ["NGC-001"]) wild: list = field(default_factory=list) target: int = 10_000 anchor: int = 9_900 # 밴드(기본 target 10000·anchor 9900): 1% 존 ≤ 10,098 · 진입 존 ≤ 10,395. SCENARIOS = [ # S01 BB9A 재현 — 중반 비종결 WC-02, 종결 WC-05 실절충가. 같은 카드 2회 없음. Scenario("S01_bb9a_mid_wc02_close_wc05", ["10300", "예", "다른 가격 제시", "10010", "예", "10005", "예", "수락", "확인"], [("wild_card_dynamic", "WC-02", "9950"), ("가격협상", "NGC-001", None), ("가격협상_카운터", "WC-05", "9980")], settled=9980, wild=["WC-02", "WC-05"]), # S02 8AB0 재현 — 절충가(9,205)가 목표가(9,000) 초과 → WC-05 미발동, 목표가 최후통첩(카드 없음). Scenario("S02_8ab0_middle_over_target_skips", ["9500", "예", "9500", "예", "다른 가격 제시", "9500", "예", "확인"], [("가격협상", "NGC-001", None), ("가격협상_카운터", None, "9000")], settled=None, wild=["WC-05"], target=9_000, anchor=8_910), # S03 와일드 5장 전부 + 협상카드 2장 — 중반 1장(WC-01)·종결 1장(WC-03)만, 협상카드는 서로 다른 2장. Scenario("S03_five_wilds_full_run", ["10300", "예", "다른 가격 제시", "10200", "예", "10150", "예", "10100", "예", "수락", "확인"], [("wild_card_dynamic", "WC-01", "10000"), ("가격협상", "NGC-*", None), ("가격협상", "NGC-*", None), ("가격협상_카운터", "WC-03", "10000")], settled=10_000, nego=["NGC-001", "NGC-003"], wild=["WC-01", "WC-02", "WC-03", "WC-04", "WC-05"]), # S04 종결 전용 와일드만 담김 + 제시가가 진입 존에 머무름 — 소진 판정이 막히지 않고 # 종결로 넘어간다(프로브 픽스 회귀: 픽스 전엔 빈 덱에서 쓴 카드를 또 꺼내는 무한 협상). Scenario("S04_closing_only_wild_no_deadlock", ["10300", "예", "10250", "예", "수락", "확인"], [("가격협상", "NGC-001", None), ("가격협상_카운터", None, "10000")], # WC-05 절충 10,075>목표가 → 미발동 → 최후통첩 settled=10_000, wild=["WC-05"]), # S05 1% 존 — 시스템 1% 카드, 수락 시 표시 금액 그대로 타결. Scenario("S05_one_pct_zone_accept", ["10050", "예", "예", "확인"], [("wild_card_1pct", None, "9950")], settled=9_950), # S06 앵커 이하 즉시 타결 — 카드 0장. Scenario("S06_priority_match_no_cards", ["9800", "예", "확인"], [], settled=9_800), # S07 목표가 초과 고수 → 설득 1장 → 최후통첩 → 결렬. Scenario("S07_hold_high_fails", ["11000", "예", "11000", "예", "다른 가격 제시", "11000", "예", "확인"], [("가격협상", "NGC-003", None), ("가격협상_카운터", None, "10000")], settled=None, nego=["NGC-003"]), # S08 협상카드 카운터(NGC-010 목표가 제안) 수락 — 협상카드도 카운터 스텝을 쓴다. Scenario("S08_nego_counter_accept", ["11000", "예", "수락", "확인"], [("가격협상_카운터", "NGC-010", "10000")], settled=10_000, nego=["NGC-010"]), # S09 min_round=2 — WC-04 는 1라운드 진입 존에서 안 나가고 2라운드에 나간다. Scenario("S09_min_round_two_defers_wc04", ["10300", "예", "10200", "예", "수락", "확인"], [("가격협상", "NGC-001", None), ("wild_card_dynamic", "WC-04", "10000")], settled=10_000, wild=["WC-04"]), # S10 종결 체인 폴백 — WC-05 무효(절충 10,175>목표) → 다음 종결 WC-03 발동. Scenario("S10_closing_chain_falls_to_wc03", ["10500", "예", "10450", "예", "다른 가격 제시", "10450", "예", "확인"], [("가격협상", "NGC-001", None), ("가격협상_카운터", "WC-03", "10000")], settled=None, wild=["WC-05", "WC-03"]), # S11 중반+종결 콤보 — WC-02 중반, 종결은 WC-05 무효 건너뛰고 WC-03. 전 카드 1회씩. Scenario("S11_mid_and_closing_combo", ["10300", "예", "다른 가격 제시", "10400", "예", "10350", "예", "수락", "확인"], [("wild_card_dynamic", "WC-02", "9950"), ("가격협상", "NGC-001", None), ("가격협상_카운터", "WC-03", "10000")], settled=10_000, wild=["WC-02", "WC-05", "WC-03"]), # S12 라운드 상한 — 협상카드 3장 각 1회(중복 없음) 후 상한 도달 → 최후통첩 → 결렬. Scenario("S12_round_cap_distinct_nego_cards", ["11000", "예", "11000", "예", "11000", "예", "11000", "예", "다른 가격 제시", "11000", "예", "확인"], [("가격협상", "NGC-*", None), ("가격협상", "NGC-*", None), ("가격협상", "NGC-*", None), ("가격협상_카운터", None, "10000")], settled=None, nego=["NGC-001", "NGC-002", "NGC-003", "NGC-004", "NGC-005"]), # S13 재생성 아님·재료 극단 — 앵커 미박제 세션(anchor=target 폴백)에서도 초과 제시·중복 없음. Scenario("S13_anchor_equals_target_fallback", ["10300", "예", "10200", "예", "수락", "확인"], [("가격협상", "NGC-001", None), ("가격협상_카운터", "WC-03", "10000")], # WC-05 절충 (10000+10200)/2=10100>목표 → 스킵 settled=10_000, wild=["WC-05", "WC-03"], anchor=10_000), # S14 역행 금지(IMK 논의 재현) — 절충 카드(9,950) 뒤에 예산 상한 카드(NGC-007, 앵커 9,900)가 # 선택돼 있어도 발동하지 않는다(설득 폴백으로도 안 나감). 낼 카드가 없어져 종결(목표가 최후통첩)로. Scenario("S14_no_offer_regression", ["10300", "예", "다른 가격 제시", "10200", "예", "수락", "확인"], [("wild_card_dynamic", "WC-02", "9950"), ("가격협상_카운터", None, "10000")], # NGC-007 이벤트가 없어야 함(역행 차단) settled=10_000, nego=["NGC-007"], wild=["WC-02"]), ] @pytest.mark.asyncio @pytest.mark.parametrize("sc", SCENARIOS, ids=[s.name for s in SCENARIOS]) async def test_negotiation_invariants(db_engine, sc: Scenario): reset_sessions() sid = _uuid.uuid4() qid, ver_id = await _seed_quote_session(sid, sc.nego, wild_numbers=sc.wild, target=sc.target, anchor=sc.anchor) try: reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0)) eng = await reg.get_engine(str(_uuid.uuid4())) svc = ChatService() trace, settled, outcome = [], None, None for ui in [*_BOILERPLATE, *sc.inputs]: r = await svc.chat(eng, Req_Chat(session_id=str(sid), user_input=ui)) assert r.result.success is True, f"{sc.name}: 턴 실패 input={ui} msg={r.msg}" trace.append(r) if r.settled_price is not None: settled = r.settled_price if r.chat_end: outcome = r.outcome # ── 기대 이벤트(카드/카운터 턴) 정확 일치 ── events = [r for r in trace if r.step in _CARD_STEPS] got = [(r.step, r.card_id) for r in events] assert len(events) == len(sc.events), f"{sc.name}: 이벤트 수 {got} ≠ 기대 {sc.events}" for r, (step, card, offer) in zip(events, sc.events): assert r.step == step, f"{sc.name}: step {r.step} ≠ {step} (전체 {got})" if card == "NGC-*": assert r.card_id and r.card_id.startswith("NGC-"), f"{sc.name}: 협상카드 기대, 실제 {r.card_id}" else: assert r.card_id == card, f"{sc.name}: card {r.card_id} ≠ {card} (전체 {got})" if offer is not None: assert offer in (r.script or ""), f"{sc.name}: 멘트에 금액 {offer} 없음 — {r.script[:80]}" # ── 공통 불변식 ── played = [r.card_id for r in events if r.card_id] assert len(played) == len(set(played)), f"{sc.name}: 카드 중복 발동 {played}" for r in events: if r.card_id in _CLOSING_WILDS: assert r.step == "가격협상_카운터", f"{sc.name}: 종결 카드 {r.card_id}가 중반({r.step})에 발동" if r.card_id in _NONCLOSING_WILDS: assert r.step == "wild_card_dynamic", f"{sc.name}: 비종결 와일드 {r.card_id}가 {r.step}에서 발동" # ── 결말 ── if sc.settled is None: assert outcome == "failure" and settled is None, f"{sc.name}: 결렬 기대, settled={settled} outcome={outcome}" else: assert outcome == "success", f"{sc.name}: 타결 기대, outcome={outcome}" assert settled == sc.settled, f"{sc.name}: 타결가 {settled} ≠ 기대 {sc.settled}" assert settled <= sc.target, f"{sc.name}: 목표가 초과 타결 {settled} > {sc.target}" finally: await _cleanup(sid, qid, ver_id)