"""AP2-shaped mandate objects (plan §7.4), modeled on the real three-mandate design. Intent (human pre-authorization) -> Cart (what the agent assembled, bound to the Intent) -> Payment (what would be charged; carries a hash of Intent+Cart). We model the SHAPE and the audit trail, not cryptographic VC signatures (stated simplification, plan §2.2). Payments are MOCK ONLY — funding_instrument is hard-wired to "MOCK" and no code path can reach a real payment API. """ from __future__ import annotations import hashlib import json import uuid from datetime import datetime, timedelta, timezone from typing import Literal, Union from pydantic import BaseModel, Field class IntentMandate(BaseModel): intent_id: str user: str sku: str qty: int = Field(gt=0) max_unit_price: float = Field(gt=0) max_total: float = Field(gt=0) allowed_vendors: Union[Literal["any"], list[str]] = "any" deliver_by: str created_at: str class CartMandate(BaseModel): cart_id: str intent_id: str vendor_id: str sku: str qty: int = Field(gt=0) unit_price: float = Field(gt=0) tax: float = Field(ge=0) shipping: float = Field(ge=0) total: float = Field(gt=0) class PaymentMandate(BaseModel): payment_id: str cart_id: str amount: float = Field(gt=0) funding_instrument: Literal["MOCK"] = "MOCK" # hard-wired: never a real rail intent_cart_hash: str status: Literal["pending", "authorized"] = "pending" def _hash(intent: IntentMandate, cart: CartMandate) -> str: blob = json.dumps({"intent": intent.model_dump(), "cart": cart.model_dump()}, sort_keys=True) return hashlib.sha256(blob.encode()).hexdigest() def build_mandate_trio(*, user: str, sku: str, qty: int, max_unit_price: float, vendor_id: str, unit_price: float, lead_time_days: int = 7, tax_rate: float = 0.0, shipping: float = 0.0, ) -> tuple[IntentMandate, CartMandate, PaymentMandate]: now = datetime.now(timezone.utc) intent = IntentMandate( intent_id=f"INT-{uuid.uuid4().hex[:10].upper()}", user=user, sku=sku, qty=qty, max_unit_price=max_unit_price, max_total=round(max_unit_price * qty, 2), allowed_vendors="any", deliver_by=(now + timedelta(days=lead_time_days + 3)).date().isoformat(), created_at=now.isoformat(), ) tax = round(unit_price * qty * tax_rate, 2) cart = CartMandate( cart_id=f"CART-{uuid.uuid4().hex[:10].upper()}", intent_id=intent.intent_id, vendor_id=vendor_id, sku=sku, qty=qty, unit_price=unit_price, tax=tax, shipping=shipping, total=round(unit_price * qty + tax + shipping, 2), ) payment = PaymentMandate( payment_id=f"PAY-{uuid.uuid4().hex[:10].upper()}", cart_id=cart.cart_id, amount=cart.total, intent_cart_hash=_hash(intent, cart), status="pending", ) return intent, cart, payment def verify_cart_against_intent(intent: IntentMandate, cart: CartMandate, payment: PaymentMandate) -> tuple[bool, str]: """Code-enforced gate (plan §8.5): a PO can only be written when all checks pass. Human 'Approve' = signing the Cart against the Intent; only then may the Payment mandate flip to 'authorized'. """ if cart.intent_id != intent.intent_id: return False, "cart is not bound to this intent" if payment.cart_id != cart.cart_id: return False, "payment is not bound to this cart" if payment.intent_cart_hash != _hash(intent, cart): return False, "intent+cart hash mismatch — mandate tampered" if cart.sku != intent.sku or cart.qty != intent.qty: return False, "cart sku/qty differs from intent" if cart.unit_price > intent.max_unit_price + 1e-9: return False, f"unit price {cart.unit_price} exceeds intent max {intent.max_unit_price}" if cart.total > intent.max_total + 1e-9: return False, f"cart total {cart.total} exceeds intent max_total {intent.max_total}" if intent.allowed_vendors != "any" and cart.vendor_id not in intent.allowed_vendors: return False, f"vendor {cart.vendor_id} not in allowed list" return True, "ok"