Spaces:
Sleeping
Sleeping
| """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" | |