feat: implement 'requires_escalation' and fix AP2 verification

Why:
- To enable conformance testing of the checkout escalation flow according to spec.
- To fix AP2 signature verification logic and missing exceptions which caused tests to fail/pass falsely.

What:
- Updated 'complete_checkout' to check for 'simulation_trigger': 'escalation_required'.
- Added '/recover/{checkout_id}' mock endpoint for 'continue_url'.
- Fixed 'Ap2VerificationError' definition and usage.
- Restored '_verify_ap2_mandate' method.

Author: 0x8000-0000

rest/python/server/exceptions.py

@@ -76,3 +76,11 @@ class InvalidRequestError(UcpError):
   def __init__(self, message: str):
     """Initialize InvalidRequestError."""
     super().__init__(message, code="INVALID_REQUEST", status_code=400)
+
+
+class Ap2VerificationError(UcpError):
+  """Raised when AP2 mandate verification fails."""
+
+  def __init__(self, message: str, code: str = "mandate_invalid_signature"):
+    """Initialize Ap2VerificationError."""
+    super().__init__(message, code=code, status_code=400)

rest/python/server/server.py

@@ -52,6 +52,15 @@ async def ucp_exception_handler(request: Request, exc: UcpError):
   )
 
 
+@app.get("/recover/{checkout_id}")
+async def recover_checkout(checkout_id: str):
+  """Mock checkout recovery UI."""
+  return {
+    "message": f"Recovering checkout {checkout_id}",
+    "status": "ui_rendering",
+  }
+
+
 # Apply business logic implementation to generated routes
 routes.ucp_implementation.apply_implementation(
   generated_routes.ucp_routes.router

rest/python/server/services/checkout_service.py

@@ -40,6 +40,7 @@
 import config
 import db
 from enums import CheckoutStatus
+from exceptions import Ap2VerificationError
 from exceptions import CheckoutNotModifiableError
 from exceptions import IdempotencyConflictError
 from exceptions import InvalidRequestError
@@ -97,6 +98,8 @@
 from ucp_sdk.models.schemas.shopping.types.line_item_resp import (
   LineItemResponse,
 )
+from ucp_sdk.models.schemas.shopping.types.message import Message
+from ucp_sdk.models.schemas.shopping.types.message_error import MessageError
 from ucp_sdk.models.schemas.shopping.types.order_confirmation import (
   OrderConfirmation,
 )
@@ -650,6 +653,28 @@ async def complete_checkout(
     checkout = await self._get_and_validate_checkout(checkout_id)
     self._ensure_modifiable(checkout, "complete")
 
+    # Check for risk signal trigger
+    if risk_signals.get("simulation_trigger") == "escalation_required":
+      checkout.status = CheckoutStatus.REQUIRES_ESCALATION
+      checkout.continue_url = AnyUrl(f"{self.base_url}/recover/{checkout.id}")
+      msg = MessageError(
+        type="error",
+        code="requires_buyer_input",
+        content="Escalation triggered by risk signal",
+        severity="requires_buyer_input",
+      )
+      checkout.messages = [Message(root=msg)]
+
+      response_body = checkout.model_dump(mode="json", by_alias=True)
+      await db.save_checkout(
+        self.transactions_session,
+        checkout.id,
+        checkout.status,
+        response_body,
+      )
+      await self.transactions_session.commit()
+      return checkout
+
     # Process Payment
     await self._process_payment(payment)
 
@@ -1248,3 +1273,16 @@ async def _process_payment(self, payment: PaymentCreateRequest) -> None:
     else:
       # Unknown handler
       raise InvalidRequestError(f"Unsupported payment handler: {handler_id}")
+
+  def _verify_ap2_mandate(self, ap2: Ap2CompleteRequest) -> None:
+    """Verify the AP2 mandate.
+
+    In this sample implementation, we simulate verification failure if the
+    mandate contains a specific trigger string.
+    """
+    mandate_str = ap2.checkout_mandate.root
+    if "invalid_signature" in mandate_str:
+      raise Ap2VerificationError(
+        "Invalid AP2 mandate signature (mock)",
+        code="mandate_invalid_signature",
+      )