o2o-negosium-original/agent/services/chat_service.py

"""ChatService — 대화형 /chat 오케스트레이션 (P7 슬라이스).

ChatEngine(동기 step 전이) + UCB Q-Table(가격협상 카드선택·학습) + DB(experience_logs) 결합.
세션 상태는 인메모리 스토어(PoC; 단일 워커). 종료 outcome 은 마지막 (state,action)에 종료보상을 역전파.
"""

import uuid
from typing import Optional

from common.enums import DBType, ErrorType
from common.database.db_session_manager import DB_SESSION_MNG
from common.logger import LOG
from config.server_configs import agent_config
from negotiation.chat.service.chat_engine import ChatEngine, ChatSession, StepView
from negotiation.chat.service.indicator import compute_indicator
from negotiation.chat.service.chat_session_repository import ChatSessionRepository
from negotiation.chat.service.negotiation_context_loader import NegotiationContextLoader
from negotiation.chat.service.script_repository import ScriptRepository
from negotiation.policies.base import EpisodeState, PolicyContext, Transition
from negotiation.policy.model_store import QTablePolicyStore
from negotiation.qtable.domain.model.snapshot import NegotiationOutcome, NegotiationSnapshot, PartnerType
from negotiation.qtable.domain.service.reward_calculator import RewardCalculator
from negotiation.qtable.domain.service.state_calculator import state_index
from negotiation.qtable.infra.repository.learning_repository import LearningRepository
from router.v1.chat.protocol import Req_Chat, Res_Chat, Res_ChatSession
from tenancy.registry import TenantEngine

# 직접 호출(데모/테스트) 폴백 기본 컨텍스트 — 운영 경로는 NegotiationContextLoader 가
# DB(negotiation.sessions·partner.items·partner.suppliers·quotation.quotations)에서 조회한다.
_DEFAULT_RQ_TYPE = "재협상"
_DEFAULT_TARGET_PRICE = 10000   # KT 목표 매입가
_DEFAULT_ANCHOR_PRICE = 9900    # 앵커링가(목표가보다 낮음). 제시가 ≤ anchor → 우선협상
_DEFAULT_REVENUE_AMOUNT = 20_000_000  # 매출액(원) — suppliers.total_revenue 미기재 시 폴백
_DEFAULT_DISTRIBUTION_CODE = "A"      # 유통 코드 — quotations.supplier_type 미지정 시 폴백


