core/negotiation.py

"""Negotiation engine (plan §9) — the deterministic referee between Buyer and Vendor.

State machine: OPEN -> COUNTER (loop <= max_turns) -> {SETTLED | WALKED}.
The engine (not the LLMs) owns the ground truth:
  - Buyer's hidden max_unit_budget   (never spoken in messages)
  - Vendor's hidden cost_floor       (from the warehouse's books, not an MCP tool)

Hard validators, enforced in code regardless of what any model says:
  1. A Buyer 'accept' above budget is rejected (converted to counter/walk).
  2. A Vendor 'accept' below floor is rejected (converted to counter at floor).
  3. Non-monotonic moves are clamped and flagged, never crash.
Every claimed number is parsed and checked against the state object; the
transcript stores raw text AND the parsed price so the LLM-judge can catch
fabricated concessions.
"""
from __future__ import annotations

import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path

from core.config import settings, traces_dir
from core.events import EventBus


@dataclass
class Turn:
    actor: str            # "buyer" | "vendor"
    action: str           # "offer" | "accept" | "walk"
    price: float | None
    raw: str              # raw model text (message shown in the timeline)
    valid: bool = True
    flags: list[str] = field(default_factory=list)


@dataclass
class NegotiationState:
    """Compact context object (plan §10) — this is ALL the models ever see."""
    sku: str
    item_name: str
    qty: int
    round: int = 0
    buyer_offer: float | None = None    # buyer's current standing offer
    vendor_ask: float | None = None     # vendor's current standing ask
    best_seen: float | None = None      # lowest vendor ask so far

    def public(self) -> dict:
        return {
            "sku": self.sku, "item": self.item_name, "qty": self.qty,
            "round": self.round, "your_last_offer": None,  # filled per side
            "buyer_offer": self.buyer_offer, "vendor_ask": self.vendor_ask,
            "best_seen": self.best_seen,
        }


