Add settlement confirmation polling for async facilitator jobs

python/x402/http/facilitator_client.py

@@ -6,8 +6,10 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
+import time
 from typing import TYPE_CHECKING, Any, TypeVar
 
 logger = logging.getLogger("x402.facilitator_client")
@@ -117,6 +119,83 @@ def _parse_async_settle_acceptance(
     )
 
 
+def _extract_job_id(response: Any) -> str | None:
+    """Pull the settlement job id out of a facilitator 202 acceptance body."""
+    try:
+        response_data = response.json()
+    except (json.JSONDecodeError, ValueError, TypeError):
+        return None
+    if not isinstance(response_data, dict):
+        return None
+    payment_job = response_data.get("paymentJob") or response_data.get("settlementJob")
+    if isinstance(payment_job, dict):
+        job_id = payment_job.get("jobId")
+        if isinstance(job_id, str) and job_id:
+            return job_id
+    return None
+
+
+def _settle_response_from_job_status(
+    status_body: Any,
+    requirements_dict: dict[str, Any],
+) -> SettleResponse | None:
+    """Map a ``GET /settle/:jobId`` body to a terminal SettleResponse.
+
+    The facilitator marks a settlement job "succeeded" once the worker finishes
+    even when the worker returned ``success: false`` (e.g. the on-chain settle
+    reverted because the authorization expired). We therefore inspect the
+    embedded ``result.success`` rather than trusting the job state alone.
+
+    Returns:
+        A terminal SettleResponse, or ``None`` if the job is still pending and
+        polling should continue.
+    """
+    if not isinstance(status_body, dict):
+        return None
+
+    status = status_body.get("status")
+    network = requirements_dict["network"]
+
+    if status == "succeeded":
+        result = status_body.get("result")
+        if isinstance(result, dict):
+            transaction = result.get("transaction") or status_body.get("txHash") or ""
+            if result.get("success"):
+                return SettleResponse(
+                    success=True,
+                    transaction=transaction,
+                    network=result.get("network") or network,
+                    payer=result.get("payer"),
+                    amount=result.get("amount") or requirements_dict.get("amount"),
+                )
+            return SettleResponse(
+                success=False,
+                transaction=transaction,
+                network=result.get("network") or network,
+                error_reason=result.get("errorReason") or "settlement_failed",
+                error_message=result.get("errorMessage"),
+                payer=result.get("payer"),
+            )
+        # Completed with no structured result — fail closed rather than assume success.
+        return SettleResponse(
+            success=False,
+            transaction=str(status_body.get("txHash") or ""),
+            network=network,
+            error_reason="settlement_result_missing",
+        )
+
+    if status == "failed":
+        return SettleResponse(
+            success=False,
+            transaction="",
+            network=network,
+            error_reason=status_body.get("error") or "settlement_failed",
+        )
+
+    # queued / processing / unknown → not terminal yet.
+    return None
+
+
 # ============================================================================
 # Async HTTP Facilitator Client (Default)
 # ============================================================================
@@ -353,13 +432,61 @@ async def _settle_http(
         )
 
         if response.status_code == 202:
-            return _parse_async_settle_acceptance(response, requirements_dict)
+            if not self._wait_for_settlement:
+                return _parse_async_settle_acceptance(response, requirements_dict)
+            job_id = _extract_job_id(response)
+            if job_id is None:
+                logger.warning(
+                    "SETTLE_HTTP: 202 acceptance had no job id; cannot confirm settlement"
+                )
+                return _parse_async_settle_acceptance(response, requirements_dict)
+            return await self._poll_settlement_job(job_id, requirements_dict)
 
         if response.status_code != 200:
             raise ValueError(f"Facilitator settle failed ({response.status_code}): {response.text}")
 
         return _parse_facilitator_response(response, SettleResponse, "settle")
 