class ChatService:
    async def chat(self, engine: TenantEngine, req: Req_Chat) -> Res_Chat:
        res = Res_Chat()
        repo = ScriptRepository(engine.config, agent_config.tenants_dir)
        sess_repo = ChatSessionRepository(engine.company_id)

        # 1) 세션 확보 / 시작 (DB 영속 — 재시작/멀티워커 안전, P8-A)
        session = await sess_repo.get(req.session_id) if req.session_id else None
        # 새 세션 컨텍스트: 요청 페이로드 대신 DB(negotiation.sessions 등)에서 1회 조회.
        #   행이 없으면(데모/테스트 직접 호출) 기본값 폴백.
        db_ctx = None if session else await NegotiationContextLoader().load(req.session_id)
        rq_type = session.rq_type if session else (db_ctx.rq_type if db_ctx else _DEFAULT_RQ_TYPE)
        chat_engine = ChatEngine(repo, rq_type=rq_type)

        # ① step desync 감지: backend 가 본 직전 봇 step(client_step)이 agent 세션 step 과 다르면 경고.
        #    agent 가 자기 step 을 정답으로 보고 진행하고(응답의 step/client_step 으로 backend 가 따라옴),
        #    추적/리싱크 트리거를 위해 로깅 + 응답 desynced 플래그로 알린다.
        desynced = False
        if session is not None and req.client_step:
            agent_step = session.step
            agent_client_step = chat_engine.step_map.get(agent_step, agent_step)
            if req.client_step not in (agent_step, agent_client_step):
                desynced = True
                LOG.w(
                    f"[ChatService] step desync: backend client_step={req.client_step!r} != "
                    f"agent step={agent_step!r}(client={agent_client_step!r}) session={session.session_id}"
                )

        if session is None:
            # session_id honoring: backend 가 보낸 session_id(= negotiation.sessions.session_id)를
            # 새 uuid 발급 없이 그대로 세션 키로 쓴다. 없으면(직접 호출/데모) 생성.
            session = ChatSession(
                session_id=req.session_id or str(uuid.uuid4()), tenant_id=engine.tenant_id, company_id=engine.company_id,
                rq_type=rq_type, action_space_size=engine.action_space_size,
                context={
                    # 매출액 = suppliers.total_revenue, 유통코드 = quotations.supplier_type 매핑 (loader).
                    #   미기재/미지정이면 기본값 폴백.
                    "revenue_amount": db_ctx.revenue_amount if db_ctx and db_ctx.revenue_amount > 0 else _DEFAULT_REVENUE_AMOUNT,
                    "distribution_code": db_ctx.distribution_code if db_ctx and db_ctx.distribution_code else _DEFAULT_DISTRIBUTION_CODE,
                    # 파트너 유형(PartnerType 값 0/1/2) — 상품별 협력사 수 DB 조회로 세션 시작 시 1회 확정.
                    #   snapshot.partner_count 로 그대로 사용(값 호환).
                    # 가격 수용률은 컨텍스트에 두지 않는다 — _snapshot 이 라운드별 제시가로 동적 계산.
                    "partner_count": int(db_ctx.partner_type) if db_ctx else int(PartnerType.SINGLE),
                    # 목표가/앵커링가: sessions 행(생성 시 박제된 anchoring_price) → 박제 ‰ → 1% 폴백 (loader).
                    "anchor_price": db_ctx.anchor_price if db_ctx else _DEFAULT_ANCHOR_PRICE,
                    "target_price": db_ctx.target_price if db_ctx else _DEFAULT_TARGET_PRICE,
                    "round": 0,
                    # 기존 공급가(품목 기준가) — 가격협상_확인 인하율 산출용.
                    "item_price": db_ctx.item_price if db_ctx else 0,
                },
            )
            view = chat_engine.start(session)
        else:
            view = chat_engine.advance(session, req.user_input)

        # 2) 응답 기본 채움 (학습 블록이 가격협상 턴에서 script/indicator 를 덮어쓸 수 있어 먼저 채운다)
        res.session_id = session.session_id
        res.step = view.step
        res.client_step = view.client_step
        res.script = view.script
        res.input_mode = view.input_mode
        res.input_options = view.input_options
        res.chat_end = view.chat_end
        res.outcome = view.outcome
        res.desynced = desynced
        # 합의가: 성공 확정 이후 턴(협상완료 요약 → 협상종료)에 내려준다. 와일드카드 수락처럼
        # 유저가 직접 입력하지 않은 가격으로 타결될 수 있어 backend 요약/입찰가는 이 값을 최우선으로 쓴다.
        if session.context.get("final_outcome") == "success" and session.context.get("input_price"):
            res.settled_price = int(session.context["input_price"])

        # 3) 학습 결합 (가격협상 카드선택 → 카드 스크립트·협상지표 / 종료 보상)
        if view.error is None and engine.action_space_size > 0:
            if view.needs_card_selection:
                await self._select_and_learn(engine, chat_engine, repo, session, res)
            elif view.outcome is not None:
                await self._terminal_learn(engine, session, view.outcome, res)

        if view.error:
            res.result.SetResult(ErrorType.NEGO_INVALID_STEP)
            res.msg = view.error

        await sess_repo.save(session)   # 진행 상태 영속화 (재시작/멀티워커 안전)
        return res

    # ---- 세션 상태 조회 (① desync 리싱크용) ----------------------------
    async def get_session_state(self, engine: TenantEngine, session_id: str) -> "Res_ChatSession":
        """backend 가 타임아웃/재진입 시 agent 의 현재 step 을 읽어 정합을 맞춘다."""
        res = Res_ChatSession()
        repo = ScriptRepository(engine.config, agent_config.tenants_dir)
        session = await ChatSessionRepository(engine.company_id).get(session_id)
        if session is None:
            res.result.SetResult(ErrorType.NEGO_SESSION_NOT_FOUND)
            return res
        step_map = repo.client_step_mapping()
        res.session_id = session.session_id
        res.step = session.step
        res.client_step = step_map.get(session.step, session.step)
        res.rq_type = session.rq_type
        res.ended = session.ended
        res.found = True
        return res

    # ---- 학습 ----------------------------------------------------------
    @staticmethod
    def _acceptance_ratio(context: dict) -> float:
        """가격 수용률 동적 계산 — 협력사 첫 제시가 대비 현재 제시가의 양보율 (설계서 공식).

        acceptance = max(0, (첫 제시가 − 현재 제시가) / 첫 제시가).
        첫 제시 라운드(양보 없음)·첫 제시가 미기록(과거 세션 호환)이면 0(low 버킷).
        """
        first = context.get("first_offer_price") or 0
        current = context.get("input_price") or 0
        if first <= 0 or current <= 0:
            return 0.0
        return max(0.0, (first - current) / first)

    def _snapshot(self, session: ChatSession, outcome: NegotiationOutcome) -> NegotiationSnapshot:
        c = session.context
        return NegotiationSnapshot(
            revenue_amount=c["revenue_amount"], distribution_code=c["distribution_code"],
            partner_count=c["partner_count"], acceptance_ratio=self._acceptance_ratio(c),
            input_price=c.get("input_price", c["anchor_price"]), anchor_price=c["anchor_price"],
            target_price=c["target_price"], round_number=c.get("round", 0), outcome=outcome,
        )

    async def _select_and_learn(self, engine: TenantEngine, chat_engine: ChatEngine,
                                scripts: ScriptRepository, session: ChatSession, res: Res_Chat):
        snap = self._snapshot(session, NegotiationOutcome.ONGOING)
        try:
            idx = state_index(snap, engine.config.state)
        except ValueError as ex:
            LOG.e_no_callstack(f"[ChatService] state error: {ex}")
            return
        policy, version_id, repo = await QTablePolicyStore.load(engine)
        ctx = PolicyContext(state_index=idx, snapshot=snap, action_space_size=engine.action_space_size,
                            episode=EpisodeState(used_action_ids=set(session.used_action_ids)))
        decision = policy.select(ctx)
        session.used_action_ids.add(decision.action_id)
        card_id = engine.mapper.get_card_id(decision.action_id)
        reward = RewardCalculator(engine.config.reward, engine.config.state).calculate(snap)
        policy.update(Transition(state_index=idx, action_id=decision.action_id, reward=reward.total, done=False))
        await QTablePolicyStore.persist_cell(repo, version_id, policy, idx, decision.action_id)
        session.context["last_state"] = idx
        session.context["last_action"] = decision.action_id
        await self._log(repo, session, idx, decision.action_id, card_id, snap, reward, decision.propensity, done=False)

        res.card_id = card_id
        res.policy = policy.name
        res.q_value = decision.q_value
        res.updated_q = float(policy.qtable.q[idx, decision.action_id])
        res.visit_count = int(policy.qtable.visits[idx, decision.action_id])
        res.reward_total = reward.total

        # 가격협상 턴 연출(선행 chat_server 의 dynamic step type=indicator 재현):
        #   ① 선택된 카드의 스크립트를 봇 메시지(script)로 출력 ② 협상지표 게이지(indicator_value) 동봉.
        #   backend/front 가 indicator/bot_chat_type 패스스루·게이지 렌더 준비 완료 → 값만 채우면 표시된다.
        card_script = scripts.card_script(decision.action_id, chat_engine.vars_for(session))
        if card_script:
            res.script = card_script
        c = session.context
        ind = compute_indicator(c.get("anchor_price", 0), c.get("input_price", 0), c.get("target_price", 0))
        if ind is not None:
            res.indicator_value = float(ind[0])
            res.indicator_range = ind[1]
            res.bot_chat_type = "indicator"

    async def _terminal_learn(self, engine: TenantEngine, session: ChatSession, outcome: str, res: Res_Chat):
        oc = NegotiationOutcome.SUCCESS if outcome == "success" else NegotiationOutcome.FAILURE
        snap = self._snapshot(session, oc)
        reward = RewardCalculator(engine.config.reward, engine.config.state).calculate(snap)
        res.reward_total = reward.total
        last_state = session.context.get("last_state")
        last_action = session.context.get("last_action")
        if last_state is None or last_action is None:
            return  # 카드선택 없이 종료된 경우(예: 담당자확인 단계 이탈)
        policy, version_id, repo = await QTablePolicyStore.load(engine)
        policy.update(Transition(state_index=last_state, action_id=last_action, reward=reward.total, done=True))
        await QTablePolicyStore.persist_cell(repo, version_id, policy, last_state, last_action)
        await self._log(repo, session, last_state, last_action,
                        engine.mapper.get_card_id(last_action), snap, reward, None, done=True)
        res.updated_q = float(policy.qtable.q[last_state, last_action])

    async def _log(self, repo: LearningRepository, session, state_index, action_id, card_id, snap, reward, propensity, done):
        data = {
            "session_id": session.session_id, "state_index": state_index, "action_id": action_id,
            "card_id": card_id, "snapshot": snap.to_dict(), "propensity": propensity,
            "turn": snap.round_number, "reward": reward.total, "done": done,
            "settled_price": int(snap.input_price) if snap.outcome == NegotiationOutcome.SUCCESS else None,
        }
        try:
            await DB_SESSION_MNG.execute_lambda_run([DBType.MAIN.value], [lambda s: repo.log_transition(s, data)])
        except Exception as ex:
            LOG.e_no_callstack(f"[ChatService] log failed: {ex}")


def reset_sessions():
    """세션은 DB(learning.chat_sessions)에 영속화된다(P8-A). 테스트는 db_engine 픽스처가 TRUNCATE 하므로 no-op."""
    pass