class NegotiationEngine:
    def __init__(self, buyer, vendor, *, budget: float, floor: float,
                 qty: int, sku: str, item_name: str, request_id: str,
                 bus: EventBus, max_turns: int | None = None):
        self.buyer = buyer
        self.vendor = vendor
        self.budget = budget          # hidden ceiling (ground truth)
        self.floor = floor            # hidden floor (ground truth)
        self.qty = qty
        self.request_id = request_id
        self.bus = bus
        self.max_turns = max_turns or settings()["negotiation"]["max_turns"]
        self.state = NegotiationState(sku=sku, item_name=item_name, qty=qty)
        self.turns: list[Turn] = []
        self.violations = {"budget": 0, "floor": 0, "monotonic": 0}
        self.vendor_actor = f"vendor:{vendor.vendor_id}"

    # ---------- validators (deterministic, outside the LLM) ----------

    def _validate_buyer(self, turn: Turn) -> Turn:
        if turn.action == "walk":
            # anti-premature-quit: with rounds left and the vendor within ~10% of
            # the ceiling, a rational buyer makes a final offer instead of walking
            ask = self.state.vendor_ask
            if (self.state.round < self.max_turns and ask is not None
                    and ask <= self.budget * 1.10):
                turn.action = "offer"
                turn.price = round(self.budget, 2)
                turn.flags.append("premature_walk_converted_to_final_offer")
                if not turn.raw or turn.raw == "(no message)":
                    turn.raw = "This is my final number — take it or we're done."
            return turn
        if turn.action == "accept":
            target = self.state.vendor_ask
            if target is None:
                turn.action, turn.valid = "offer", False
                turn.flags.append("accept_without_ask")
                turn.price = round(self.budget * 0.85, 2)
            elif target > self.budget:
                # HARD RULE 1: buyer may never accept above budget
                self.violations["budget"] += 1
                turn.valid = False
                turn.flags.append(f"BLOCKED_accept_over_budget({target}>{self.budget})")
                if self.state.round >= self.max_turns:
                    turn.action, turn.price = "walk", None
                else:
                    turn.action, turn.price = "offer", round(self.budget, 2)
            else:
                turn.price = target
        elif turn.action == "offer":
            if turn.price is None:
                turn.valid = False
                turn.flags.append("offer_without_price")
                turn.price = round(self.budget * 0.85, 2)
            if turn.price > self.budget:  # offer itself may never exceed ceiling
                turn.flags.append(f"clamped_offer_to_budget({turn.price}->{self.budget})")
                turn.price = round(self.budget, 2)
            prev = self.state.buyer_offer
            if prev is not None and turn.price < prev:  # buyer walking offers backwards
                self.violations["monotonic"] += 1
                turn.flags.append(f"non_monotonic_buyer({turn.price}<{prev})")
                turn.price = prev
        return turn

    def _validate_vendor(self, turn: Turn) -> Turn:
        if turn.action == "walk":
            # A public/free-tier model occasionally walks despite a profitable
            # standing offer. The referee preserves vendor autonomy below floor,
            # but will not allow an economically dominated walk above floor.
            target = self.state.buyer_offer
            if target is not None and target >= self.floor:
                turn.action = "accept"
                turn.price = target
                turn.flags.append("profitable_walk_converted_to_accept")
                if not turn.raw or turn.raw == "(no message)":
                    turn.raw = "That standing offer works for us. We have a deal."
            return turn
        if turn.action == "accept":
            target = self.state.buyer_offer
            if target is None:
                turn.action, turn.valid = "offer", False
                turn.flags.append("accept_without_offer")
                turn.price = round(max(self.floor * 1.15, self.floor + 0.05), 2)
            elif target < self.floor:
                # HARD RULE 2: vendor may never accept below floor
                self.violations["floor"] += 1
                turn.valid = False
                turn.flags.append(f"BLOCKED_accept_below_floor({target}<{self.floor})")
                turn.action, turn.price = "offer", round(self.floor, 2)
            else:
                turn.price = target
        elif turn.action == "offer":
            if turn.price is None:
                turn.valid = False
                turn.flags.append("offer_without_price")
                turn.price = round(max(self.floor * 1.15, self.floor + 0.05), 2)
            if turn.price < self.floor:  # ask may never fall below floor
                turn.flags.append(f"clamped_ask_to_floor({turn.price}->{self.floor})")
                turn.price = round(self.floor, 2)
            prev = self.state.vendor_ask
            if prev is not None and turn.price > prev:  # vendor raising the ask
                self.violations["monotonic"] += 1
                turn.flags.append(f"non_monotonic_vendor({turn.price}>{prev})")
                turn.price = prev
        return turn

    # ---------- main loop ----------

    def run(self) -> dict:
        bus, st = self.bus, self.state
        bus.state("buyer", status="negotiating", vendor=self.vendor.vendor_id,
                  sku=st.sku, round=0)

        outcome, agreed = "WALKED", None
        while st.round < self.max_turns:
            st.round += 1

            # --- buyer speaks ---
            turn = self._validate_buyer(self.buyer.act(st))
            self.turns.append(turn)
            bus.say("buyer", turn.raw, price=turn.price, action=turn.action,
                    round=st.round, vendor=self.vendor.vendor_id, flags=turn.flags)
            if turn.action == "walk":
                outcome = "WALKED"
                break
            if turn.action == "accept":
                outcome, agreed = "SETTLED", turn.price
                break
            st.buyer_offer = turn.price

            # early-terminate: floor provably above ceiling and vendor at floor
            if self.floor > self.budget and st.vendor_ask is not None \
                    and abs(st.vendor_ask - self.floor) < 1e-9:
                bus.state("buyer", status="walking",
                          reason="vendor floor above our ceiling")
                outcome = "WALKED"
                break

            # --- vendor speaks ---
            turn = self._validate_vendor(self.vendor.act(st))
            self.turns.append(turn)
            bus.say(self.vendor_actor, turn.raw, price=turn.price, action=turn.action,
                    round=st.round, flags=turn.flags)
            if turn.action == "walk":
                outcome = "WALKED"
                break
            if turn.action == "accept":
                outcome, agreed = "SETTLED", turn.price
                break
            st.vendor_ask = turn.price
            st.best_seen = turn.price if st.best_seen is None else min(st.best_seen, turn.price)

            # overlap check (plan §9 settlement condition): ask within budget AND
            # buyer's offer already >= ask -> deal closes at the ask
            if st.vendor_ask <= self.budget and st.buyer_offer >= st.vendor_ask:
                outcome, agreed = "SETTLED", st.vendor_ask
                break

            bus.state("buyer", round=st.round, buyer_offer=st.buyer_offer,
                      vendor_ask=st.vendor_ask, vendor=self.vendor.vendor_id)

        # final safety assertion — a settled price must sit inside [floor, budget]
        if outcome == "SETTLED":
            assert agreed is not None
            assert agreed <= self.budget + 1e-9, "IMPOSSIBLE: settled above budget"
            assert agreed >= self.floor - 1e-9, "IMPOSSIBLE: settled below floor"

        result = {
            "request_id": self.request_id,
            "sku": st.sku,
            "vendor_id": self.vendor.vendor_id,
            "outcome": outcome,
            "settled": outcome == "SETTLED",
            "agreed_unit_price": agreed,
            "rounds": st.round,
            "violations": self.violations,
        }
        transcript_ref = self._save_transcript(result)
        result["transcript_ref"] = transcript_ref
        bus.state("buyer", status="negotiation_done", **{k: v for k, v in result.items()
                                                         if k != "violations"})
        return result

    def _save_transcript(self, result: dict) -> str:
        neg_dir = traces_dir() / "negotiations"
        neg_dir.mkdir(exist_ok=True)
        path = neg_dir / f"{self.request_id}-{self.vendor.vendor_id}.json"
        path.write_text(json.dumps({
            "meta": {
                "request_id": self.request_id, "sku": self.state.sku,
                "vendor_id": self.vendor.vendor_id, "qty": self.qty,
                "budget": self.budget, "floor": self.floor,
                "ts": datetime.now(timezone.utc).isoformat(),
            },
            "turns": [{"actor": t.actor, "action": t.action, "price": t.price,
                       "raw": t.raw, "valid": t.valid, "flags": t.flags}
                      for t in self.turns],
            "result": result,
        }, indent=2, ensure_ascii=False), encoding="utf-8")
        return str(path.relative_to(Path(traces_dir()).parent))