game/sim/customer.py

"""CustomerAgent state machine (R5) — full MVP behaviors.

States: SPAWNING → BROWSE ⇄ LINGER → CONSIDER → QUEUE → PAY → EXIT
Talk is interruptible from BROWSE/LINGER/CONSIDER and restores return_state.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional, Sequence

class AgentState(str, Enum):
    SPAWNING = "SPAWNING"
    BROWSE = "BROWSE"
    LINGER = "LINGER"
    CONSIDER = "CONSIDER"
    QUEUE = "QUEUE"
    PAY = "PAY"
    TALK = "TALK"
    EXIT = "EXIT"

# Legal forward transitions used by tick_state (talk is orthogonal).
LEGAL_TRANSITIONS = {
    AgentState.SPAWNING: {AgentState.BROWSE, AgentState.EXIT},
    AgentState.BROWSE: {AgentState.LINGER, AgentState.CONSIDER, AgentState.TALK, AgentState.EXIT},
    AgentState.LINGER: {AgentState.BROWSE, AgentState.CONSIDER, AgentState.TALK, AgentState.EXIT},
    AgentState.CONSIDER: {AgentState.QUEUE, AgentState.BROWSE, AgentState.TALK, AgentState.EXIT},
    AgentState.QUEUE: {AgentState.PAY, AgentState.EXIT},
    AgentState.PAY: {AgentState.EXIT},
    AgentState.TALK: {AgentState.BROWSE, AgentState.LINGER, AgentState.CONSIDER, AgentState.EXIT},
    AgentState.EXIT: set(),
}

@dataclass
class CustomerAgent:
    agent_id: str
    npc_id: str
    kind: str  # regular | passerby
    preferred_atmosphere: dict
    desired_tags: List[str]
    stage: str = "STRANGER"
    state: AgentState = AgentState.SPAWNING
    patience: float = 60.0
    consider_left: float = 0.0
    target_slot: Optional[str] = None
    target_product: Optional[str] = None
    consider_match: float = 0.0
    return_state: Optional[AgentState] = None
    tick_counter: int = 0
    browse_ticks: int = 0
    linger_ticks: int = 0
    path: List[str] = field(default_factory=list)
    path_index: int = 0
    last_bark: Optional[str] = None
    entered_attract: float = 0.0
    purchase_done: bool = False
    queue_wait: float = 0.0
    history: List[str] = field(default_factory=list)

    def start(self, patience: float) -> None:
        self.patience = float(patience)
        self._enter(AgentState.BROWSE)
        self.path = ["door", "browse_1", "browse_2", "counter"]
        self.path_index = 0
        self.browse_ticks = 0
        self.linger_ticks = 0
        self.purchase_done = False
        self.queue_wait = 0.0

    def _enter(self, new_state: AgentState) -> None:
        self.state = new_state
        self.history.append(new_state.value)

    def can_transition(self, new_state: AgentState) -> bool:
        return new_state in LEGAL_TRANSITIONS.get(self.state, set())

    def current_marker(self) -> str:
        if not self.path:
            return "door"
        return self.path[min(self.path_index, len(self.path) - 1)]

    def advance_path(self) -> None:
        if self.path_index < len(self.path) - 1:
            self.path_index += 1

    def begin_talk(self) -> None:
        if self.state in (AgentState.EXIT, AgentState.PAY, AgentState.QUEUE):
            return
        self.return_state = self.state
        self._enter(AgentState.TALK)

    def end_talk(self) -> None:
        if self.state != AgentState.TALK:
            return
        restore = self.return_state or AgentState.BROWSE
        self.return_state = None
        self._enter(restore)

    def drain_patience(self, delta: float) -> bool:
        if self.state in (AgentState.TALK, AgentState.PAY, AgentState.QUEUE, AgentState.CONSIDER):
            # CONSIDER still drains lightly; queue/pay/talk hold patience
            if self.state != AgentState.CONSIDER:
                return False
            self.patience -= delta * 0.25
        else:
            self.patience -= delta
        if self.patience <= 0:
            self._enter(AgentState.EXIT)
            return True
        return False

    def begin_consider(self, product_id: str, slot: str, match: float, consider_sec: float = 4.0) -> bool:
        """BROWSE/LINGER → CONSIDER when a product is interesting enough."""
        if self.state not in (AgentState.BROWSE, AgentState.LINGER):
            return False
        if match < 0.35:
            return False
        self.target_product = product_id
        self.target_slot = slot
        self.consider_match = float(match)
        self.consider_left = float(consider_sec)
        self._enter(AgentState.CONSIDER)
        return True

    def join_queue(self) -> bool:
        if self.state != AgentState.CONSIDER:
            return False
        self.queue_wait = 0.0
        self._enter(AgentState.QUEUE)
        return True

    def begin_pay(self) -> bool:
        if self.state != AgentState.QUEUE:
            return False
        self._enter(AgentState.PAY)
        return True

    def complete_pay(self) -> bool:
        if self.state != AgentState.PAY:
            return False
        self.purchase_done = True
        self._enter(AgentState.EXIT)
        return True

    def abandon(self) -> None:
        if self.state != AgentState.EXIT:
            self._enter(AgentState.EXIT)

    def tick_state(
        self,
        dt: float,
        *,
        display_products: Optional[Sequence[dict]] = None,
        queue_ready: bool = True,
        buy_threshold: float = 0.45,
    ) -> AgentState:
        """Advance the agent one sim step. Pure logic — no I/O.

        display_products items: {id, slot, match} match in [0,1].
        """
        self.tick_counter += 1
        if self.state == AgentState.EXIT:
            return self.state
        if self.state == AgentState.SPAWNING:
            self.start(self.patience if self.patience > 0 else 60.0)
            return self.state
        if self.state == AgentState.TALK:
            # talk freezes browse progression; patience held
            return self.state

        if self.drain_patience(dt):
            return self.state

        if self.state == AgentState.BROWSE:
            self.browse_ticks += 1
            if self.browse_ticks % 2 == 0:
                self.advance_path()
            # linger after enough browsing
            if self.browse_ticks >= 3 and self.kind == "regular":
                self._enter(AgentState.LINGER)
                return self.state
            # try consider from displays
            if display_products:
                best = max(display_products, key=lambda x: float(x.get("match", 0.0)))
                if float(best.get("match", 0.0)) >= buy_threshold:
                    self.begin_consider(
                        str(best.get("id") or best.get("product_id") or ""),
                        str(best.get("slot") or ""),
                        float(best.get("match", 0.0)),
                    )
            return self.state

        if self.state == AgentState.LINGER:
            self.linger_ticks += 1
            if display_products:
                best = max(display_products, key=lambda x: float(x.get("match", 0.0)))
                if float(best.get("match", 0.0)) >= buy_threshold * 0.9:
                    self.begin_consider(
                        str(best.get("id") or best.get("product_id") or ""),
                        str(best.get("slot") or ""),
                        float(best.get("match", 0.0)),
                    )
                    return self.state
            if self.linger_ticks >= 4:
                self._enter(AgentState.BROWSE)
                self.browse_ticks = 0
            return self.state

        if self.state == AgentState.CONSIDER:
            self.consider_left -= dt
            if self.consider_left <= 0:
                if self.consider_match >= buy_threshold:
                    self.join_queue()
                else:
                    self._enter(AgentState.BROWSE)
                    self.target_product = None
            return self.state

        if self.state == AgentState.QUEUE:
            self.queue_wait += dt
            if queue_ready and self.queue_wait >= 0.5:
                self.begin_pay()
            return self.state

        if self.state == AgentState.PAY:
            # one tick to settle purchase
            self.complete_pay()
            return self.state

        return self.state

    def to_dict(self) -> dict:
        return {
            "agent_id": self.agent_id,
            "npc_id": self.npc_id,
            "kind": self.kind,
            "state": self.state.value,
            "patience": self.patience,
            "stage": self.stage,
            "target_product": self.target_product,
            "entered_attract": self.entered_attract,
            "purchase_done": self.purchase_done,
            "history": list(self.history),
        }

    @classmethod
    def from_dict(cls, data: dict) -> "CustomerAgent":
        agent = cls(
            agent_id=str(data["agent_id"]),
            npc_id=str(data["npc_id"]),
            kind=str(data.get("kind") or "passerby"),
            preferred_atmosphere=dict(data.get("preferred_atmosphere") or {}),
            desired_tags=list(data.get("desired_tags") or []),
            stage=str(data.get("stage") or "STRANGER"),
            patience=float(data.get("patience") or 60.0),
            target_product=data.get("target_product"),
            entered_attract=float(data.get("entered_attract") or 0.0),
            purchase_done=bool(data.get("purchase_done") or False),
        )
        st = data.get("state")
        if st:
            agent.state = AgentState(st)
        agent.history = list(data.get("history") or [])
        return agent