src/session.py

"""Per-connection state: identity, sliding memory, pending order handshake, lightweight facts."""

from __future__ import annotations

from collections import deque
from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Any, Literal

from pydantic import BaseModel, Field


class Message(BaseModel):
    role: Literal["user", "assistant", "system", "tool"]
    content: str
    tool_name: str | None = None


class CartItem(BaseModel):
    sku: str
    quantity: int = Field(ge=1)
    unit_price: str
    currency: str = Field(default="USD")


def _normalize_order_items(raw: list[Any]) -> tuple[tuple[str, int, str, str], ...]:
    """Stable key for comparing create_order payloads regardless of list ordering."""

    normalized: list[tuple[str, int, str, str]] = []
    for row in raw:
        if not isinstance(row, Mapping):
            continue
        sku = str(row.get("sku", "")).strip()
        try:
            qty = int(row["quantity"])
        except (KeyError, TypeError, ValueError):
            continue
        unit_price = str(row.get("unit_price", "")).strip()
        currency = str(row.get("currency", "USD") or "USD").strip().upper()
        if sku and qty > 0 and unit_price:
            normalized.append((sku, qty, unit_price, currency))
    normalized.sort(key=lambda t: t[0])
    return tuple(normalized)


@dataclass
class PendingOrder:
    """Snapshot of a proposed create_order; equality guards against argument drift post-prompt."""

    customer_id: str
    items_key: tuple[tuple[str, int, str, str], ...]
    confirmed: bool = False

    @classmethod
    def from_create_order_args(cls, *, session_customer_id: str, args: dict[str, Any]) -> PendingOrder:
        items = args.get("items")
        raw_list = items if isinstance(items, list) else []
        return cls(customer_id=session_customer_id, items_key=_normalize_order_items(raw_list))

    def matches(self, args: dict[str, Any]) -> bool:
        other = PendingOrder.from_create_order_args(session_customer_id=self.customer_id, args=args)
        return self.customer_id == other.customer_id and self.items_key == other.items_key


@dataclass
class Session:
    customer_id: str | None = None
    customer_name: str | None = None
    customer_email: str | None = None
    cart: list[CartItem] = field(default_factory=list)
    pending_confirmation: PendingOrder | None = None
    history: deque[Message] = field(default_factory=lambda: deque(maxlen=12))
    facts: dict[str, Any] = field(default_factory=dict)

    def add_message(self, msg: Message) -> None:
        self.history.append(msg)