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

"""Phase 1 결정 스택 잔여분 검증 — 규칙 데이터화 + 선택카드 우선순위 prior.

1. 와일드카드 진입 임계(wildcard_1pct_ratio/entry_ratio)·카운터 라운드 상한(max_counter_rounds)이
   하드코딩이 아니라 테넌트 config(negotiation.*)로 주입된다.
2. 의도층 prior: 갑이 견적에서 고른 카드 순서가 콜드 스타트 선택을 결정하고,
   학습(Q·방문수)이 쌓이면 영향이 소멸한다 — Q-table 오염 없음.
"""

import os

import numpy as np
import pytest

from negotiation.chat.service.chat_engine import ChatEngine, ChatSession
from negotiation.chat.service.script_repository import ScriptRepository
from negotiation.policies.base import EpisodeState, PolicyContext
from negotiation.policies.qtable_policy import UCBQTablePolicy
from negotiation.qtable.domain.model.q_table import QTable
from negotiation.qtable.domain.model.snapshot import NegotiationSnapshot
from tenancy.config_loader import TenantConfigLoader

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


def _engine(**rule_overrides) -> ChatEngine:
    cfg = TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0).load("imarketkorea")
    for k, v in rule_overrides.items():
        setattr(cfg.negotiation, k, v)
    return ChatEngine(ScriptRepository(cfg, _TENANTS_DIR), rq_type="재협상")


def _session(price, anchor=10000, rnd=1, **ctx_over):
    ctx = {"input_price": price, "anchor_price": anchor, "target_price": anchor + 100,
           "round": rnd, "allow_selected_wildcards": False}
    ctx.update(ctx_over)
    return ChatSession(session_id="00000000-0000-0000-0000-00000000d001", tenant_id="imarketkorea",
                       company_id="imarketkorea", step="가격협상_확인", action_space_size=0, context=ctx)


# ---- 규칙 데이터화 -----------------------------------------------------------
def test_default_rules_loaded_from_config():
    eng = _engine()
    assert eng.rules.wildcard_1pct_ratio == 1.02
    assert eng.rules.wildcard_entry_ratio == 1.05
    assert eng.rules.max_counter_rounds == 3


def test_wildcard_threshold_is_config_driven():
    # 기본(1.02): anchor 10000, 제시 10800 → 임계 밖 → 일반 가격협상
    view = _engine().advance(_session(10800), "예")
    assert view.step == "가격협상"
    # 임계를 1.10 으로 완화한 테넌트 → 같은 가격에서 1% 인하 와일드카드 발동
    view = _engine(wildcard_1pct_ratio=1.10).advance(_session(10800), "예")
    assert view.step == "wild_card_1pct"


def test_max_counter_rounds_is_config_driven():
    # round=3(카운터 2회 경과), 제시 11000: 기본 상한 3 → 아직 협상 지속
    view = _engine().advance(_session(11000, rnd=3), "예")
    assert view.step == "가격협상"
    # 상한 1 → 종결 국면 진입: 곧장 실패가 아니라 종결 전술 발동 지점(force_closing)으로
    s = _engine(max_counter_rounds=1), _session(11000, rnd=3)
    view = s[0].advance(s[1], "예")
    assert view.step == "가격협상" and s[1].context.get("force_closing") is True
    # 종결 전술까지 소진(closing_played) 후에도 target(10100) 초과 → 결렬
    view = _engine(max_counter_rounds=1).advance(_session(11000, rnd=3, closing_played=True), "예")
    assert view.step == "협상실패"
    # 종결 후 제시가가 target 이하로 내려오면 결렬이 아니라 타결 (새 규칙 — 구현 전엔 무조건 실패)
    view = _engine(max_counter_rounds=1).advance(
        _session(10050, rnd=3, closing_played=True, wildcard_used=True), "예")
    assert view.step == "협상완료"


# ---- 선택카드 우선순위 prior ---------------------------------------------------
def _snap():
    return NegotiationSnapshot(revenue_amount=1, distribution_code="A", partner_count=1,
                               acceptance_ratio=0.1, input_price=900, anchor_price=800, target_price=1000)


def _ctx(prior=None, mask=None, n=11):
    return PolicyContext(state_index=0, snapshot=_snap(), action_space_size=n,
                         available_mask=mask, prior_bonus=prior, episode=EpisodeState())


def test_prior_decides_cold_start_order():
    """콜드 스타트(Q=0·방문 0)에서는 갑이 먼저 고른 카드(높은 prior)가 먼저 나간다."""
    qt = QTable(2, 11)
    prior = np.zeros(11)
    prior[7], prior[2] = 0.3, 0.15          # 선택 순서: action7 → action2
    mask = np.zeros(11, dtype=bool)
    mask[2] = mask[7] = True
    p = UCBQTablePolicy(qt, mark_visits=False)
    assert p.select(_ctx(prior=prior, mask=mask)).action_id == 7


def test_prior_decays_as_learning_accumulates():
    """학습이 쌓이면(Q·방문수) prior 는 1/(1+visits) 로 감쇠 — Q 가 지배한다."""
    qt = QTable(2, 11)
    qt.q[0, 2] = 1.0                         # action2 가 학습상 우월
    qt.visits[0, 2] = 5
    qt.visits[0, 7] = 5                      # 탐색 보너스 동률
    prior = np.zeros(11)
    prior[7] = 0.3                           # 갑 선호는 action7
    mask = np.zeros(11, dtype=bool)
    mask[2] = mask[7] = True
    p = UCBQTablePolicy(qt, mark_visits=False)
    assert p.select(_ctx(prior=prior, mask=mask)).action_id == 2


def test_no_prior_keeps_existing_behavior():
    """prior 미주입(None) 시 기존 UCB 동작 그대로 — 회귀 없음."""
    qt = QTable(2, 4)
    qt.q[0] = np.array([1.0, 9.0, 2.0, 0.0])
    qt.visits[0] = np.array([5, 5, 5, 5])
    p = UCBQTablePolicy(qt, exploration_constant=0.1, mark_visits=False)
    assert p.select(_ctx(n=4)).action_id == 1


def test_selection_prior_built_from_quotation_order():
    """ChatService._selection_prior — 견적 선택 순서 → prior 배열 (앞선 선택일수록 큼)."""
    from services.chat_service import ChatService

    class _Mapper:
        _m = {i: f"NGC-B{i + 1:03d}" for i in range(11)}
        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

    session = ChatSession(session_id="00000000-0000-0000-0000-00000000d002", tenant_id="t", company_id="t",
                          context={"selected_nego_card_numbers": ["NGC-B008", "NGC-B003"]})
    prior = ChatService._selection_prior(_Engine(), session)
    assert prior is not None
    assert prior[7] > prior[2] > 0           # 먼저 고른 NGC-B008(action7) 이 더 큼
    assert prior[[0, 1, 4, 10]].sum() == 0   # 미선택 카드는 0

    # 선택이 1장이면 순서 정보가 없어 None
    session.context["selected_nego_card_numbers"] = ["NGC-B008"]
    assert ChatService._selection_prior(_Engine(), session) is None