Spaces:
Runtime error
Runtime error
File size: 2,629 Bytes
f7e2a40 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """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)
|