autosource / protocols /ap2 /mandates.py
Shanmuk4622's picture
deploy replay dashboard
7513e0a verified
Raw
History Blame Contribute Delete
4.33 kB
"""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"