+    async def _poll_settlement_job(
+        self,
+        job_id: str,
+        requirements_dict: dict[str, Any],
+    ) -> SettleResponse:
+        """Poll ``GET /settle/:jobId`` until the settlement is terminal (async)."""
+        client = self._get_async_client()
+        url = f"{self._url}/settle/{job_id}"
+        deadline = time.monotonic() + self._settlement_poll_timeout
+
+        while True:
+            try:
+                response = await client.get(url, headers=self._get_settle_headers())
+                if response.status_code == 200:
+                    result = _settle_response_from_job_status(response.json(), requirements_dict)
+                    if result is not None:
+                        return result
+                elif response.status_code != 404:
+                    logger.warning(
+                        "SETTLE_POLL: unexpected status=%d for job %s",
+                        response.status_code,
+                        job_id,
+                    )
+            except (json.JSONDecodeError, ValueError, TypeError) as exc:
+                logger.warning("SETTLE_POLL: bad status body for job %s: %s", job_id, exc)
+
+            if time.monotonic() >= deadline:
+                logger.error(
+                    "SETTLE_POLL: settlement job %s did not confirm within %.0fs",
+                    job_id,
+                    self._settlement_poll_timeout,
+                )
+                return SettleResponse(
+                    success=False,
+                    transaction=job_id,
+                    network=requirements_dict["network"],
+                    error_reason="settlement_timeout",
+                )
+            await asyncio.sleep(self._settlement_poll_interval)
+
     async def _settle_data_http(
         self,
         settlement_type: str,
@@ -612,7 +739,15 @@ def _settle_http(
         logger.info("SETTLE_HTTP: response status=%d", response.status_code)
 
         if response.status_code == 202:
-            return _parse_async_settle_acceptance(response, requirements_dict)
+            if not self._wait_for_settlement:
+                return _parse_async_settle_acceptance(response, requirements_dict)
+            job_id = _extract_job_id(response)
+            if job_id is None:
+                logger.warning(
+                    "SETTLE_HTTP: 202 acceptance had no job id; cannot confirm settlement"
+                )
+                return _parse_async_settle_acceptance(response, requirements_dict)
+            return self._poll_settlement_job(job_id, requirements_dict)
 
         if response.status_code != 200:
             logger.error(
@@ -624,6 +759,51 @@ def _settle_http(
 
         return _parse_facilitator_response(response, SettleResponse, "settle")
 
+    def _poll_settlement_job(
+        self,
+        job_id: str,
+        requirements_dict: dict[str, Any],
+    ) -> SettleResponse:
+        """Poll ``GET /settle/:jobId`` until the settlement is terminal.
+
+        Returns a failure SettleResponse (rather than optimistically reporting
+        success) if the job reverts or does not confirm within the timeout, so
+        the caller can keep the session for retry / re-challenge.
+        """
+        client = self._get_client()
+        url = f"{self._url}/settle/{job_id}"
+        deadline = time.monotonic() + self._settlement_poll_timeout
+
+        while True:
+            try:
+                response = client.get(url, headers=self._get_settle_headers())
+                if response.status_code == 200:
+                    result = _settle_response_from_job_status(response.json(), requirements_dict)
+                    if result is not None:
+                        return result
+                elif response.status_code != 404:
+                    logger.warning(
+                        "SETTLE_POLL: unexpected status=%d for job %s",
+                        response.status_code,
+                        job_id,
+                    )
+            except (json.JSONDecodeError, ValueError, TypeError) as exc:
+                logger.warning("SETTLE_POLL: bad status body for job %s: %s", job_id, exc)
+
+            if time.monotonic() >= deadline:
+                logger.error(
+                    "SETTLE_POLL: settlement job %s did not confirm within %.0fs",
+                    job_id,
+                    self._settlement_poll_timeout,
+                )
+                return SettleResponse(
+                    success=False,
+                    transaction=job_id,
+                    network=requirements_dict["network"],
+                    error_reason="settlement_timeout",
+                )
+            time.sleep(self._settlement_poll_interval)
+
     def _settle_data_http(
         self,
         settlement_type: str,

python/x402/http/facilitator_client_base.py

@@ -157,6 +157,15 @@ class FacilitatorConfig:
     http_client: Any = None  # Optional httpx.Client or httpx.AsyncClient
     auth_provider: AuthProvider | None = None
     identifier: str | None = None
+    # Async settlement (facilitator returns 202 + a job id) handling.
+    # When True, ``settle`` polls the facilitator's job-status endpoint until
+    # the settlement reaches a terminal on-chain state instead of optimistically
+    # reporting success the moment the job is enqueued. Callers that settle in a
+    # background context (e.g. the session reaper) should enable this so a
+    # revert/expiry is surfaced as a real failure rather than silently dropped.
+    wait_for_settlement: bool = False
+    settlement_poll_interval: float = 2.0
+    settlement_poll_timeout: float = 120.0
 
 
 # ============================================================================
@@ -180,6 +189,13 @@ def __init__(self, config: FacilitatorConfig | dict[str, Any] | None = None) ->
             self._identifier = self._url
             self._http_client = None
             self._owns_client = True
+            self._wait_for_settlement = bool(config.get("wait_for_settlement", False))
+            self._settlement_poll_interval = max(
+                0.0, float(config.get("settlement_poll_interval", 2.0))
+            )
+            self._settlement_poll_timeout = max(
+                0.0, float(config.get("settlement_poll_timeout", 120.0))
+            )
         else:
             config = config or FacilitatorConfig()
 
@@ -189,6 +205,9 @@ def __init__(self, config: FacilitatorConfig | dict[str, Any] | None = None) ->
             self._identifier = config.identifier or self._url
             self._http_client = config.http_client
             self._owns_client = config.http_client is None
+            self._wait_for_settlement = config.wait_for_settlement
+            self._settlement_poll_interval = max(0.0, float(config.settlement_poll_interval))
+            self._settlement_poll_timeout = max(0.0, float(config.settlement_poll_timeout))
 
     @property
     def url(self) -> str: